KDS CDC Performance And Scaling Notes

Scope

This note captures implementation-level conclusions from branch investigation of KDS CDC batching, export behavior, and future horizontal scale.

Current Throughput Findings

1. Two-minute delay was configured batching behavior

Observed behavior where 10 messages appeared to wait about 2 minutes before processing was caused by current AppHost settings, not a hidden lock or unexplained pause.

Relevant settings in backend/Kds.Ai.AppHost/AppHost.cs:

• kds-cdc-max-record-count = 1000
• kds-cdc-max-batch-age-seconds = 120
• kds-cdc-sqs-wait-time-seconds = 20

Relevant worker logic:

• KdsCdcBatchBuilder.ShouldClose(...) closes only on count, byte, age, or visibility safety limits.
• KdsCdcWorkerService flushes open batches only when ShouldClose(...) returns true.
• When queue poll returns no messages, worker still waits on next SQS long poll before re-checking batch age.

Consequence:

• small bursts such as 10 messages can sit in open batch until MaxBatchAgeSeconds is reached,
• actual flush time can drift slightly beyond 120 seconds because of SQS long polling.

This was reduced by tightening batching configuration.

2. Curated export was doing redundant work per slice

One incoming poll can be split into multiple tenant/entity slices. Before recent fix, each slice ran full curated_to_s3 export for all KDS CDC curated targets, even when only one entity changed.

Example impact:

• 3 slices in one batch,
• 4 curated export targets,
• 12 Redshift UNLOAD operations for very small input volume.

staging_to_curated already selected only relevant curated targets, but curated_to_s3 ignored that and exported full catalog.

Fix applied in code:

• backend/Kds.Ai.Integrations/KdsCdc/KdsCdcCuratedExportCatalog.cs
• backend/Kds.Ai.Integrations/KdsCdc/KdsCdcCuratedToS3Jobs.cs

Current behavior should export only targets that actually succeeded upstream in staging_to_curated.

3. Slice pipelines are still serialized

PipelineLaunchingKdsCdcMicroBatchSink.HandoffAsync(...) processes batch.TenantSlices sequentially.

Consequence:

• one worker poll can fan out into several slices,
• each slice waits for previous slice pipeline to finish,
• downstream latency grows linearly with slice count.

This is still a likely next bottleneck after redundant export removal.

Horizontal Scaling Conclusions

What already scales safely

SQS FIFO preserves ordering within one MessageGroupId. Current worker also prevents one open micro-batch from mixing messages across groups.

Relevant implementation points:

• KdsCdcBatchBuilder forces batch boundary when incoming message uses different MessageGroupId.
• mock contract enforces Sequence.StreamKey == MessageGroupId.
• production path uses transport.MessageGroupId as ordering stream key.

This means:

• records from different groups should not be mixed into one micro-batch,
• many workers can process many independent groups in parallel,
• FIFO queue remains source of per-group ordering.

What does not scale cleanly yet

Current duplicate and ordering protection are process-local, in-memory singletons.

Relevant classes:

• backend/Kds.Ai.Integrations/KdsCdc/KdsCdcMessageOrderingValidator.cs
• backend/Kds.Ai.Integrations/KdsCdc/KdsCdcDuplicateDetector.cs
• registered as singletons in backend/Kds.Ai.Integrations/DependencyInjectionExtensions.cs

Both maintain in-memory dictionaries of previously seen ordering or duplicate state.

Consequence in multi-instance deployment:

• one worker instance does not know what another instance has already seen,
• restart, rebalance, timeout recovery, or redelivery can move a message group to another instance,
• new instance loses local ordering and duplicate history,
• correctness checks become weaker or inconsistent across instances.

So horizontal scale is only partially safe today. Queue-level ordering remains, but application-level statefulness is not distributed.

Real throughput depends on shard count, not only worker count

Twenty worker instances help only if traffic is spread across many independent MessageGroupId values.

If one hot stream produces 300 messages per second under one message group, FIFO semantics effectively serialize that stream. Additional workers will not increase throughput much for that stream.

If traffic is spread across many groups, workers can scale horizontally by consuming different groups in parallel.

Current mock producer builds stream key as:

{sourceAccountId}:{storeGuid}

If production uses same effective shard key, scaling characteristics will be approximately:

