TASK 8: Add Checkpointing, Replay, Failure Handling, And Lineage Support

Summary

Implement the durability and recovery layer for KDS CDC across all five framework stages. This task defines per-stage lifecycle completion/failure rules, durable upstream lineage, source-coordinate checkpoint persistence, replay entry points from raw/load/curated durable artifacts, and standardized diagnostics across record-level quarantine and batch-level failures. It excludes broad operational validation scenarios, which remain in Task 9.

Current Codebase Findings

• docs/features/ETL/kds-cdc/index.md requires five separate stage batches with explicit lifecycle, UpstreamBatchId chaining, and completion before checkpoint advancement.
• The same doc requires checkpointing based on Debezium/source coordinates and SQS ordering metadata, not wall-clock time alone.
• The same doc also requires replay from raw, tenant/entity slices, coordinate ranges where available, load-artifact retry, and curated retry.
• backend/Kds.Ai.Integrations/Etl/StageExecutionSupport.cs is the current shared stage helper for:
• checkpoint advancement after stage success,
• failed-package persistence,
• failing stage batches with diagnostics.
• Current StageExecutionSupport.AdvanceCheckpointAsync(...) persists CheckpointValue as the stage completion timestamp string (CompletedAt.UtcDateTime.ToString("O")), not source coordinates.
• backend/Kds.Ai.Data/Services/IngestionCheckpointService.cs builds deterministic stage checkpoint keys as:

tenant:{tenantId}:business-date:{yyyy-MM-dd}:stage:{pipelineStage}

• backend/Kds.Ai.Data/Models/Ingestion/UpsertIngestionCheckpointRequest.cs and backend/Kds.Ai.Data/Entities/IngestionCheckpoint.cs show the current checkpoint model is intentionally small:
• TenantId
• SourceType
• CheckpointKey
• CheckpointValue
• LastSuccessfulBatchId
• optional BusinessDate
• optional PipelineStage

There are no first-class columns today for Debezium connector identity, source table, coordinate range, SQS message group, SQS sequence range, or replay pointers.

• backend/Kds.Ai.Data/Entities/IngestionBatch.cs already carries stage lineage and durable pointers needed as a foundation:
• BatchId
• CompanyId
• TenantId
• TenantDatabase
• BusinessDate
• PipelineStage
• Status
• DetailsJson
• ErrorDetails
• ResetOfBatchId
• UpstreamBatchId
• IsCurrent
• RawS3Path
• LoadS3Path
• backend/Kds.Ai.Data/Services/IngestionBatchLifecycleService.cs already implements important lifecycle semantics:
• successful stage batch becomes current,
• successful rerun supersedes previous current and sets ResetOfBatchId,
• failed rerun does not replace previous current.
• backend/Kds.Ai.Data/Services/StageUpstreamBatchResolver.cs already resolves the current successful upstream stage batch or validates an explicitly requested upstream batch id. It enforces same tenant, same business date, requested pipeline stage, Succeeded, and IsCurrent.
• backend/Kds.Ai.Warehouse/Services/Storage/FailedPackageWriter.cs persists tenant-scoped failed package references plus diagnostics and rejects payloads containing unsafe secret-like tokens.
• backend/Kds.Ai.Warehouse/Models/Storage/FailedPackageWriteRequest.cs and backend/Kds.Ai.Integrations/Etl/StageBatchFailureRequest.cs show current batch-level failure diagnostics are tenant-scoped and stage-scoped. They assume a resolved tenant and a created stage batch already exist.
• backend/Kds.Ai.Integrations/Etl/StageDiagnosticsEnvelope.cs already provides a structured wrapper carrying Source, CompanyId, TenantId, BusinessDate, PipelineStage, BatchId, UpstreamBatchId, Status, StartedAt, CompletedAt, and typed details. This is a useful durable envelope for KDS CDC lineage metadata.
• backend/Kds.Ai.Warehouse/Abstractions/Normalization/IRawObjectReader.cs already gives a durable raw replay seam by S3 path.
• backend/tests/Kds.Ai.Data.Tests/Storage/S3EtlStorageLiveTests.cs already prove raw/load/curated artifacts can be reopened from storage after write.
• backend/Kds.Ai.Warehouse/Services/Synthetic/SyntheticKdsFeedService.cs is the clearest current checkpoint example, but it uses a simple numeric incremental checkpoint with one stage (load_to_staging) and does not model multi-stage lineage or source-coordinate range metadata.
• Grep across backend found no general KDS CDC replay orchestrator yet. Existing replay references are limited to Brink replay-related trigger enums and replayable endpoint definitions, not a durable CDC replay workflow.

