Streaming & CDC Reconciliation › Checkpoint and State Recovery

Checkpoint and State Recovery

Checkpoint and state recovery is what separates a streaming reconciler that survives a restart from one that replays a day of the log or, worse, resumes with a silently inconsistent view. A continuous reconciler is a long-lived process holding three pieces of state that must move together: the consumer offsets it has read, the low watermark it has sealed, and the in-flight comparison state it has buffered. This reference sits inside the streaming reconciliation pipelines stage and specializes it for durability, written for the data engineers, migration specialists, Python pipeline builders, and platform operations teams who must guarantee that a crash costs seconds of reprocessing, not a re-run, and never a wrong verdict.

The core discipline is atomicity. A checkpoint that commits the watermark but not the offsets, or the offsets but not the buffered state, leaves the reconciler in a torn condition that no restart can reason about: it will either re-seal events it already reconciled or skip events it never did. The only safe checkpoint is one where offsets, watermark, and buffer advance as a single all-or-nothing unit, and the only safe restart reconstructs from that unit deterministically — the same property that makes exactly-once reconciliation with idempotent sinks achievable without distributed transactions. When a checkpoint is nonetheless damaged, recovering from checkpoint corruption is a drilled procedure, not an improvisation.

Atomic checkpoint and deterministic restart of a streaming reconciler A running reconciler holding offsets, watermark, and in-flight buffer writes all three together through a two-phase stage-then-promote write into a durable checkpoint. On restart the checkpoint is validated by hash; a valid checkpoint restores and resumes from the committed offset, while an integrity failure falls back to the previous good epoch. restart valid integrity fail Running reconciler offsets watermark in-flight buffer Two-phase write stage → promote Durable checkpoint N + integrity hash Validate on restart verify hash Restore & resume from committed offset Fall back to checkpoint N−1
Offsets, watermark, and buffer are written as one atomic unit through a stage-then-promote write; on restart an integrity check either restores and resumes from the committed offset or degrades to the previous good checkpoint.

Architectural Boundaries

This workload consumes the reconciler’s live state — consumer offsets from the transport, the low watermark from watermark alignment, and the in-flight comparison buffer — and produces a durable, integrity-checked checkpoint plus a restart path that reconstructs that state deterministically. It owns when and how state is persisted and restored; it does not decide what is compared or render parity verdicts. The boundary matters because a reconciler that couples checkpoint logic into its comparison loop cannot be tested for crash-safety in isolation, and crash-safety is precisely the property that only fails in production.

Prerequisites

Step-by-Step Implementation

1. Define an atomic checkpoint record

Bundle every piece of resumable state into one immutable record with a content hash. Nothing outside this record may be needed to resume, or the checkpoint is not actually complete.

python
import hashlib
import json
import logging
import os
from dataclasses import dataclass, asdict, field
from typing import Dict, List, Optional

logging.basicConfig(
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", level=logging.INFO
)
logger = logging.getLogger("recon.streaming.checkpoint")


