Persist ingestion checkpoints at source boundaries

An ingestion checkpoint marks the latest source position whose downstream work has completed. Advance it before that work completes and a crash can skip changes. Advance it only afterward and failures may cause replay, but not loss.

Keep a separate checkpoint for every consumer and independently ordered source boundary. For an event stream, that may be a partition offset. SQL Server Change Data Capture uses Log Sequence Numbers. A paged change API might provide a durable continuation token.

Do not collapse those positions into one global LastProcessedAt. Two records can share a timestamp. Clocks can differ. One partition can advance while another fails. Use a timestamp only when the source defines it as a complete, stable cursor.

Model the source boundary

Model the checkpoint key from the source’s ordering boundary:

public sealed record CheckpointKey(
    string Source,
    string Consumer,
    string Partition);

public sealed record IngestionCheckpoint(
    CheckpointKey Key,
    string Position);

public sealed record SourceBatch(
    IReadOnlyList<SourceChange> Changes,
    string CheckpointPosition);

Consumer separates independent projections of the same source. A search index and an audit archive can read the same changes but advance independently. Partition is the smallest stream whose order and progress can be tracked on its own. A source without partitions can use a fixed value such as default.

Keep resume semantics in the adapter

Keep Position opaque outside the source adapter. The adapter knows what the value contains and how to resume from it.

The adapter also owns the position’s resume semantics. A checkpoint might identify the last processed event, the upper bound of an inclusive CDC range, or the next page to request. The ingestion loop passes the value back unchanged. It does not parse or increment it.

Apply first, then advance the checkpoint

public async Task IngestAsync(
    CheckpointKey key,
    ISourceReader source,
    IIngestionTarget target,
    ICheckpointStore checkpoints,
    CancellationToken cancellationToken)
{
    IngestionCheckpoint? current = await checkpoints.GetAsync(
        key,
        cancellationToken);

    while (true)
    {
        SourceBatch? batch = await source.ReadAsync(
            current?.Position,
            cancellationToken);

        if (batch is null)
        {
            return;
        }

        await target.ApplyAsync(
            batch.Changes,
            cancellationToken);

        bool advanced = await checkpoints.TryAdvanceAsync(
            key,
            expectedPosition: current?.Position,
            newPosition: batch.CheckpointPosition,
            cancellationToken);

        if (!advanced)
        {
            throw new InvalidOperationException(
                $"Checkpoint no longer matches the expected position for {key}.");
        }

        current = new IngestionCheckpoint(
            key,
            batch.CheckpointPosition);
    }
}

Move the checkpoint forward only after all work covered by the batch has succeeded. If you save it before ApplyAsync completes, a crash can make the next run skip changes permanently.

Another failure window remains when the target and checkpoint store cannot share a transaction. The target may commit, then the checkpoint write may fail. The next run reads the batch again. Make target writes idempotent with stable document and chunk identifiers, upserts, or another deduplication rule. The safe default is replay, not loss.

Handle concurrency and expired positions

TryAdvanceAsync should use a conditional write. It advances only when the stored position still equals expectedPosition. This stops two workers from silently overwriting the same checkpoint, but it does not assign the work. Use the source client’s partition ownership mechanism, a lease, or another explicit claim when several workers can process the same boundary. If the conditional update fails, stop the attempt and reload the checkpoint instead of continuing from stale local state.

A stored checkpoint can also become unusable. Event Hubs discards events outside its configured retention period, and a SQL Server CDC LSN must remain inside the capture instance’s validity interval. Detect an expired position and enter an explicit recovery or reconciliation path. Never reset silently to the newest position because that hides the missing changes.

Do not build custom checkpoint storage when the source client already supplies the coordination model. Azure Event Hubs processors combine partition ownership with per-partition checkpoints. Update a checkpoint only after the downstream work for that event or batch has succeeded. Ownership handoffs can still deliver some events to both the old and new owner, so downstream idempotency remains necessary. Choose a checkpoint frequency that balances replay volume against storage writes.

Keep document identity separate from progress. A source checksum answers whether a known document changed. A checkpoint answers how far a consumer has processed an ordered source. Many ingestion pipelines need both.

Use source-boundary checkpoints when the source exposes ordered changes that can be resumed. For a small source that can be scanned cheaply and does not provide a reliable cursor, a full reconciliation pass with stable IDs and checksums is usually simpler and safer than manufacturing a checkpoint from timestamps.