Lesson 3: Durability

The problem: six hours in, and the machine reboots

Your pipeline now runs in constant memory. It also takes six hours on the full corpus. At the four hour mark, the machine restarts for an operating system update, and you have nothing: the output file is incomplete, you do not know how far it got, and the only option is to start again.

A job that cannot survive interruption is a job you cannot run on real data, because on a long enough run something always interrupts it.

Checkpointing and resumability

A checkpoint is a record of progress durable enough to survive the process dying. Resuming means reading the checkpoint and skipping work already done.

The simplest workable design has three parts: a stable identifier per work item, a durable record of completed identifiers, and a skip check at the start of processing.

def load_completed(checkpoint_path: Path) -> set[str]:
    """Read the ids already processed. Returns empty set on first run."""
    if not checkpoint_path.exists():
        return set()
    with checkpoint_path.open(encoding="utf-8") as f:
        return {line.strip() for line in f if line.strip()}


def process_corpus(folder: Path, output: Path, checkpoint: Path) -> RunReport:
    completed = load_completed(checkpoint)

    with output.open("a", encoding="utf-8") as out_file, \
         checkpoint.open("a", encoding="utf-8") as ckpt_file:

        for raw in read_documents(folder):
            if raw.id in completed:
                continue

            document = clean_document(raw)
            out_file.write(json.dumps(asdict(document), ensure_ascii=False) + "\n")
            out_file.flush()

            ckpt_file.write(raw.id + "\n")
            ckpt_file.flush()

Several details are doing real work here.

Both files open in append mode. A resumed run adds to the existing output rather than truncating it.

The output is written before the checkpoint. Order matters. If the process dies between the two writes, the record exists in the output but not in the checkpoint, so a resumed run will process it again. If you reversed the order, a crash would leave an item marked done that was never written, and the record would be lost forever. Prefer duplicating work over losing data, then make the write idempotent so duplication is harmless.

Both writes are flushed. Without flush(), data sits in a buffer that a hard kill discards. Flushing on every record costs performance, and for a job where records take hundreds of milliseconds each, that cost is irrelevant next to losing four hours of work. For very fast records, flush every N records instead and accept that a crash loses up to N.

The identifier must be stable across runs. Using the file path or a hash of the content works. Using a counter does not, because the file order might differ. Using a timestamp certainly does not.

[IMAGE PROMPT M4-3
Purpose: Show the checkpoint write ordering and what a resumed run does after an interruption.
Visual type: Sequence and state diagram with an interruption point.
Prompt: A clean educational diagram in two horizontal bands. The upper band, labelled "First run", shows a left-to-right sequence of five item boxes labelled "doc 1" through "doc 5". Beneath each of the first three, two small stacked markers are shown labelled "written to output" and "id in checkpoint". Beneath doc 4, only the first marker is shown, and the second is missing. A jagged interruption symbol sits over doc 4 labelled "process killed here". Doc 5 has no markers. The lower band, labelled "Resumed run", shows the same five item boxes, with docs 1 to 3 greyed and marked "skipped, id in checkpoint", doc 4 marked "reprocessed, write is idempotent", and doc 5 marked "processed normally". A note on the right of the upper band reads "output written before checkpoint, so a crash duplicates rather than loses".
Required elements: Five item boxes in both bands, paired markers under completed items, a single marker under the interrupted item, an interruption symbol, skip and reprocess labels in the lower band, the ordering note.
Style: Clean educational illustration, professional, uncluttered, high contrast, flat vector.
Layout: Two horizontal bands stacked vertically and aligned so the item boxes line up between them, note text to the right of the upper band.
Text labels: "First run", "Resumed run", "doc 1", "doc 2", "doc 3", "doc 4", "doc 5", "written to output", "id in checkpoint", "process killed here", "skipped, id in checkpoint", "reprocessed, write is idempotent", "processed normally", "output written before checkpoint, so a crash duplicates rather than loses".
Aspect ratio: 16:9
Accessibility: Use marker presence, greying, and explicit text labels rather than colour alone to show state.
Avoid: Code screenshots, decorative icons, tiny text, logos, watermarks.
Alt text: Two-band diagram showing a first run where three documents are fully checkpointed and a fourth is written but interrupted before checkpointing, and a resumed run that skips the first three, reprocesses the fourth idempotently, and continues.
END IMAGE PROMPT]