@dataclass(frozen=True)
class Checkpoint:
    offsets: Dict[str, int]          # partition -> committed offset
    low_watermark_ms: int
    buffered: List[dict] = field(default_factory=list)
    epoch: int = 0

    def content_hash(self) -> str:
        # Deterministic serialization: sorted keys, no whitespace drift.
        payload = json.dumps(asdict(self), sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(payload.encode("utf-8")).hexdigest()

2. Commit with a stage-then-promote write

Write the new checkpoint to a temporary path, fsync it, then atomically rename it over the live pointer. The previous checkpoint is retained so a failed promote leaves a recoverable predecessor.

python
def commit_checkpoint(cp: Checkpoint, directory: str) -> str:
    """Durably persist a checkpoint via stage-then-atomic-promote; keep the prior one."""
    os.makedirs(directory, exist_ok=True)
    digest = cp.content_hash()
    record = {"hash": digest, "data": asdict(cp)}
    tmp = os.path.join(directory, f".cp-{cp.epoch}.tmp")
    final = os.path.join(directory, f"cp-{cp.epoch:012d}.json")
    with open(tmp, "w", encoding="utf-8") as fh:
        json.dump(record, fh, sort_keys=True, separators=(",", ":"))
        fh.flush()
        os.fsync(fh.fileno())
    os.replace(tmp, final)            # atomic on POSIX; previous epoch untouched
    logger.info("checkpoint epoch=%d committed hash=%s", cp.epoch, digest[:12])
    return final

3. Validate and restore, degrading to the previous checkpoint

On restart, load the newest checkpoint, recompute its hash, and reject it if it does not match — then fall back to the previous good epoch and replay forward from its committed offset. Replay is safe precisely because the downstream apply is idempotent.

python
def load_latest_valid(directory: str) -> Optional[Checkpoint]:
    """Return the newest integrity-valid checkpoint, scanning backwards on failure."""
    files = sorted(
        (f for f in os.listdir(directory) if f.startswith("cp-")), reverse=True
    )
    for name in files:
        try:
            with open(os.path.join(directory, name), encoding="utf-8") as fh:
                record = json.load(fh)
            cp = Checkpoint(**record["data"])
            if cp.content_hash() != record["hash"]:
                logger.error("integrity mismatch in %s; trying previous epoch", name)
                continue
            logger.info("restored checkpoint epoch=%d watermark=%d", cp.epoch, cp.low_watermark_ms)
            return cp
        except (OSError, ValueError, KeyError, TypeError) as exc:
            logger.error("checkpoint %s unreadable: %s", name, exc)
    logger.critical("no valid checkpoint found in %s", directory)
    return None

Durability Strategy Trade-Off

The interval and backend are chosen from the reprocessing the business tolerates and the write budget available.

Strategy Recovery cost Write overhead Consistency guarantee Compliance / regulatory
Frequent local + atomic rename Seconds of replay High write rate Strong on one host; lost if the host dies Adequate with replicated storage; single-host risk must be documented.
Interval to object store Minutes of replay Moderate Survives host loss; bounded by interval Strong: versioned, immutable objects give an audit trail of every checkpoint.
Changelog / transaction log Near-zero replay Continuous append Strongest; exactly-once with idempotent sink Strongest: the log is itself an append-only audit record of state transitions.
No checkpoint (replay from start) Full re-run None None; correct but slow Rarely acceptable: unbounded replay window breaks recovery-time objectives.

Scaling and Performance

Checkpoint cost is dominated by buffered-state size, so the same tolerated-lateness value that bounds the aligner’s memory also bounds checkpoint write volume — keep it tight. Shard the reconciler by key range and checkpoint each shard independently so a restart touches only the failed shard’s state, and store checkpoints under a per-shard prefix so recovery is a targeted read. For very high throughput, switch from full-state snapshots to an incremental changelog that appends only state deltas between epochs and compacts periodically, trading a continuous write for near-zero replay. Whatever the backend, keep at least two epochs so the stage-then-promote fallback always has a predecessor, and verify the integrity hash on every read, because a checkpoint that is only validated at write time cannot catch storage bit-rot.

Failure Modes and Diagnostic Runbook

  • Torn checkpoint. Cause: a crash between staging and promote, or a non-atomic backend. Detection signal: integrity hash mismatch on the newest epoch. Remediation: the loader auto-degrades to epoch N−1; ensure the backend’s rename is truly atomic so this stays rare.
  • Diverged watermark and offsets. Cause: offsets and watermark were committed separately rather than as one record. Detection signal: on restart the watermark is ahead of the reprocessable offset range. Remediation: enforce the single-record checkpoint; never persist the two independently.
  • Checkpoint bit-rot. Cause: silent storage corruption of an old epoch. Detection signal: hash mismatch discovered on read during a deep recovery. Remediation: follow recovering from checkpoint corruption; keep enough epochs to reach a clean one.
  • Replay double-count. Cause: restart replayed events into a non-idempotent sink. Detection signal: discrepancy counts inflate after a recovery. Remediation: adopt exactly-once reconciliation with idempotent sinks so replay is a no-op.

Deep Dives