# 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:

```text
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/`:
   - `IKdsCdcCheckpointCoordinator`
   - `KdsCdcCheckpointCoordinator`
   - `KdsCdcCheckpointPayload`
   - `KdsCdcReplayRequest`
   - `KdsCdcReplayLevel`
   - `IKdsCdcReplayCoordinator`
   - `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.

2. KDS CDC lineage and recovery metadata contracts under `backend/Kds.Ai.Integrations/KdsCdc/Models/`:
   - `KdsCdcCheckpointRange`
   - `KdsCdcSqsOrderingRange`
   - `KdsCdcReplayPointer`
   - `KdsCdcFailureDiagnostic`
   - `KdsCdcQuarantineDiagnostic`
   - `KdsCdcStageLineageDetails`

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

3. Optional additive persistence seam if current `IngestionCheckpoint` row cannot carry required CDC recovery facts cleanly:
   - under `backend/Kds.Ai.Data/Entities/`, `Repositories/`, `Configurations/`, and EF migration files,
   - 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.

4. Optional replay entry seam under `backend/Kds.Ai.ApiService/Jobs/KdsCdc/` or `backend/Kds.Ai.Integrations/KdsCdc/`:
   - `IKdsCdcReplayLauncher`
   - `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:

```text
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:

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

Recommended replay coordinator shape:

```csharp
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.