Retrying Failed Items Without Duplicating Successful Work

Lesson 7 established a taxonomy for classifying failures and routing the retryable ones into a retry queue. This lesson addresses a problem that appears the moment you actually act on that queue: retrying is not automatically safe. If an item has side effects — writing a row to a database, sending a notification, appending to a report, incrementing a counter — retrying it naively risks doing that side effect twice, which is often worse than the original failure. Understanding why this happens, and how to prevent it, is the difference between a retry mechanism that repairs a job and one that quietly corrupts its output.

Why Retries Risk Duplication

The core problem is that a failure can occur after a side effect has already happened, but before your code learns that it succeeded. Consider this sequence for a single item:

1. Your code calls the model. The model call succeeds.
2. Your code writes the result to the database. The write itself succeeds on the server.
3. The network connection drops before the "success" acknowledgment reaches your code.
4. Your code sees a timeout, concludes the write failed, and marks the item for retry.
5. On retry, your code writes the result to the database again.

From your program's point of view, step 3 and step 4 look exactly like a genuine write failure — there's no way to distinguish "the write never happened" from "the write happened but the confirmation was lost" using only the error you observed. This is not a hypothetical edge case; it's an inherent property of any operation performed over an unreliable network, and it becomes a near-certainty at the scale of a bulk job with thousands of items. The fix is not to make retries less frequent — that only reduces the odds without eliminating the problem — but to make the operation being retried safe to perform more than once.

Idempotency: The Property That Makes Retries Safe

An operation is idempotent if performing it multiple times has the same effect as performing it once. UPDATE users SET status = 'processed' WHERE id = 42 is idempotent — running it five times leaves the row in exactly the same state as running it once. INSERT INTO results (record_id, output) VALUES (42, '...') is not idempotent on its own — running it twice creates two rows, unless the table has a uniqueness constraint that prevents it.

The goal of a safe retry mechanism is to make every operation your pipeline performs idempotent with respect to a stable identifier — which is exactly why Lesson 2 insisted on a stable record_id carried through the whole pipeline from the start. Idempotency isn't a property you bolt on to the retry logic; it's a property you design into the operations the retry logic calls.

Making Database Writes Idempotent

The most common and most reliable way to make a write idempotent is an upsert: insert a new row if none exists for this identifier, or update the existing one if it does, using a database-level uniqueness constraint to guarantee this atomically regardless of how many times it runs.

import sqlite3


def upsert_result(conn: sqlite3.Connection, record_id: str, output: str) -> None:
    conn.execute(
        """
        INSERT INTO results (record_id, output)
        VALUES (?, ?)
        ON CONFLICT(record_id) DO UPDATE SET output = excluded.output
        """,
        (record_id, output),
    )
    conn.commit()

ON CONFLICT(record_id) DO UPDATE requires record_id to have a UNIQUE constraint in the table's schema — that constraint is what makes this operation genuinely safe to run any number of times: the first call inserts the row, and every subsequent call with the same record_id simply overwrites the same row with the same (or corrected) value, never creating a duplicate. Without the uniqueness constraint, ON CONFLICT has nothing to detect a conflict against, and the statement degrades back into a plain insert that duplicates on retry.

Making Non-Database Side Effects Idempotent: Idempotency Keys

Not every side effect is a database write you control the schema for. Sending an email, calling a third-party webhook, or triggering a downstream system are all operations where you often cannot rely on a uniqueness constraint at the destination. The general-purpose solution is an idempotency key: a unique value (typically the record_id, or a value derived from it) that you record locally before performing the side effect, and check before performing it again.

class IdempotencyTracker:
    """Tracks which operations have already been performed, keyed by a
    stable id, so retries can skip work that already happened."""

    def __init__(self, conn: sqlite3.Connection):
        self._conn = conn
        self._conn.execute(
            """
            CREATE TABLE IF NOT EXISTS completed_operations (
                idempotency_key TEXT PRIMARY KEY,
                completed_at TEXT DEFAULT CURRENT_TIMESTAMP
            )
            """
        )
        self._conn.commit()

    def already_done(self, key: str) -> bool:
        row = self._conn.execute(
            "SELECT 1 FROM completed_operations WHERE idempotency_key = ?", (key,)
        ).fetchone()
        return row is not None

    def mark_done(self, key: str) -> None:
        self._conn.execute(
            "INSERT OR IGNORE INTO completed_operations (idempotency_key) VALUES (?)",
            (key,),
        )
        self._conn.commit()


async def send_notification_idempotently(tracker: IdempotencyTracker, record_id: str):
    key = f"notify:{record_id}"
    if tracker.already_done(key):
        print(f"skipping {record_id}: notification already sent")
        return
    await actually_send_notification(record_id)
    tracker.mark_done(key)


async def actually_send_notification(record_id: str) -> None:
    print(f"sending notification for {record_id}")

