Designing Idempotent Retries for Partial Batch Failures

A digitization batch is not atomic. When a run of 4,000 preservation masters stops on file 2,731 — a scanner disconnect, a full disk, an unreadable master — the naive fix is to press “retry the batch.” That re-runs all 4,000 files, and unless every operation is idempotent it re-derives 2,730 already-good JPEG 2000 access copies, appends a second messageDigestCalculation event to objects that already have one, and forks the provenance record of every completed item. This page, part of the Error Handling & Retry Logic stage of the Automated Ingestion & Batch Scanning Workflows pipeline, shows how to design retries that touch only the files that actually failed — so a re-run converges the batch to “all done” without ever double-processing work that already succeeded. It is the counterpart to the reliability guarantees in Implementing Celery for Asynchronous Ingestion Tasks: late acknowledgment guarantees a failed message is redelivered, and this guarantees that redelivery is safe.

# Root-Cause Analysis of Non-Idempotent Retries

Re-running a partially failed batch corrupts records only when one of four concrete conditions holds. Each is fixable in the task design, not the infrastructure.

1. At-least-once redelivery treated as exactly-once. Every durable queue — Celery on RabbitMQ, RQ, SQS — guarantees at-least-once delivery, never exactly-once. A worker can finish the work, write the derivative, and die before its acknowledgment reaches the broker; the broker then redelivers the identical message. If the task assumes it runs once, that redelivery is a duplicate execution the code never planned for.
2. No per-file checkpoint. When progress is tracked only at batch granularity (“batch 8817: running”), the system cannot answer the one question a retry must ask: which files in this batch already completed? Without a per-file record, the only safe assumption is “none,” so everything reruns.
3. Side effects before commit. A task that writes the derivative to preservation storage, emits the PREMIS event, and then marks the file done has a fatal window: a crash after the side effect but before the mark leaves a completed side effect with no record of it. The retry sees an unmarked file and performs the side effect again — a duplicate derivative or a duplicate event.
4. Duplicate event emission. Even with correct ordering, an event log that appends unconditionally will hold two validation events for one object if the task runs twice. Auditors reading the provenance chain cannot tell a genuine re-validation from an accidental replay, and fixity-history math over the object breaks.

The unifying fix is an idempotency key that identifies a unit of work by what it is and what is being done to it, not by when it ran. For a preservation file the natural key is the pair (sha256_of_bytes, operation): the SHA-256 content digest that the pipeline already computes for fixity, joined with a stable operation name such as derivative:jp2 or event:messageDigestCalculation. Two invocations that share a key are, by definition, the same work — so the second must be skipped, not repeated. The state machine below shows how a batch of N files, split into done, failed, and pending, converges under retries that touch only the incomplete set.

Each retry pass selects only the failed ∪ pending files, skips any whose (sha256, operation) key is already recorded, and executes the rest — so repeated runs monotonically shrink the incomplete set until the batch converges.

# Step-by-Step Resolution: A Content-Keyed Idempotent Processor

The design has four moving parts: an idempotency key derived from content and operation, a checkpoint store holding the set of completed keys, a resume-from-offset selector that reprocesses only the incomplete files, and a strict record-then-acknowledge ordering so a crash can never lose the fact that work was done.

# The idempotency key and checkpoint store

The key must be stable across process restarts and independent of wall-clock time or batch position. Computing it from the file’s own bytes means a redelivered message, a resumed batch, and a manual re-run all derive the identical key for the same work. The checkpoint store is any transactional key-value surface — a database table with a unique constraint, a Redis set, or a WORM-friendly ledger; the only requirement is that “check membership” and “record membership” behave atomically per key.

from __future__ import annotations

import hashlib
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol

logger = logging.getLogger("archival.ingest.idempotent")


def compute_sha256(path: Path, *, chunk_size: int = 1 << 20) -> str:
    """Stream a file through SHA-256 so large masters never load fully into memory."""
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(chunk_size), b""):
            digest.update(block)
    return digest.hexdigest()


@dataclass(frozen=True)
class IdempotencyKey:
    """Identifies a unit of work by content and operation, not by time or position."""

    content_hash: str  # SHA-256 of the file bytes
    operation: str      # e.g. "derivative:jp2", "event:messageDigestCalculation"

    def token(self) -> str:
        return f"{self.operation}:{self.content_hash}"


