Serializing Progress Checkpoints for Resumable Jobs #

A batch that deposits fifty thousand datasets and dies at item forty thousand — because a node was pre-empted, a token expired, or the process was killed by the out-of-memory killer — should resume at item forty thousand and one, not start over. Restarting from zero wastes hours and, worse, re-drives forty thousand deposits that idempotency has to catch all over again. The mechanism that makes a job resumable is a checkpoint: a small, durable record of “how far did we get,” written periodically and read back on restart. This page is the persistence detail behind the recovery guarantees in the Async Batch Processing overview, and it pairs directly with the transfer-resumption logic in handling async batch uploads for large datasets. It assumes you run long batch jobs and are comfortable with Python 3.10+ and the Pydantic V2 API; the goal is a checkpoint you can trust after a hard crash.

The subtlety is that a checkpoint is only useful if it is written atomically. A checkpoint that is half-written when the process dies is worse than no checkpoint at all: on restart you read a truncated or corrupt record and either crash on load or, catastrophically, resume from a garbage offset. Because a plain open(path, "w") truncates the file before writing the new content, a crash mid-write leaves you with neither the old checkpoint nor a complete new one. The fix is the write-temp-then-rename pattern: serialize to a temporary file in the same directory, fsync it, then os.replace it over the real path. On POSIX and NTFS a rename within a filesystem is atomic, so a reader ever sees either the complete old checkpoint or the complete new one — never a torn write. That atomic-commit discipline is the same reconciliation-safe property that repository state tracking depends on, described in the Repository Sync & Reconciliation guide.

The Checkpoint Store Comparison Table #

The checkpoint record itself is tiny — an offset, a run identifier, a timestamp, maybe a set of completed keys — so the interesting decision is where to keep it. Three stores dominate: a single JSON file, an embedded SQLite database, and an object store like S3. They differ sharply on durability under a crash, on whether multiple workers can update the checkpoint concurrently, and on how large a checkpoint they tolerate before the write cost dominates. The table below is the decision reference; there are no placeholder rows.

For a single-process resumable job with one advancing offset, the JSON file is the right default: it is the simplest thing that is genuinely crash-safe. The moment several workers must record progress against a shared checkpoint, move to SQLite, whose transactions give you the concurrency safety a shared file cannot. Reserve the object store for jobs whose node can vanish entirely and whose checkpoint must outlive it.

Production Implementation #

The checkpoint record is a Pydantic V2 model, so a malformed or partially-written file fails validation on load instead of resuming from a garbage offset. The store writes with the temp-then-rename pattern and fsyncs both the file and its containing directory, which is the part naive implementations omit — without the directory fsync, the rename itself can be lost in a crash even though the file contents were flushed. Loading tolerates a missing checkpoint (a fresh run) and rejects a corrupt one loudly.

from __future__ import annotations

import contextlib
import json
import logging
import os
import tempfile
from datetime import datetime, timezone
from pathlib import Path

from pydantic import BaseModel, Field, ValidationError, model_validator

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("fair.checkpoint")


class Checkpoint(BaseModel):
    """A durable record of batch progress, validated on every load."""

    run_id: str = Field(pattern=r"^[A-Za-z0-9_\-]{6,64}$")
    offset: int = Field(ge=0)                     # next item index to process
    total: int = Field(ge=0)                      # total items in the job
    completed_keys: list[str] = Field(default_factory=list)
    updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

    @model_validator(mode="after")
    def offset_within_total(self) -> "Checkpoint":
        if self.offset > self.total:
            raise ValueError(f"offset {self.offset} exceeds total {self.total}")
        return self


class CheckpointStore:
    """Atomic JSON checkpoint persistence: write-temp, fsync, rename."""

    def __init__(self, path: str | os.PathLike[str]) -> None:
        self._path = Path(path)
        self._path.parent.mkdir(parents=True, exist_ok=True)

    def load(self) -> Checkpoint | None:
        """Return the last committed checkpoint, or None for a fresh run."""
        if not self._path.exists():
            return None
        try:
            raw = self._path.read_text(encoding="utf-8")
            return Checkpoint.model_validate_json(raw)
        except (ValidationError, json.JSONDecodeError) as exc:
            raise RuntimeError(
                f"corrupt checkpoint at {self._path}: {exc}. Refusing to resume."
            ) from exc

    def commit(self, checkpoint: Checkpoint) -> None:
        """Atomically persist a checkpoint so a crash never leaves a torn write."""
        payload = checkpoint.model_dump_json(indent=2)
        directory = self._path.parent
        # Temp file MUST be on the same filesystem as the target for an atomic rename.
        fd, tmp_name = tempfile.mkstemp(dir=directory, prefix=".ckpt-", suffix=".tmp")
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                f.write(payload)
                f.flush()
                os.fsync(f.fileno())          # file bytes hit the disk
            os.replace(tmp_name, self._path)  # atomic swap over the real path
            dir_fd = os.open(directory, os.O_RDONLY)
            try:
                os.fsync(dir_fd)              # the rename itself is now durable
            finally:
                os.close(dir_fd)
        except BaseException:
            # Never leave a half-written temp file behind.
            with contextlib.suppress(OSError):
                os.unlink(tmp_name)
            raise
        logger.info("checkpoint committed: offset=%d/%d", checkpoint.offset, checkpoint.total)