Current repo constraints this task must respect:

1. Existing checkpoint storage is too small for full KDS CDC recovery semantics if used alone; KDS CDC must persist richer coordinate/order/replay facts either by extending checkpoint persistence, by using structured batch details plus checkpoint pointers, or by a deliberate additive metadata record.
2. Stage-level failed-package flow works only after tenant resolution and stage-batch creation. Record-level pre-tenant quarantine remains the earlier Task 4 seam and must not be collapsed into tenant-scoped failed packages.
3. Current upstream resolver and lifecycle service already establish valid stage lineage/current-batch semantics. Task 8 should build on those seams instead of inventing a second lineage model.

Dependencies

• ../../docs/features/ETL/kds-cdc/index.md
• ../../docs/features/ETL/kds-cdc/implementation-breakdown.md
• ../../docs/features/ETL/etl-source-adapter-framework.md
• ./TASK5.md
• ./TASK6.md
• ./TASK7.md

Source References

• ../../docs/features/ETL/kds-cdc/implementation-breakdown.md#task-checklist
• ../../docs/features/ETL/kds-cdc/index.md#batch-lifecycle-and-lineage
• ../../docs/features/ETL/kds-cdc/index.md#checkpointing-and-replay
• ../../docs/features/ETL/kds-cdc/index.md#quarantine-and-failed-packages
• ../../docs/features/ETL/kds-cdc/index.md#supported-corner-cases
• ../../docs/features/ETL/kds-cdc/index.md#selected-behavior-summary
• ../../docs/features/ETL/etl-source-adapter-framework.md

Scope From Breakdown

• implement per-stage batch lifecycle, upstream chaining, and explicit completion/failure behavior
• persist checkpoints using Debezium/source coordinates and SQS ordering metadata
• support replay from raw, load, curated, and coordinate-bounded recovery points
• standardize quarantine and failed-package diagnostics across record-level and batch-level failures

Files to Create

1. KDS CDC checkpoint and replay seam under backend/Kds.Ai.Integrations/KdsCdc/Replay/ or .../Checkpointing/:
2. IKdsCdcCheckpointCoordinator
3. KdsCdcCheckpointCoordinator
4. KdsCdcCheckpointPayload
5. KdsCdcReplayRequest
6. KdsCdcReplayLevel
7. IKdsCdcReplayCoordinator
8. KdsCdcReplayCoordinator

Responsibilities:

• compute durable success point per stage,
• translate stage success facts into checkpoint payloads,
• resolve replay starting points from raw/load/curated durable state,
• enforce stage-by-stage recovery order.

1. KDS CDC lineage and recovery metadata contracts under backend/Kds.Ai.Integrations/KdsCdc/Models/:
2. KdsCdcCheckpointRange
3. KdsCdcSqsOrderingRange
4. KdsCdcReplayPointer
5. KdsCdcFailureDiagnostic
6. KdsCdcQuarantineDiagnostic
7. KdsCdcStageLineageDetails

Responsibilities:

• standardize durable coordinate and ordering ranges,
• standardize replay pointers to raw/load/curated artifacts,
• standardize failure and quarantine diagnostics payloads.

1. Optional additive persistence seam if current IngestionCheckpoint row cannot carry required CDC recovery facts cleanly:
2. under backend/Kds.Ai.Data/Entities/, Repositories/, Configurations/, and EF migration files,
3. examples: KdsCdcCheckpointMetadata, IKdsCdcCheckpointMetadataRepository, EfKdsCdcCheckpointMetadataRepository.

