Checkpoint and State Recovery › Recovering from Checkpoint Corruption

Recovering from Checkpoint Corruption

This page is the runbook for the worst restart a streaming reconciler can face: the newest checkpoint is damaged, and resuming from it would either crash the process or — far worse — restore a silently inconsistent state that produces confident wrong verdicts. It sits under the checkpoint and state recovery reference and assumes you already write hashed, versioned checkpoints. Recovery here is a deterministic procedure — detect, walk back, replay forward — not an improvisation, because a reconciler that guesses during recovery is a reconciler nobody can trust.

Problem Framing

A reconciler pod is killed mid-write during a node drain. On restart it loads checkpoint epoch 4,812, and the integrity hash does not match the stored digest: the file was torn between staging and promotion, or truncated by a full disk, or silently bit-rotted in object storage weeks ago and only now read. The temptation is to “fix it up” — trust the offsets, ignore the buffer — but a partially trusted checkpoint is the one thing recovery must never do. The safe procedure is to reject the damaged epoch outright, walk back to the most recent epoch whose hash verifies, and replay forward from its committed offset, relying on an idempotent sink so replay is a no-op for already-applied work.

Walking back to the last clean checkpoint epoch Four checkpoint epochs are shown left to right: N is corrupt, N minus one is corrupt, N minus two verifies clean, and N minus three is older clean. Recovery scans from the newest backward, rejecting corrupt epochs until it reaches the clean epoch N minus two, from which it replays forward into an idempotent sink. scan newest → oldest, stop at first clean epoch replay forward into idempotent sink epoch N corrupt epoch N−1 corrupt epoch N−2 clean ✓ epoch N−3 clean
Recovery scans epochs newest-first, rejecting each failed integrity check until the most recent clean epoch is found, then replays forward from its committed offset into an idempotent sink.

Implementation

The recovery routine scans epochs newest-first, verifies each hash, and returns the first clean one together with the forward replay range. It refuses to return anything if no clean epoch exists, forcing an explicit re-snapshot decision rather than a silent bad resume.

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

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


@dataclass(frozen=True)
class RecoveredState:
    epoch: int
    committed_offset: int
    low_watermark_ms: int


def _verify(path: str) -> Optional[dict]:
    """Return the checkpoint payload only if its stored hash matches recomputation."""
    try:
        with open(path, encoding="utf-8") as fh:
            record = json.load(fh)
        payload = json.dumps(record["data"], sort_keys=True, separators=(",", ":"))
        if hashlib.sha256(payload.encode("utf-8")).hexdigest() != record["hash"]:
            logger.error("integrity mismatch: %s", os.path.basename(path))
            return None
        return record["data"]
    except (OSError, ValueError, KeyError) as exc:
        logger.error("unreadable checkpoint %s: %s", os.path.basename(path), exc)
        return None


def recover(directory: str) -> RecoveredState:
    """Walk epochs newest-first; return the first clean one or raise for re-snapshot."""
    epochs: List[str] = sorted(
        (f for f in os.listdir(directory) if f.startswith("cp-")), reverse=True
    )
    for name in epochs:
        data = _verify(os.path.join(directory, name))
        if data is None:
            continue                                  # reject, keep walking back
        state = RecoveredState(
            epoch=data["epoch"],
            committed_offset=min(data["offsets"].values()),
            low_watermark_ms=data["low_watermark_ms"],
        )
        logger.info("recovered clean epoch=%d; replay from offset=%d",
                    state.epoch, state.committed_offset)
        return state
    logger.critical("no clean checkpoint in %s; full re-snapshot required", directory)
    raise RuntimeError("checkpoint history exhausted; re-snapshot required")

Key Implementation Notes

  • Verify on read, always — not only on write. Bit-rot happens after a checkpoint is written and validated. Recomputing the hash at read time is the only way to catch storage corruption of an old epoch, and it is what makes “walk back to the last clean one” possible.
  • Reject, never repair. A corrupt checkpoint is discarded whole. Partially trusting its offsets while discarding its buffer produces a torn state that reconciles wrong; the discipline that keeps recovery safe is treating each epoch as atomic.
  • Replay is safe only with an idempotent sink. Walking back means re-applying work between the clean epoch and the crash. That is correct exactly when the downstream apply deduplicates, per exactly-once reconciliation with idempotent sinks; without it, recovery double-counts.
  • Retain enough epochs to survive a bad run. Keep more than two epochs so a run of consecutive corrupt checkpoints — a bad disk, a partial outage — can still be walked past. The retention depth is a recovery-time-objective decision, and losing all epochs forces a full re-snapshot.

Verification

Assert that recovery skips corrupt epochs and lands on the newest clean one, and raises when none survive.

python
import tempfile

def _write(directory, epoch, offsets, wm, corrupt=False):
    data = {"epoch": epoch, "offsets": offsets, "low_watermark_ms": wm, "buffered": []}
    payload = json.dumps(data, sort_keys=True, separators=(",", ":"))
    digest = hashlib.sha256(payload.encode("utf-8")).hexdigest()
    if corrupt:
        digest = "0" * 64                              # simulate a torn/bit-rotted file
    with open(os.path.join(directory, f"cp-{epoch:012d}.json"), "w") as fh:
        json.dump({"hash": digest, "data": data}, fh, sort_keys=True, separators=(",", ":"))

with tempfile.TemporaryDirectory() as d:
    _write(d, 10, {"0": 500}, 4_000, corrupt=True)     # newest, corrupt
    _write(d, 9, {"0": 450}, 3_500)                    # clean
    state = recover(d)
    assert state.epoch == 9 and state.committed_offset == 450
    logger.info("checkpoint recovery verified")

Operational Considerations

Make recovery a drilled procedure: periodically kill a reconciler mid-write in staging and confirm it walks back and resumes without a double-count, so the path is exercised before a real node drain needs it. Alert on every integrity-mismatch event even when recovery succeeds, because a rising mismatch rate signals a failing disk or a non-atomic backend that must be fixed before it eats through the retained epochs. Store checkpoints on versioned, replicated storage so a single-host failure never destroys the history, keep the retention depth aligned to your recovery-time objective, and record every recovery — the epoch reached, the replay range, and the mismatch that triggered it — in append-only audit storage so a regulator can see exactly how a recovered run reconstructed its state.