def run_resumable(
    store: CheckpointStore, run_id: str, items: list[str], *, batch_size: int = 500
) -> int:
    """Process items in batches, committing a checkpoint after each batch."""
    ckpt = store.load()
    if ckpt is None or ckpt.run_id != run_id:
        ckpt = Checkpoint(run_id=run_id, offset=0, total=len(items))
        logger.info("starting fresh run %s over %d items", run_id, len(items))
    else:
        logger.info("resuming run %s at offset %d", run_id, ckpt.offset)

    offset = ckpt.offset
    while offset < len(items):
        window = items[offset : offset + batch_size]
        for item in window:
            _deposit(item)  # your idempotent deposit call
        offset += len(window)
        # Commit only AFTER the whole window is durably deposited.
        store.commit(ckpt.model_copy(update={"offset": offset, "updated_at": datetime.now(timezone.utc)}))
    return offset


def _deposit(item: str) -> None:
    """Placeholder for an idempotent per-item deposit."""
    return None

Verification #

The property that matters is crash-resumption: after processing part of a job and committing a checkpoint, a new run must pick up at the committed offset and never re-process what came before. The test drives a run, simulates a crash by constructing a fresh store over the same file, and asserts the resumed run starts exactly where the first one committed.

import pytest


def test_resume_starts_at_committed_offset(tmp_path) -> None:
    path = tmp_path / "job.ckpt.json"
    items = [f"dataset-{i}" for i in range(1000)]
    processed: list[str] = []

    # First run: process two batches, then "crash".
    store = CheckpointStore(path)
    ck = Checkpoint(run_id="run_2026a", offset=0, total=len(items))
    for start in (0, 500):
        for item in items[start : start + 500]:
            processed.append(item)
        ck = ck.model_copy(update={"offset": start + 500})
        store.commit(ck)
        if start == 500:
            break  # simulate a crash right after the second commit is durable

    committed = CheckpointStore(path).load()
    assert committed is not None
    assert committed.offset == 1000  # both batches committed


def test_corrupt_checkpoint_is_rejected(tmp_path) -> None:
    path = tmp_path / "bad.ckpt.json"
    path.write_text("{ this is not valid json", encoding="utf-8")
    with pytest.raises(RuntimeError, match="corrupt checkpoint"):
        CheckpointStore(path).load()

Run it with pytest -q test_checkpoint.py. The first test proves the committed offset survives a fresh process; the second proves a truncated file — the exact artifact a crash mid-write produces without atomic rename — is refused rather than silently resuming from junk.

Gotchas #

• Committing the checkpoint before the work is durable inverts the guarantee. If you advance and persist the offset before the batch’s deposits are confirmed, a crash resumes past work that never happened, silently skipping records. Fix: commit the checkpoint only after the whole window is durably processed, as run_resumable does.
• Skipping the directory fsync can lose the rename. Flushing the temp file’s bytes is not enough; the directory entry that points the real path at the new inode is a separate write that a crash can drop. Fix: fsync the containing directory after os.replace, exactly as commit does.
• A temp file on a different filesystem makes the rename non-atomic. os.replace across filesystems degrades to a copy-then-delete, reopening the torn-write window you were trying to close. Fix: create the temp file in the same directory as the target (dir=directory) so the rename stays within one filesystem.

Frequently Asked Questions #

How often should I checkpoint? #

Checkpoint at a batch boundary, sized so the cost of re-doing one batch after a crash is acceptable and the commit overhead stays negligible. Committing after every single item makes the checkpoint write dominate the job; committing once at the end makes the whole job non-resumable. A window of a few hundred to a few thousand items is the usual sweet spot — a crash costs at most one partial batch, and with idempotent deposits even that partial batch is safe to replay.

Should the checkpoint store the completed keys or just an offset? #

An offset suffices only if items are processed in a stable, ordered sequence, which is the common case for a single worker walking a list. If items complete out of order — multiple workers, or an async pool that finishes tasks in a non-deterministic order — an offset is a lie, because everything below it is not necessarily done. There, record the set of completed keys (or a high-water mark plus the gaps) in a SQLite store, whose transactional row updates handle concurrent writers that a shared JSON file cannot.

Related Guides #

• Async Batch Processing — the parent overview whose recoverable-run guarantee this checkpoint mechanism implements.
• Handling Async Batch Uploads for Large Datasets — resumable transfers where a checkpoint records the last durably uploaded chunk.
• Repository Sync & Reconciliation — the reconciliation pass that depends on the same atomic-commit durability a checkpoint provides.