Guidance:

• prefer minimal additive persistence,
• use it only if CheckpointValue + LastSuccessfulBatchId + structured batch details are not sufficient to safely query and replay CDC progress.

1. Optional replay entry seam under backend/Kds.Ai.ApiService/Jobs/KdsCdc/ or backend/Kds.Ai.Integrations/KdsCdc/:
2. IKdsCdcReplayLauncher
3. KdsCdcReplayLauncher

Responsibilities:

• start bounded replay from explicit raw batch id, load batch id, curated batch id, or coordinate range,
• route replay through existing stage jobs rather than bypassing framework stages.

Files to Modify

• KDS CDC stage jobs from Tasks 5, 6, and 7 under backend/Kds.Ai.Integrations/KdsCdc/ so each stage records the coordinate/order/replay facts needed by Task 8.
• backend/Kds.Ai.Integrations/Etl/StageExecutionSupport.cs because current checkpoint advancement writes completion timestamp, which is insufficient for KDS CDC source-coordinate replay semantics.
• Checkpoint persistence seams:
• backend/Kds.Ai.Data/Models/Ingestion/UpsertIngestionCheckpointRequest.cs
• backend/Kds.Ai.Data/Entities/IngestionCheckpoint.cs
• backend/Kds.Ai.Data/Services/IngestionCheckpointService.cs
• backend/Kds.Ai.Data/Repositories/IIngestionCheckpointRepository.cs
• repository implementation/configuration/migrations if additive fields or companion metadata table are required.

Guidance: do not overwrite KDS CDC coordinate facts with plain completion timestamps.

• Batch detail and lineage usage seams:
• backend/Kds.Ai.Data/Entities/IngestionBatch.cs
• backend/Kds.Ai.Data/Repositories/IIngestionBatchRepository.cs
• backend/Kds.Ai.Data/Services/StageUpstreamBatchResolver.cs
• only if additive lineage queries or replay selection methods are concretely needed.
• Failure diagnostics seams where standardization is needed:
• backend/Kds.Ai.Integrations/Etl/StageBatchFailureRequest.cs
• backend/Kds.Ai.Warehouse/Models/Storage/FailedPackageWriteRequest.cs
• backend/Kds.Ai.Warehouse/Services/Storage/FailedPackageWriter.cs
• KDS CDC quarantine seam from Task 4.

Guidance: keep record-level quarantine and batch-level failed package flows distinct while aligning their diagnostic shape.

• Tests under:
• backend/tests/Kds.Ai.Integrations.Tests/KdsCdc/
• backend/tests/Kds.Ai.Data.Tests/
• additive storage/replay tests if new orchestration seam is introduced.

Required Behavior

Per-Stage Lifecycle And Explicit Completion

• KDS CDC keeps one IngestionBatch per framework stage:
• extract_raw
• normalize_load
• load_to_staging
• staging_to_curated
• curated_to_s3
• Every stage must:
• call StartStageBatchAsync(...) before work,
• persist stage details and durable pointers,
• call CompleteStageBatchAsync(...) after success,
• call failure handling only on batch-level failure,
• advance checkpoint only after completion.
• If a stage fails, downstream stages for that tenant/entity slice do not run.
• Successful rerun semantics should continue to rely on current IsCurrent / ResetOfBatchId lifecycle behavior.

Upstream Batch Chaining And Lineage

• UpstreamBatchId chaining is mandatory across all KDS CDC stages.
• Lineage must remain durable enough to answer:
• which raw batch produced this load batch,
• which load batch produced this curated batch,
• which curated batch produced this curated export,
• which source-coordinate range and SQS ordering range a stage batch covered.
• Replay from any downstream stage must be able to follow lineage back to raw durable state.

Checkpoint Semantics