class CheckpointStore(Protocol):
    """A transactional completed-set. Membership and recording are atomic per key."""

    def is_complete(self, key: IdempotencyKey) -> bool: ...

    def mark_complete(self, key: IdempotencyKey, *, batch_id: str) -> None: ...

# The per-file processor: check, skip or execute, then record

The processor is the heart of the design. Before doing any work it computes the key and consults the store. A hit means the work provably already completed, so it logs a skip decision and returns without a side effect. A miss means the work runs, and only after the side effect succeeds does the processor record the key — act, then record for operations whose side effect is externally observable and safe to repeat once, with the record closing the window on all subsequent retries.

from typing import Callable

Operation = Callable[[Path], dict[str, object]]


class IdempotentProcessor:
    """Runs a single operation on a file at most once, keyed by content and op name."""

    def __init__(self, store: CheckpointStore, *, batch_id: str) -> None:
        self._store = store
        self._batch_id = batch_id

    def process(self, path: Path, operation: str, run: Operation) -> dict[str, object]:
        content_hash = compute_sha256(path)
        key = IdempotencyKey(content_hash=content_hash, operation=operation)
        log_ctx = {
            "file": str(path),
            "operation": operation,
            "content_hash": content_hash,
            "key": key.token(),
            "batch_id": self._batch_id,
        }

        if self._store.is_complete(key):
            logger.info("idempotent_skip", extra={**log_ctx, "decision": "skip"})
            return {"status": "skipped", "key": key.token()}

        logger.info("idempotent_execute", extra={**log_ctx, "decision": "execute"})
        result = run(path)  # the side effect: derive, validate, or emit an event

        # Record ONLY after the side effect succeeds; a crash before this line
        # leaves the key absent, so the next retry re-runs — never double-records.
        self._store.mark_complete(key, batch_id=self._batch_id)
        logger.info("idempotent_recorded", extra={**log_ctx, "decision": "recorded"})
        return {"status": "processed", "key": key.token(), "result": result}

# Resume-from-offset across the batch

At the batch level, the retry driver never iterates the whole manifest blindly. It asks the store which files are already complete for the operation and processes only the remainder, so a re-run of the 4,000-file batch above dispatches roughly 1,270 files, not 4,000. Because process is itself idempotent, even a file that slips through the pre-filter (classified pending but actually completed by an in-flight worker) is caught by the per-file check — the two layers are defence in depth, not redundancy.

def retry_batch(
    processor: IdempotentProcessor,
    files: list[Path],
    operation: str,
    run: Operation,
) -> dict[str, int]:
    """Reprocess only the incomplete files; converge the batch without double-work."""
    counts = {"processed": 0, "skipped": 0, "failed": 0}
    for path in files:
        try:
            outcome = processor.process(path, operation, run)
            counts["skipped" if outcome["status"] == "skipped" else "processed"] += 1
        except (OSError, ValueError) as exc:
            counts["failed"] += 1
            logger.error(
                "idempotent_failed",
                extra={"file": str(path), "operation": operation, "error": str(exc)},
            )
    logger.info("batch_retry_summary", extra={"operation": operation, **counts})
    return counts

A batch converges when a full pass yields failed == 0; each pass records more keys, so the incomplete set shrinks monotonically and the loop terminates. Deterministic failures — a corrupt master that raises on every attempt — never enter the completed set and are escalated to a quarantine queue rather than retried forever, the same terminal-failure discipline described under Error Handling & Retry Logic.

# Ordering for event emission: record-then-act

Content derivation is safe to act then record because a duplicate derivative overwrites deterministically to the same bytes. Preservation event emission is not: appending is the side effect, so a second append is a permanent duplicate in the provenance chain. For events, invert the order to record-then-act — reserve the idempotency key inside the same transaction that writes the event, so the unique constraint rejects the second emission atomically. This is how deduplicated events must be written when emitting PREMIS events for preservation actions: the event’s identifier is derived from (content_hash, eventType) so a replayed task collides with the existing event instead of forking the object’s history.

# Validation and Verification