This pattern works for genuinely any side effect, not just notifications, because it doesn't depend on the destination system supporting uniqueness constraints at all — the safety lives entirely in your own tracking table. The one requirement it does have is that mark_done must be called after the side effect definitely succeeded, and that the check-then-act sequence (already_done then perform the effect then mark_done) needs to be structured so that a crash between "effect performed" and "marked done" is the only failure window left — and that window is exactly the same one described at the start of this lesson, just narrowed as far as it can go. It cannot be eliminated entirely without a distributed transaction spanning both systems, which is rarely available in practice; the goal is to shrink the unsafe window, not claim perfect safety.

Idempotency at the Pipeline Level: Skipping Already-Succeeded Records

The same idea applies one level up, to the pipeline's own retry loop. Before retrying any record, check its already-recorded status — if a record's status is already SUCCEEDED, retrying it is pure waste (and a real risk, if any part of processing has non-idempotent side effects that weren't fully guarded).

def build_retry_batch(all_records: list, previous_results: dict[str, ItemResult]) -> list:
    """Return only the records that genuinely need to be retried:
    excludes anything already marked successful."""
    retry_batch = []
    for record in all_records:
        prior = previous_results.get(record.record_id)
        if prior is not None and prior.success:
            continue  # already done — never re-submit
        retry_batch.append(record)
    return retry_batch

This function is the safety net that makes the rest of the retry mechanism forgiving of mistakes: even if something upstream mistakenly includes an already-succeeded record in a retry list, build_retry_batch filters it back out before any request is made. Building this kind of defensive check at the boundary between "decide what to retry" and "actually retry it" is cheap insurance against a class of bugs that are otherwise very easy to introduce when retry logic evolves over time.

Retrying from the Correct Stage, Not from Scratch

Recall from Lesson 7 that a DOWNSTREAM_WRITE failure preserves the model's output on the ItemResult even though the item overall failed. A correct retry uses that preserved output to skip re-calling the model entirely:

async def retry_item(client, result: ItemResult, record) -> ItemResult:
    if result.output is not None:
        # The model call already succeeded last time; only the write failed.
        try:
            await write_result_downstream(result.record_id, result.output)
            return ItemResult(record_id=result.record_id, success=True, output=result.output)
        except Exception as exc:
            return ItemResult(
                record_id=result.record_id, success=False, output=result.output,
                failure=classify_exception(exc, FailureStage.DOWNSTREAM_WRITE),
            )
    # No prior output — the model call itself needs to be retried from scratch.
    return await process_item_with_failure_tracking(client, record)

Checking result.output is not None first is what turns Lesson 7's decision to preserve partial output into an actual cost and correctness saving: an item that only failed at the write stage is repaired with zero additional model calls, and an item that failed at the model call stage correctly starts over from that stage.

Testing Idempotency Logic

Idempotency logic is a natural fit for dependency-injected tests using an in-memory SQLite database and no network access:

def test_idempotency_tracker_prevents_duplicate_action():
    conn = sqlite3.connect(":memory:")
    tracker = IdempotencyTracker(conn)

    assert tracker.already_done("notify:1") is False
    tracker.mark_done("notify:1")
    assert tracker.already_done("notify:1") is True
    print("PASS: idempotency tracker correctly remembers completed operations")


def test_build_retry_batch_excludes_succeeded_records():
    records = [PipelineRecord(record_id=str(i), source_data={}) for i in range(3)]
    previous = {
        "0": ItemResult(record_id="0", success=True, output="done"),
        "1": ItemResult(record_id="1", success=False),
    }
    retry_batch = build_retry_batch(records, previous)
    ids = {r.record_id for r in retry_batch}
    assert ids == {"1", "2"}  # "0" already succeeded, "2" was never attempted
    print("PASS: retry batch excludes already-succeeded records")


test_idempotency_tracker_prevents_duplicate_action()
test_build_retry_batch_excludes_succeeded_records()

Common Mistakes

• Assuming "the write failed" and "the write never happened" are the same thing. A timeout or dropped connection after a successful server-side write looks identical to your code as a genuine failure, which is exactly why idempotent writes (via upsert or an idempotency key) are necessary rather than optional.
• Relying on application-level checks alone without a database-level uniqueness constraint. Checking "does this row already exist?" in application code before inserting has a race condition under concurrency — two workers can both check, both see nothing, and both insert. A UNIQUE constraint enforced by the database is what actually guarantees no duplicates, regardless of timing.
• Re-running the entire item on any retry, including stages that already succeeded. This wastes cost on repeated model calls and, for any side effect not fully protected by idempotency tracking, increases the chance of a duplicate action.

Best Practices

• Design every side-effecting operation to be idempotent with respect to a stable record_id before writing retry logic, using upserts for database writes and an idempotency key table for anything else.
• Check whether a record already succeeded before including it in any retry batch, as a defensive filter independent of how the retry batch was constructed.
• Preserve partial results across failed stages so a retry can resume from the point of actual failure instead of repeating already-successful, and potentially costly, work.