Checkpoint granularity. Per record is simplest and safest. Per batch is faster and loses up to one batch on a crash. For very large jobs, storing the checkpoint in a database rather than a file is worth doing, because a set of two million identifiers held in memory is roughly a few hundred megabytes, which is real but usually acceptable.

Graceful shutdown. Handling the signal a scheduler sends before killing a process lets you finish the current record cleanly:

import signal

shutting_down = False


def request_shutdown(signum: int, frame: object) -> None:
    global shutting_down
    shutting_down = True
    print("Shutdown requested, finishing current document")


signal.signal(signal.SIGTERM, request_shutdown)
signal.signal(signal.SIGINT, request_shutdown)

for raw in read_documents(folder):
    if shutting_down:
        break
    ...

Note that this handles SIGTERM and SIGINT, which are polite requests. SIGKILL cannot be caught, which is exactly why the checkpoint must be durable rather than relying on cleanup at exit.

Idempotent writes

Idempotent means an operation can be applied more than once without changing the result beyond the first application. Because your resume strategy deliberately reprocesses the interrupted item, writes must be idempotent or a resumed run produces duplicates.

Three approaches, in increasing order of robustness.

Deduplicate on read. Write freely, and have the consumer discard repeats by identifier. Simple, and it pushes the problem to everyone downstream.

Write to a keyed destination. If the output is a database or a key-value store, use the record identifier as the key and upsert. Writing the same record twice overwrites rather than appends, so the operation is naturally idempotent.

Write one file per record. Output to output/{record_id}.json rather than appending to a shared file. Reprocessing overwrites the file. This scales badly past a few hundred thousand files on most filesystems, but it is unbeatable for simplicity when the count is moderate.

For append-only JSONL, the honest position is that appending is not idempotent, so pair it with either deduplication on read or a compaction step that rewrites the file keeping the last record per identifier.

The atomic write pattern, which prevents a different failure. A process killed mid-write leaves a partially written file that looks complete:

import os


def write_atomic(path: Path, content: str) -> None:
    """Write via a temporary file and rename, so readers never see a partial file."""
    temp = path.with_suffix(path.suffix + ".tmp")
    temp.write_text(content, encoding="utf-8")
    os.replace(temp, path)          # atomic on the same filesystem

os.replace is atomic on POSIX systems and on Windows, so a reader either sees the old file or the complete new one, never a half-written one. Use this for anything a separate process might read while you are writing, such as a run report or a manifest.

Run reports and rejection reports

A batch job that prints nothing has one bit of output: it finished or it did not. That is not enough to operate it.

Module 2 established that rejected documents are records carrying a reason. A run report is the summary of what a whole run did.

from dataclasses import dataclass, field
from datetime import datetime, timezone


@dataclass(frozen=True)
class RunReport:
    run_id: str
    started_at: datetime
    finished_at: datetime
    input_folder: str
    output_path: str

    documents_seen: int
    documents_accepted: int
    documents_rejected: int
    documents_skipped_resumed: int

    rejections_by_reason: dict[str, int]
    encodings_used: dict[str, int]

    was_resumed: bool
    completed_normally: bool
    error: str | None = None

Every field earns its place by answering a question someone will ask.

documents_seen against the sum of accepted, rejected, and skipped must balance. If it does not, records were lost somewhere, and that is the first check to run.

rejections_by_reason tells you whether the run was normal. Two percent rejected as duplicates is expected. Forty percent rejected as not_readable means a scraper broke, and it is a far more useful alert than a job that simply took longer than usual.

encodings_used reveals corpus composition. If ten percent of documents needed the lossy fallback, you have a data quality problem worth investigating rather than a pipeline problem.

completed_normally distinguishes a job that finished from one that was interrupted, which changes what the output means.

Write the report atomically at the end, and write it even when the run fails:

try:
    report = run_pipeline(folder, output, checkpoint)
except Exception as exc:
    report = partial_report(error=str(exc), completed_normally=False)
    raise
finally:
    write_atomic(report_path, json.dumps(asdict(report), default=str, indent=2))

The finally block guarantees the report exists whichever way the run ended, and the raise preserves the failure for the caller. Note default=str in json.dumps, which handles the datetime fields that JSON cannot serialize natively.

The rejection report is the detail behind the counts, written as JSONL alongside the output:

write_jsonl(rejected_path, (asdict(doc) for doc in rejected))

Each rejected document carries its source path, reason, and provenance from Module 2. When someone asks why a particular recipe is missing from the corpus, that file answers it in one grep.