Checkpointing and Resume for Interrupted Batch Runs

A high-volume ingestion run that dies halfway through must resume without re-writing the maintenance records it already committed — double-writing a certifying-staff entry corrupts the very audit trail the pipeline exists to protect. This page covers persisting per-batch offsets and idempotency keys so an interrupted run resumes exactly once, building on the batch-orchestration model from Async Batch Processing for High-Volume Logs. The design principle is simple: commit the record and advance the checkpoint as one logical unit, and make every write idempotent so a replay is a no-op.

Prerequisites & inputs

Checkpointing assumes an ordered, replayable source: a batch cursor you can rewind to a known offset, whether that is a file line number, a database sequence, or a message-queue offset. Each work item must expose a stable identity that survives a restart — for a maintenance record, that is a natural key such as the work-order ID plus the aircraft registration, for example an APU (ATA 49) oil-servicing entry on N88MH keyed as WO-5521:N88MH.

Two pieces of state make resume safe. The offset records how far the run has consumed, so a restart knows where to begin reading. The idempotency key — derived from the record’s natural identity — records what has already been committed, so even if the offset is slightly stale the commit itself refuses to duplicate. The offset optimizes where to restart; the idempotency key guarantees correctness if you restart in the wrong place. Relying on the offset alone is the classic bug: a crash between “record written” and “offset advanced” replays the last record on resume. The broader resilience patterns this depends on are set out in Designing Fault-Tolerant MRO Pipelines.

Checkpoint and resume

Checkpoint and resumeBatchcursorCheckpointstoreCrash /restartResumeexactly-once

Step-by-step implementation

  1. Derive a deterministic idempotency key. Hash the record’s natural identity so the same source record always yields the same key across runs.
  2. Guard the commit. Before writing, check the committed-key set; if the key is present, skip the write and treat it as success.
  3. Advance the offset only after the commit is durable. Persist the offset in the same checkpoint record as the committed keys, so a restart never sees an offset ahead of the data.
  4. Load and resume. On startup, read the checkpoint, seek the cursor to the saved offset, and let the idempotency guard absorb any record that was committed but whose offset update was lost to the crash.
from __future__ import annotations

import hashlib
import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path

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


@dataclass
class Checkpoint:
    """Durable resume state: how far we read, and what we already committed.

    Persisting these together supports the continuing-airworthiness record
    obligation in EASA Part-M M.A.305, under which each maintenance record must
    be retained and remain uniquely attributable to one aircraft.
    """

    offset: int = 0
    committed_keys: set[str] = field(default_factory=set)
    updated_at: str = ""

    def to_json(self) -> str:
        return json.dumps({
            "offset": self.offset,
            "committed_keys": sorted(self.committed_keys),
            "updated_at": self.updated_at,
        })

    @classmethod
    def load(cls, path: Path) -> "Checkpoint":
        if not path.exists():
            return cls()
        data = json.loads(path.read_text(encoding="utf-8"))
        return cls(
            offset=data["offset"],
            committed_keys=set(data["committed_keys"]),
            updated_at=data.get("updated_at", ""),
        )


def idempotency_key(work_order_id: str, registration: str) -> str:
    natural = f"{work_order_id}|{registration}".encode("utf-8")
    return hashlib.sha256(natural).hexdigest()


@dataclass
class ResumableBatchRunner:
    checkpoint_path: Path
    _ledger: dict[str, dict] = field(default_factory=dict)

    def _persist(self, cp: Checkpoint) -> None:
        cp.updated_at = datetime.now(timezone.utc).isoformat()
        tmp = self.checkpoint_path.with_suffix(".tmp")
        tmp.write_text(cp.to_json(), encoding="utf-8")
        tmp.replace(self.checkpoint_path)  # atomic swap: never a half-written checkpoint

    def run(self, records: list[dict]) -> Checkpoint:
        cp = Checkpoint.load(self.checkpoint_path)
        logger.info("resuming at offset=%d committed=%d", cp.offset, len(cp.committed_keys))

        for index in range(cp.offset, len(records)):
            record = records[index]
            key = idempotency_key(record["work_order_id"], record["aircraft_registration"])
            if key in cp.committed_keys:
                logger.info("skip duplicate key=%s offset=%d", key[:12], index)
            else:
                self._ledger[key] = record  # the actual exactly-once write
                cp.committed_keys.add(key)
                logger.info("commit key=%s offset=%d", key[:12], index)
            cp.offset = index + 1
            self._persist(cp)  # commit + offset advance persisted as one unit
        return cp

Worked example

Stage a batch of APU-related records for N88MH, including WO-5521:N88MH (an ATA 49 oil-quantity servicing) and WO-5522:N88MH (an ATA 79 oil-system filter change). Start the run; suppose the process is killed by the batch scheduler right after committing WO-5521 and writing its checkpoint, but before the next record. The checkpoint on disk now reads offset=1 with one committed key.

Restart the runner. Checkpoint.load reads offset=1, so the loop begins at index 1 and processes WO-5522 — WO-5521 is never re-read. Now imagine a nastier crash: the process died after the _ledger write for WO-5521 but before _persist completed, leaving offset=0. On resume the loop re-reads index 0, recomputes the same idempotency key, finds it already in committed_keys from the recovered checkpoint — or, if that too was lost, the ledger’s own key guard refuses the second write — and logs skip duplicate. Either way WO-5521:N88MH is written exactly once. The atomic tmp.replace() guarantees the checkpoint file is never observed half-written, so a crash mid-persist leaves the previous consistent checkpoint intact.

Edge cases & compliance gotchas

  • Offset ahead of data. If you advance the offset before the commit is durable, a crash in between skips a record permanently — a silent gap in the airworthiness record. Always persist the commit and the offset together, offset last within the same write.
  • Non-deterministic keys. Deriving the idempotency key from a wall-clock timestamp or a UUID generated at ingest time breaks resume: the replayed record gets a fresh key and duplicates. Key only on the record’s intrinsic identity.
  • Timezone drift in the natural key. If a record’s identity ever incorporates a date, normalize it to ISO 8601 UTC first; a local-time key on N88MH recorded either side of midnight would otherwise hash to two different keys for one event.
  • Backpressure interaction. Checkpoint granularity and in-flight concurrency are linked — a run using the bounded-queue approach in Backpressure and Rate Limiting in Asyncio Ingestion Pipelines must checkpoint only records whose downstream writes have actually acknowledged, not merely those dequeued.

Exactly-once resume is not achieved by careful offset bookkeeping alone; it is achieved by making the write itself idempotent so that imperfect bookkeeping is survivable. Persist the commit and offset atomically, key every record on its intrinsic identity, and an interrupted overnight run picks up precisely where it stopped — with no duplicated maintenance entry and no gap an auditor could ever find.

Related pages