• one store = one ordered lane,
• many stores = many parallel lanes.

If production traffic has high store_guid variation, that is good news for horizontal scale. It means 300 messages per second are likely to be distributed across many ordered lanes instead of one hot lane.

In that shape, queue fan-out across many worker instances is realistic, as long as per-store ordering remains the intended correctness boundary.

Production Readiness Recommendations

1. Decide and document production shard key

Most important scale decision is what MessageGroupId represents in production.

Recommended default:

• keep one ordered lane per store or equivalent store-scoped stream.

Reason:

• preserves natural local ordering boundary,
• avoids one global hot lane,
• allows horizontal scale across many stores.

2. Move ordering and duplicate state to durable/shared storage

Before large multi-instance rollout, move validator and duplicate memory out of process.

Possible backends:

• Redis,
• DynamoDB,
• Postgres,
• dedicated checkpoint table keyed by ordering stream.

Minimum requirement:

• store last accepted ordering token per stream,
• store duplicate detection keys for configured replay window,
• make ownership independent from single worker process lifetime.

Recommended first choice for current platform:

• use existing Redis resource as shared fast state for ordering and duplicate windows.

Why Redis fits well here:

• very fast read/write path for per-message checks,
• natural TTL support for duplicate windows,
• easy key-per-stream model,
• better operational fit than adding another new state service just for hot worker memory.

Suggested Redis state model:

• kds:cdc:order:{streamKey} -> last accepted ordering token for one stream,
• kds:cdc:dup:{streamKey}:{dedupeKey} -> duplicate marker with TTL,
• optional kds:cdc:lease:{streamKey} -> short lease if stream-level ownership coordination becomes necessary.

Redis should hold hot coordination state, not replace durable replay truth. S3 raw artifacts, ingestion batches, and durable checkpoints should remain system of record.

3. Keep downstream stages idempotent

Redelivery and replay remain normal events in distributed ingestion.

Required properties:

• staging load safe to retry,
• curated merge safe to retry,
• curated export safe to retry,
• durable replay path available without depending on original worker instance.

4. Revisit slice-level concurrency after correctness model is hardened

After distributed correctness is addressed, next throughput lever is bounded concurrency inside slice handoff/pipeline execution.

Suggested direction:

• allow parallel slice execution only across independent tenant/entity slices,
• preserve ordering within one stream or one tenant/entity lane where required,
• cap concurrency to avoid overloading Redshift or S3.

Changes Required Before Go-Live

This section describes what should change before KDS CDC is treated as production-ready for sustained multi-instance load.

1. Freeze MessageGroupId contract around store-level ordering

Production should explicitly define MessageGroupId as store-scoped stream identity, or another shard key with equally clear ordering semantics.

Expected outcome:

• all records for one store stay in one ordered FIFO lane,
• different stores can scale across many workers,
• throughput grows with number of active stores instead of being trapped behind one global sequence.

This should be documented as hard contract between Debezium/SQS publisher and KDS CDC worker.

2. Replace in-memory ordering and duplicate tracking with shared state

Current singletons are acceptable for one worker process, but not strong enough for resilient horizontal scale.

Before go-live, ordering and duplicate checks should survive:

• worker restart,
• worker replacement,
• visibility timeout recovery,
• group reassignment to another instance,
• temporary scale-out/scale-in events.

Required change:

• persist per-stream last accepted ordering token,
• persist duplicate window for accepted messages,
• make validator state readable and writable by any worker instance.

Without this, 20 workers may still process queue safely most of time, but correctness under failover and redelivery will remain too dependent on luck and timing.

3. Add explicit ownership and recovery rules for stream processing

Before go-live, document and implement what happens when one worker takes a message group and another later continues same group.

Need clear rules for:

• when ordering token advances,
• when duplicate key is committed,
• what happens if downstream handoff succeeds but ack is delayed,
• what state replay or retry reads after partial failure.

This is less about raw throughput and more about making multi-instance behavior deterministic.

If Redis is used, these rules should be implemented with atomic update behavior, for example transaction or Lua-script style compare-and-set semantics, so two workers cannot advance same stream state inconsistently.

Redis-Based Distributed State Design

This section describes recommended Redis design for replacing current in-memory worker state.

Goals