• Checkpointing must use source coordinates and SQS ordering metadata, not wall-clock time alone.
• Minimum checkpoint metadata required by KDS CDC is:
• Debezium connector/source identity,
• source database/schema/table,
• Debezium offset/source coordinates,
• SQS message group id,
• SQS sequence number or message id,
• global order range,
• last successful raw batch id,
• last successful load batch id,
• last successful curated batch id,
• last successful curated export batch id.
• Task 8 must choose one durable representation for those facts:
• extended checkpoint persistence,
• structured checkpoint payload in CheckpointValue plus durable batch details,
• or an additive metadata table keyed from the checkpoint row.

Business Date And Checkpoint Boundary

• BusinessDate = DateOnly.FromDateTime(DateTime.UtcNow) remains acceptable for stage identity.
• Real CDC progress must not depend on BusinessDate; it must depend on durable source-coordinate and ordering facts.
• Daily checkpoint-key rotation may stay if KDS CDC still preserves actual replay position outside wall-clock-only values.
• Coordinate replay must be able to locate relevant batches/checkpoints without the operator knowing the processing-date partition that originally carried the stage batch.
• If checkpoint keys remain business-date-scoped, add an index/query path through coordinate metadata, batch details, or additive checkpoint metadata so replay by coordinate can search across processing dates.

Replay Levels

• Replay must support all documented recovery levels:
• raw batch replay,
• tenant/entity replay,
• coordinate replay,
• load artifact retry,
• curated retry.
• Replay must use durable artifacts and stage details, not SQS retention, once messages were acknowledged.
• Preferred recovery behavior:
• raw replay reuses Task 5 raw artifacts,
• load retry reuses Task 6 load artifacts,
• curated retry reuses Task 7 durable curated stage state,
• coordinate replay resolves bounded raw batches or stage details that cover the requested coordinate range.
• coordinate replay works across processing-date boundaries and does not require caller-supplied BusinessDate unless intentionally narrowing the search.

Failure And Quarantine Standardization

• Record-level quarantine remains for:
• malformed Debezium envelope,
• missing primary key,
• missing or ambiguous tenant mapping,
• unsupported source table,
• invalid value preventing row mapping,
• duplicate message that cannot be reconciled safely.
• Batch-level failure remains for:
• raw S3 write failure,
• durable job creation failure,
• ordering gap or out-of-order condition,
• unsafe Parquet production failure,
• Redshift COPY failure,
• curated merge failure,
• curated export failure.
• Diagnostics should be standardized across both record-level quarantine and batch-level failed package flows so operators can correlate:
• failure reason,
• source payload or pointer,
• source table and primary key,
• Debezium operation and coordinates,
• SQS message id/group/sequence,
• resolved or unresolved company/tenant mapping,
• raw S3 path,
• load S3 path when available,
• pipeline stage,
• batch id,
• retry/replay guidance.

Stage-Specific Checkpoint Success Points

• extract_raw checkpoint advancement must happen only after raw S3 write, durable job metadata creation, and stage completion.
• normalize_load checkpoint advancement must happen only after load artifacts are durable and stage completion succeeds.
• load_to_staging checkpoint advancement must happen only after Redshift apply succeeds and stage completion succeeds.
• staging_to_curated checkpoint advancement must happen only after curated merge succeeds and stage completion succeeds.
• curated_to_s3 checkpoint advancement must happen only after curated export succeeds, cleanup behavior is satisfied where needed, and stage completion succeeds.

Redelivery, Duplicate, And Replay Safety

• If SQS redelivers after raw write or job creation retries, downstream replay and dedupe must rely on source coordinates plus durable batch metadata, not raw object uniqueness or SQS retention.
• Duplicate CDC messages must remain idempotent across replay and retry paths.
• Out-of-order or missing sequence conditions must preserve enough diagnostics for later bounded replay.

Proposed Implementation Shape

High-level lifecycle and checkpoint flow:

stage work succeeds
  -> complete stage batch
  -> compute KDS CDC checkpoint payload from durable stage facts
  -> persist checkpoint row + any additive metadata needed
  -> mark stage current success point for replay