Prove idempotency behaviourally, not by inspecting configuration:

• Double-run assertion. Call retry_batch twice on a fully successful batch. The second pass must report processed == 0 and skipped == len(files). Any non-zero processed on the second pass is a missing or non-deterministic key.
• Crash-injection. Kill the worker between run(path) and mark_complete. On retry the file re-executes exactly once and the derivative’s SHA-256 is byte-identical to the first — confirming the act-then-record window is safe for derivations.
• Event-count invariant. After any number of retries, query the provenance store for each object: exactly one event per (content_hash, eventType). Re-derive the object’s checksum from storage and confirm it matches the recorded messageDigestCalculation event, the same fixity check applied before promotion.
• Key stability. Recompute IdempotencyKey for a file across two separate processes; the token() values must be identical. A key that varies by run defeats the entire design.

Propagate content_hash and batch_id through every structured log line so a single object’s skip/execute history is reconstructable across distributed workers — the same discipline the queue layer applies to redelivered tasks.

# Edge Cases and Gotchas

Three archival-specific traps recur. Multi-page TIFF masters must key on the bytes of the whole container, not per page, or a resumed batch re-derives the volume; conversely, if pages are ingested as separate objects, each needs its own content hash. In-place metadata rewrites (an ExifTool pass that mutates the master) change the file’s SHA-256, so the post-write bytes yield a different key than the pre-write bytes — pin the idempotency key to the original captured digest recorded at ingest, never to a re-read that a normalization step may have altered. Result-backend expiry must never be the checkpoint store: a Redis result backend that expires task results silently forgets which files completed, so a retry after expiry reprocesses everything. Keep completion keys in durable storage that outlives the queue, exactly as the audit trail is kept durable rather than trusting the broker’s result backend. Backoff between passes should follow the exponential backoff with jitter policy so a converging batch does not hammer the storage array on every retry.

For the hashing primitive, see the Python hashlib documentation; for the event model these keys deduplicate, see the Library of Congress PREMIS Data Dictionary.

# Frequently Asked Questions

# Why key on the content hash instead of the file path or a batch-relative index?

Because a path or index is a property of when and where the file was processed, not of the work itself. A resumed batch, a redelivered queue message, and a manual re-run may all assign different paths, offsets, or task IDs to the same master, so those keys fail to recognise repeat work. The SHA-256 of the bytes is intrinsic to the file: two invocations that share it are provably the same content, so joining it with the operation name yields a key that is stable across restarts, workers, and retries — the only kind of key that can safely gate a skip.

# Should I record completion before or after the side effect?

It depends on whether repeating the side effect once is harmless. For deterministic derivations — a JPEG 2000 access copy that overwrites to identical bytes — act first, then record, so a crash before recording simply re-derives the same output. For event emission and any append-only side effect, record and act inside one transaction (record-then-act) so the unique constraint rejects a duplicate atomically; appending twice would permanently fork the provenance chain, which no later cleanup can safely undo.

# How does this interact with a queue’s at-least-once delivery?

They are complementary layers. The queue guarantees a failed message is redelivered, so no work is lost; idempotent processing guarantees that redelivery re-runs the work at most once in effect, so no work is duplicated. Late acknowledgment without idempotency gives you duplicates; idempotency without reliable delivery gives you silent loss. You need both: the broker keeps the message until the task returns, and the content-keyed checkpoint ensures the returned task did its work exactly once.

# What stops a batch from retrying forever?

Two things. Each pass records completion keys, so the incomplete set shrinks monotonically and a batch of transient failures converges to failed == 0 within a few passes. Deterministic failures — a corrupt or unreadable master that raises on every attempt — never enter the completed set, so they are detected by a bounded retry count and routed to a quarantine queue for human review rather than looping. Convergence handles the transient case; escalation handles the terminal case.

# Related

• Error Handling & Retry Logic — the parent stage: retry budgets, quarantine routing, and terminal-failure escalation this design plugs into.
• Implementing Celery for Asynchronous Ingestion Tasks — the at-least-once delivery guarantee that makes safe redelivery both necessary and possible.
• Emitting PREMIS Events for Preservation Actions — how deduplicated, content-keyed events are written so a replayed task never forks provenance.