• preserve per-stream ordering across many worker instances,
• preserve duplicate detection across worker restarts and failover,
• keep per-message checks fast enough for sustained high throughput,
• avoid making Redis the long-term replay system of record.

Recommended Redis Keys

Assume stream identity is equivalent to MessageGroupId, and in production that should usually be store-scoped, for example:

{sourceAccountId}:{storeGuid}

Recommended keys:

kds:cdc:order:{streamKey}
kds:cdc:dup:{streamKey}:{dedupeKey}
kds:cdc:lease:{streamKey}
kds:cdc:batch:{batchId}

Meaning:

• kds:cdc:order:{streamKey}
• stores last accepted ordering token for stream
• value can include token plus metadata, for example last message id, worker id, updated timestamp
• kds:cdc:dup:{streamKey}:{dedupeKey}
• stores marker that message identity was already accepted
• set with TTL equal to duplicate detection window
• kds:cdc:lease:{streamKey}
• optional short-lived ownership key for active processing of one stream
• prevents overlapping processing when failover or redelivery creates temporary contention
• kds:cdc:batch:{batchId}
• optional transient marker connecting open batch to worker id and stream set
• useful for observability and recovery diagnostics, not required for minimal correctness

Key Payload Suggestions

kds:cdc:order:{streamKey} value can be JSON or Redis hash containing:

• lastOrderingToken
• lastMessageId
• lastSequenceNumber
• lastSourceCoordinates
• updatedAtUtc
• workerInstanceId

kds:cdc:dup:{streamKey}:{dedupeKey} can be simple string value such as:

• accepted message id,
• or compact status marker like 1

TTL should match or slightly exceed DuplicateCacheWindowMinutes.

Dedupe Key Strategy

Use same conceptual duplicate identity as current code, but move result to shared Redis state.

Suggested dedupe keys:

• mock:

mock|{entity}|{primaryKeyValue}|{orderingStreamKey}|{orderingStreamPosition}

• production:

prod|{sourceDatabase}|{sourceSchema}|{sourceTable}|{primaryKeyValue}|{sourceCoordinates}

This keeps behavior aligned with existing detector logic while making state cross-instance.

Ordering State Strategy

Use one Redis ordering record per stream.

Expected worker behavior per message:

1. derive streamKey from MessageGroupId
2. derive comparable orderingToken
3. read current kds:cdc:order:{streamKey}
4. atomically compare current token with incoming token
5. reject or quarantine duplicates/out-of-order messages
6. advance stored token only if incoming token is valid next accepted position

For production Debezium-style flow, exact definition of valid advancement depends on source coordinate semantics:

• if coordinates are strictly monotonic, require incoming > stored
• if source can redeliver same coordinate, treat equal token as duplicate instead of advancement
• if gaps matter, persist enough metadata to identify whether missing token indicates hard ordering break or acceptable sparse monotonic progression

Atomic Update Requirement

Do not implement ordering with plain GET then SET.

Required behavior:

• compare stored ordering token and incoming token atomically,
• detect duplicate marker atomically,
• write duplicate marker and advance ordering state as one logical decision.

Recommended implementation options:

• Redis transaction with optimistic concurrency,
• Redis Lua script,
• or equivalent server-side atomic primitive.

Best fit is usually one Lua script per message decision because it:

• reads current ordering state,
• checks duplicate marker,
• validates incoming token,
• writes new ordering state,
• writes duplicate TTL key,
• returns structured result such as accepted, safe_duplicate, suspicious_duplicate, out_of_order, or gap_detected.

Optional Stream Lease

SQS FIFO should already limit concurrent delivery per group, so lease is optional for baseline correctness.

Lease becomes useful when you want stronger protection during:

• visibility timeout expiry,
• slow downstream handoff,
• worker crash and reassignment,
• debugging overlapping ownership.

Lease model:

• worker sets kds:cdc:lease:{streamKey} with short TTL and unique worker token,
• worker renews while actively processing that stream,
• another worker can take over only after TTL expiry or explicit release,
• release must check worker token to avoid deleting another worker's lease.

This is coordination guardrail, not substitute for durable ordering state.

Recommended Worker Algorithm

For each received message:

1. parse and validate envelope
2. compute streamKey, orderingToken, and dedupeKey
3. optionally acquire or confirm stream lease
4. call atomic Redis decision function
5. if result is safe_duplicate, acknowledge without downstream work
6. if result is suspicious_duplicate, quarantine and acknowledge
7. if result is out_of_order or gap_detected, fail batch safely or quarantine according to contract
8. if result is accepted, continue batching and downstream flow

At batch flush:

1. keep current durable handoff rule: acknowledge only after raw write and durable internal state exist
2. do not roll back Redis ordering state just because later stage fails
3. rely on replay from durable raw artifacts rather than trying to rewind hot Redis memory

This last point matters. Redis should represent accepted intake progress, not entire downstream success history.

Why Redis State Should Advance Before Final Ack

Ordering and duplicate state belong to intake correctness boundary, not final warehouse completion boundary.

If message has been accepted into durable raw handoff path, later downstream retry should come from durable replay, not from pretending intake never saw message.

Therefore recommended split is:

• Redis state advances when worker accepts message into active ordered intake path,
• SQS message is deleted only after durable raw handoff succeeds,
• downstream stage failures are recovered from raw/replay lineage, not by clearing Redis ordering memory.

Failure And Recovery Behavior

Expected behavior with Redis-backed state:

• worker crashes before durable raw handoff:
• SQS redelivers later
• duplicate marker may already exist
• decision function should allow safe replay path based on durable-handoff status, or treat message as safe duplicate only if durable handoff can be proven
• worker crashes after durable raw handoff but before SQS delete:
• redelivery occurs
• duplicate marker exists
• message should be acknowledged without duplicating downstream state
• message group moves to another instance:
• new worker reads same Redis ordering state
• ordering validation remains consistent across instances

Because of first failure mode, implementation should be careful not to mark message permanently duplicated before there is enough durable evidence that replay path exists.

Practical rule:

• duplicate marker can be staged during intake,
• but message should be considered fully safe duplicate on redelivery only when durable handoff metadata confirms prior acceptance.

One way to model this is two-phase duplicate state:

• pending until raw write and durable batch creation succeed,
• committed after durable handoff,
• TTL on both states,
• redelivery logic checks status before deciding whether to ack immediately.

Recommended Scope Split

Use Redis for:

• hot ordering state,
• hot duplicate windows,
• optional short leases,
• short-lived operational markers.

Use durable DB/S3 state for:

• replay truth,
• batch lineage,
• raw artifact provenance,
• durable handoff confirmation,
• audit and recovery.

Go-Live Recommendation

For current architecture, Redis is best next step for distributed hot state.

Recommended rollout order:

1. introduce Redis-backed ordering state per stream
2. introduce Redis-backed duplicate markers with committed/pending semantics
3. add multi-instance failover tests
4. scale workers gradually
5. add optional lease only if overlap is observed or stronger ownership semantics are needed

Test Cases To Cover Before Go-Live

This section lists minimum test cases that should be covered before KDS CDC is treated as production-ready under multi-instance load.

A. Ordering Correctness

1. In-order messages for one stream

Scenario:

• one MessageGroupId
• messages arrive in correct order
• one worker instance

Expected result:

• all messages accepted
• ordering state advances monotonically
• one or more batches flush normally
• no quarantine, no duplicate classification

2. In-order messages for one stream across many workers

Scenario:

• one MessageGroupId
• worker A processes first part of stream
• worker B processes later part after restart, rebalance, or redelivery

Expected result:

• worker B reads shared ordering state correctly
• no false out-of-order detection caused by worker switch
• no duplicate downstream processing

3. Duplicate message with same ordering token and same payload

Scenario:

• same stream
• same message redelivered after prior acceptance

Expected result:

• classified as safe duplicate
• acknowledged without duplicate downstream work
• no ordering corruption

4. Duplicate identity with conflicting payload or metadata

Scenario:

• same duplicate key
• different body hash or conflicting sequence metadata

Expected result:

• classified as suspicious duplicate
• quarantined with diagnostics
• does not silently overwrite state

5. Out-of-order message for one stream

Scenario:

• stored ordering token is newer than incoming token

Expected result:

• message is rejected or batch fails according to contract
• durable diagnostics written
• later tokens are not advanced incorrectly

6. Ordering gap behavior

Scenario:

• stream appears to skip expected position or comparable token range

Expected result:

• behavior matches chosen production rule
• either accepted as sparse monotonic progress or failed as gap
• outcome is deterministic across worker instances

B. Redis State Semantics

7. Atomic ordering update under contention

Scenario:

• two workers attempt to process same stream nearly simultaneously

Expected result:

• only one valid ordering transition wins
• no lost update
• no split-brain ordering state

8. Duplicate marker pending -> committed transition

Scenario:

• message accepted into intake
• duplicate marker created as pending
• raw write and durable handoff complete
• marker transitions to committed

Expected result:

• redelivery before commit does not produce wrong final outcome
• redelivery after commit is safely acknowledged as duplicate

9. Redis restart or transient Redis failure

Scenario:

• Redis unavailable during intake or briefly restarted

Expected result:

• system fails safely
• no silent acceptance without correctness checks if those checks depend on Redis
• recovery path is clear and observable

10. Duplicate TTL expiry behavior

Scenario:

• duplicate window expires
• old message is redelivered later

Expected result:

• behavior matches designed replay contract
• message is either replayed from durable truth safely or explicitly treated as too old for hot dedupe
• no hidden corruption from expired cache state

C. SQS Redelivery And Failover

11. Worker crash after receive, before durable raw handoff

Scenario:

• worker receives message
• worker updates hot state
• worker crashes before raw S3 write and durable handoff complete

Expected result:

• redelivery handled safely
• message is not permanently lost because of hot-state update
• duplicate logic distinguishes unfinished versus committed prior work

12. Worker crash after durable raw handoff, before SQS delete

Scenario:

• raw write succeeded
• durable internal state exists
• worker crashes before message delete

Expected result:

• redelivery becomes safe duplicate or replay-aware recovery path
• no duplicate curated side effects

13. Visibility timeout expiry during slow downstream processing

Scenario:

• one worker is slow
• SQS redelivers same stream to another worker candidate

Expected result:

• no overlapping unsafe advancement of ordering state
• lease or atomic state rules prevent double-processing corruption

14. Scale-out during active traffic

Scenario:

• system starts with 1 worker
• scale to N workers while queue already active

Expected result:

• throughput improves for many-stream traffic
• no ordering regressions
• no burst of false duplicates or false quarantines

D. Batching And Slice Execution

15. Low-volume stream flushes by age

Scenario:

• small number of messages for one store
• count threshold not reached

Expected result:

• batch flushes within configured age target
• no unexpected multi-minute stall beyond accepted SQS poll drift

16. High-volume multi-store traffic

Scenario:

• many store_guid values active concurrently
• sustained high message rate

Expected result:

• workers process many streams in parallel
• no cross-store mixing
• throughput tracks shard count reasonably well

17. One hot store plus many quiet stores

Scenario:

• one stream dominates message volume
• many other streams produce low traffic

Expected result:

• hot store remains ordered
• quiet stores are not starved unnecessarily
• throughput bottleneck is visible as one hot lane, not random system instability

18. One poll resolves into many tenant/entity slices

Scenario:

• mixed entity traffic for one or more stores
• worker produces multiple resolved slices

Expected result:

• slice execution behavior matches intended concurrency model
• no conflicting overlap for same correctness lane
• stage latency and export behavior remain bounded

E. Downstream Idempotency

19. Replay same raw artifact twice

Scenario:

• same durable raw batch is replayed twice

Expected result:

• staging and curated state remain correct
• no duplicate final records

20. Retry load_to_staging from same load artifact

Scenario:

• Redshift load partially fails or is retried intentionally

Expected result:

• repeated load is safe
• staging state converges to expected result

21. Retry staging_to_curated for same slice

Scenario:

• curated merge retried after downstream or transient failure

Expected result:

• curated state converges correctly
• no duplicate semantic rows

22. Retry curated_to_s3 after partial export

Scenario:

• export fails mid-flight or after partial artifact creation

Expected result:

• retry produces clean final export
• no stale or mixed curated artifacts remain

23. Narrow export scope remains enforced

Scenario:

• upstream slice changes only one curated target

Expected result:

• only affected curated export target runs
• system does not regress to full-catalog export per slice

F. Observability And Operations

24. Hot shard detection

Scenario:

• one or few stores dominate traffic

Expected result:

• metrics/logs show lag concentrated in those streams
• operators can distinguish shard skew from worker-wide slowdown

25. Quarantine and duplicate metrics

Scenario:

• system receives malformed, duplicate, suspicious, and out-of-order traffic

Expected result:

• metrics and logs clearly separate each category
• operators can tell whether issue is source quality, replay, or ordering failure

26. Backlog growth and recovery

Scenario:

• incoming rate exceeds current processing rate for period of time

Expected result:

• queue depth and oldest age become visible
• after adding workers or traffic subsides, backlog drains without correctness regressions

G. Recommended Test Layers

These cases should not all live in one test style.

Recommended split:

• deterministic unit tests for Redis decision logic, duplicate rules, ordering rules, and key transitions
• integration tests for worker + Redis + SQS-like redelivery behavior
• stress or soak tests for many-store parallel traffic
• opt-in live readiness tests for real infra behavior under failover and scale changes

4. Make downstream idempotency an explicit release gate

Downstream stages already lean toward retry-safe behavior, but go-live should require proof, not assumption.

Need verified guarantees for:

• extract_raw duplicate durability handling,
• normalize_load replay from same raw artifact,
• load_to_staging safe retry of same load artifact,
• staging_to_curated safe re-run of same tenant/entity slice,
• curated_to_s3 safe re-export after partial or repeated failure.

Target property:

• same message or same slice can be replayed without corrupting curated state.

5. Add bounded parallelism in slice handoff

Current handoff runs slices sequentially. That leaves throughput on table once one poll resolves into many tenant/entity slices.

Before or soon after go-live, change handoff from strict serial execution to bounded concurrency across independent slices.

Guardrails:

• never break ordering inside one stream,
• do not let one store have overlapping conflicting slice execution,
• cap concurrency so Redshift COPY and UNLOAD remain stable.

This should likely be controlled by configuration, not hard-coded fan-out.

6. Tune batching for latency versus throughput using production traffic shape

Current batching values were suitable for investigation, not final production tuning.

Before go-live, set:

• MaxRecordCount,
• MaxBatchAgeSeconds,
• MaxBufferedBytes,
• SQS visibility timeout,
• worker instance count,

using realistic production traffic and target freshness.

Because you expect high store_guid variation, recommended approach is:

• smaller age window for fast flush,
• moderate count limit per store stream,
• more workers rather than one very large batch window.

That usually gives better end-to-end latency while still preserving throughput through shard parallelism.

7. Add scale and failover tests that match real rollout shape

Before go-live, test not only happy-path throughput, but also distributed failure behavior.

Minimum scenarios:

• many stores producing in parallel,
• one hot store plus many quiet stores,
• worker crash after receive but before ack,
• worker crash after raw handoff but before ack,
• message redelivery to different instance,
• scale from 1 worker to N workers during active traffic,
• duplicate and out-of-order injections under multi-instance load.

Success criteria should include both throughput and correctness:

• no mixed ordering within store stream,
• no silent data loss,
• no permanent duplicate amplification,
• bounded replay behavior.

8. Add operational visibility around stream lag and shard hot spots

Before go-live, operators should be able to see whether throughput issue is caused by:

• one hot store,
• too few worker instances,
• slow Redshift stages,
• repeated redelivery,
• ordering/duplicate rejections,
• backlog concentrated in few message groups.

Useful production metrics:

• queue depth,
• oldest message age,
• messages per second by worker,
• slice count per batch,
• stage duration by entity,
• per-group lag or at least per-store lag samples,
• duplicate/quarantine/failure counts.

9. Keep curated export scope narrow

Recent fix removed one major waste pattern. Before go-live, preserve this rule explicitly:

• export only curated targets affected by successful upstream slice work,
• do not regress to full-catalog export per slice.

At 300 messages per second, redundant Redshift UNLOAD work can become more expensive than ingest itself.

Bottom Line

• current flow can scale horizontally across many independent FIFO groups,
• current flow is not yet strong enough for carefree 20-instance deployment because duplicate and ordering memory are local to one process,
• records should not be mixed across groups, but failover and rebalancing can weaken correctness checks,
• best long-term path is: good shard key, shared ordering state, idempotent downstream stages, then bounded parallelism.