stage work fails
  -> if record-local: quarantine record with standardized diagnostics
  -> if batch-level: write failed package with standardized diagnostics
  -> fail stage batch
  -> do not advance checkpoint

Recommended checkpoint coordinator shape:

public interface IKdsCdcCheckpointCoordinator
{
    Task AdvanceAsync(
        KdsCdcCheckpointPayload payload,
        CancellationToken ct = default);
}

Recommended replay coordinator shape:

public interface IKdsCdcReplayCoordinator
{
    Task ReplayAsync(
        KdsCdcReplayRequest request,
        CancellationToken ct = default);
}

Recommended replay request direction:

• explicit raw batch id replay
• explicit load batch id retry
• explicit curated batch id retry
• tenant/entity bounded replay
• coordinate range replay

Recommended durable payload direction:

• keep checkpoint row as lookup anchor,
• keep rich coordinate/order/replay details in structured payload or additive metadata,
• keep stage DetailsJson queryable and source-faithful,
• avoid encoding recovery semantics only in ephemeral logs.

Testing

• Unit tests for checkpoint payload construction:
• source coordinate range persisted,
• SQS ordering metadata persisted,
• last successful stage batch ids persisted,
• timestamp-only checkpoint regression prevented.
• Data tests for checkpoint persistence shape:
• current checkpoint row stores enough anchor data,
• additive metadata table persists/retrieves KDS CDC replay facts if introduced,
• daily stage key behavior remains deterministic.
• Lifecycle/lineage tests for:
• UpstreamBatchId chain across all five KDS CDC stages,
• successful rerun supersedes prior current,
• failed rerun does not replace current,
• downstream stage cannot resolve failed/non-current upstream batch.
• Failure diagnostics tests for:
• record-level quarantine payload shape,
• batch-level failed-package payload shape,
• shared diagnostic fields between both flows,
• unsafe token filtering still enforced.
• Replay tests for:
• raw batch replay reopens raw artifact and routes through downstream stages,
• load retry reuses load artifact without re-reading raw,
• curated retry reuses curated durable state without re-reading SQS,
• coordinate-bounded replay resolves correct batches or replay pointers,
• coordinate-bounded replay can find batches across processing-date checkpoint keys,
• replay does not depend on SQS retention after acknowledgement.
• Regression tests ensuring Brink and synthetic existing checkpoint behavior remain unchanged unless explicitly extended in a generic safe way.

Not required in this task:

• broad load/perf/operations scenario validation,
• production admin UI for replay,
• alerting/ops runbooks,
• new business transformation logic.

Acceptance Criteria

• KDS CDC stage lifecycle is explicitly defined and implemented for all five stages with completion before checkpoint advancement.
• UpstreamBatchId lineage is durable and supports replay tracing across all stages.
• KDS CDC checkpoints persist Debezium/source coordinates and SQS ordering metadata rather than only completion timestamps.
• Replay is supported from raw, load, curated, tenant/entity, and coordinate-bounded recovery points.
• Standardized diagnostics exist across record-level quarantine and batch-level failed packages while preserving their distinct boundaries.
• Failed stages do not advance checkpoints.
• Successful rerun/current-batch semantics remain correct.
• Replay after acknowledgement does not rely on SQS retention.
• Coordinate replay does not require knowing the original processing BusinessDate.
• Tests cover checkpoint persistence, lineage, replay levels, failure diagnostics, and rerun semantics.

Notes

• Use docs/features/ETL/kds-cdc/index.md#checkpointing-and-replay and #batch-lifecycle-and-lineage as source of truth.
• Current StageExecutionSupport.AdvanceCheckpointAsync(...) timestamp-based checkpoint behavior is not sufficient for KDS CDC and should be treated as a concrete gap this task closes.
• Build on existing IngestionBatchLifecycleService and StageUpstreamBatchResolver semantics instead of inventing a second lineage model.
• Keep pre-tenant quarantine from Task 4 separate from tenant-scoped failed-package handling.