Checkpointing and Replay for Batch Restarts

A batch job that has processed eight million streaming line items and crashes at row 7,999,001 presents a specific financial risk: restart from the beginning and every already-committed royalty accrual gets counted twice, or restart from an arbitrary guess and some rows never get counted at all. Durable checkpointing removes the guesswork by persisting exactly how far a batch got before it failed, so a restart can resume from the last committed offset instead of the last line the worker happened to be holding in memory. This pattern sits alongside the wider concerns covered under Async Batch Processing for High-Volume Streams and is what makes long-running royalty jobs safe to kill and restart without a manual reconciliation pass.

Checkpoint Storage Trade-offs

Where a checkpoint lives determines how quickly a restarted worker can find its place and how much drift is tolerable between the checkpoint write and the actual data commit. The comparison below covers the four storage patterns seen most often in royalty batch runners.

For any batch whose output feeds a royalty ledger directly, the relational-table pattern is the default: writing the checkpoint update in the same transaction as the ledger insert means there is no window in which the two can disagree.

Prerequisites & Assumptions

The code below assumes Python 3.11+, a PostgreSQL-backed ledger accessed through psycopg[binary]>=3.1, and batches that are already keyed by a stable, sortable offset — typically a monotonically increasing transaction_id or a composite of reporting_period and row sequence number. It assumes the deduplication approach from building idempotent ingestion pipelines in Python is already in place; checkpointing and idempotent upserts are complementary, not substitutes for each other — a checkpoint tells a worker where to resume, while idempotency keys protect against any overlap that resuming introduces.

Implementation

Step 1: Define a checkpoint schema tied to a stable offset

The checkpoint must reference something that will not shift between runs. Reporting period plus a monotonically increasing sequence number is more durable than an array index, since a source file can be re-delivered with rows reordered.

from dataclasses import dataclass
from datetime import datetime, timezone

@dataclass(frozen=True)
class BatchCheckpoint:
    batch_id: str
    reporting_period: str          # e.g. "2026-06"
    last_committed_offset: int     # monotonic sequence within the period
    row_count_committed: int
    status: str                    # "in_progress" | "complete" | "failed"
    updated_at: datetime

def new_checkpoint(batch_id: str, reporting_period: str) -> BatchCheckpoint:
    return BatchCheckpoint(
        batch_id=batch_id,
        reporting_period=reporting_period,
        last_committed_offset=0,
        row_count_committed=0,
        status="in_progress",
        updated_at=datetime.now(timezone.utc),
    )

Step 2: Commit the checkpoint in the same transaction as the ledger write

Advancing the checkpoint outside the ledger transaction is the most common source of double-counted royalties after a restart — a crash between the two writes leaves the checkpoint either ahead of or behind the actual committed rows.

import psycopg

def commit_chunk_with_checkpoint(
    conn: psycopg.Connection,
    batch_id: str,
    reporting_period: str,
    rows: list[dict],
    new_offset: int,
) -> None:
    with conn.transaction():
        with conn.cursor() as cur:
            cur.executemany(
                """
                INSERT INTO royalty_ledger
                    (idempotency_key, isrc, reporting_period, stream_count, royalty_amount)
                VALUES (%(idempotency_key)s, %(isrc)s, %(reporting_period)s,
                        %(stream_count)s, %(royalty_amount)s)
                ON CONFLICT (idempotency_key) DO NOTHING
                """,
                rows,
            )
            cur.execute(
                """
                INSERT INTO batch_checkpoint
                    (batch_id, reporting_period, last_committed_offset, row_count_committed, status)
                VALUES (%s, %s, %s, %s, 'in_progress')
                ON CONFLICT (batch_id) DO UPDATE SET
                    last_committed_offset = EXCLUDED.last_committed_offset,
                    row_count_committed = batch_checkpoint.row_count_committed + EXCLUDED.row_count_committed,
                    status = 'in_progress'
                """,
                (batch_id, reporting_period, new_offset, len(rows)),
            )

Both statements commit or roll back together, so last_committed_offset can never point past rows that were not actually written to royalty_ledger.

Step 3: Resume from the last committed offset on restart

On startup, a batch runner should treat any in_progress checkpoint as evidence of a prior crash and resume from its recorded offset rather than reprocessing the whole reporting period.

def resume_batch(conn: psycopg.Connection, batch_id: str, reporting_period: str) -> int:
    with conn.cursor() as cur:
        cur.execute(
            "SELECT last_committed_offset FROM batch_checkpoint WHERE batch_id = %s",
            (batch_id,),
        )
        row = cur.fetchone()
        return row[0] if row else 0

def replay_from_offset(source_rows: list[dict], start_offset: int) -> list[dict]:
    """Skip rows already committed; the idempotency key ON CONFLICT clause
    is the second line of defense if this offset comparison is ever off by one."""
    return [r for r in source_rows if r["sequence"] > start_offset]

Because the ledger insert also carries an idempotency key, a resume that accidentally replays a row already committed is a no-op rather than a duplicate accrual — the offset check and the ON CONFLICT guard are deliberately redundant.

Step 4: Mark the checkpoint complete only after a reconciliation pass

Before flipping status to complete, compare the sum of stream_count and royalty_amount committed under the checkpoint against the source file’s own declared totals, if the DSP provides them.

def finalize_batch(conn: psycopg.Connection, batch_id: str, expected_row_total: int) -> None:
    with conn.transaction():
        with conn.cursor() as cur:
            cur.execute(
                "SELECT row_count_committed FROM batch_checkpoint WHERE batch_id = %s",
                (batch_id,),
            )
            (committed,) = cur.fetchone()
            assert committed == expected_row_total, (
                f"batch {batch_id} committed {committed} rows, expected {expected_row_total}"
            )
            cur.execute(
                "UPDATE batch_checkpoint SET status = 'complete' WHERE batch_id = %s",
                (batch_id,),
            )

Verification & Validation

After any restart, run a split-sum assertion comparing total royalty amounts committed for the affected reporting_period against the source report’s declared total before releasing the batch for downstream payout calculation. Confirm the ISRC resolution rate for the resumed portion matches the rate observed before the crash — a drop suggests the replay skipped rows rather than resuming cleanly. Every checkpoint advance should also write an audit log entry recording batch_id, last_committed_offset, and the transaction that produced it, so a financial audit can reconstruct exactly which restart produced which committed rows.

Edge Cases & Gotchas

A checkpoint that advances past a row written to a staging table but not yet flushed to the ledger creates a false sense of progress; keep the checkpoint update inside the same transaction boundary as the actual ledger write, never after an async flush. Watch for reporting periods that straddle a daylight-saving transition when the offset is derived from a timestamp rather than a pure sequence number — a naive comparison can skip or reprocess rows around the transition. If a source file is redelivered with rows in a different order than the original run, an offset based on array position rather than a stable transaction_id will resume from the wrong point entirely. A corrupted or missing checkpoint row should never default silently to offset zero for a partially-committed period; require an explicit rebuild step that reconciles against the ledger’s own committed rows for that batch_id before resuming. Finally, territory-partitioned batches that shard work across workers need one checkpoint row per shard, not one per batch, or a crash in one territory’s worker will falsely appear to roll back progress in territories that finished cleanly.

Related

Checkpointing depends on the same idempotency guarantees built out in building idempotent ingestion pipelines in Python, and both patterns exist to keep restarts inside Async Batch Processing for High-Volume Streams safe for royalty-bearing data. Batches that exhaust their retry budget entirely before a checkpoint can be marked complete should route their remaining rows through the dead-letter queue patterns for failed records rather than blocking the checkpoint indefinitely.