Idempotency Isn't Enough: Three Retry Bugs in Pipelines with Retrying

Problem: green tests, red pipeline

You built a pipeline with durable state. Every stage has a stable identity, the database enforces a uniqueness constraint, and the retry mechanism handles exceptions. Unit tests pass. Integration tests pass too. Only in production do you discover that retrying a non-deterministic stage that uses an AI model produces a different artifact from the first attempt — and downstream stages quietly start working with the wrong value. Another time, a parallel job reads stale state because it ran before the job it actually depended on. Or the process crashes between persisting an artifact and durably creating the next job, leaving the entire pipeline stuck.

These are not just theoretical edge cases. Bugs like these can remain hidden even with a completely green test suite because typical tests do not reproduce the exact execution interleavings that occur during retries. The underlying cause is usually the same: an individual database write is idempotent, but the process as a whole is not.

In this article, we will walk through three different failure modes. Each one breaks a different invariant and needs a slightly different regression test. At the end, we will turn these rules into a reusable set of invariants and test patterns that can catch these bugs before they reach production.

Act 1: the canonical artifact invariant

Failure 1: the local value trap

Consider a stage with a durable identity that calls a non-deterministic AI model. Its identifier is stable, for example pipeline_run_123/stage_456. On the first run, the model generates artifact A, which is persisted using INSERT ... ON CONFLICT DO NOTHING. The transaction commits, so A now exists durably in the database. A moment later, the process crashes.

When the stage runs again, it generates artifact B. It is different because the AI model is non-deterministic. The code tries to persist B, but the uniqueness constraint tied to the stage identity causes a conflict, so the new write is skipped. A is still in the database. At the level of the write itself, everything looks correct — the operation is idempotent.

The problem appears a moment later. The write operation does not return the persisted value because the INSERT was skipped, and the retry continues with the local value B. Downstream validation and stages therefore receive B instead of the canonical A that is actually stored in the database. The system becomes inconsistent: durable state says A, while the rest of the process continues based on B.

Invariant: the write must either successfully persist the proposed artifact or read the artifact that has already been assigned to the same stable identity. Every downstream decision must use the canonical value returned by that operation. In other words, execution should always continue from the state that actually won in durable storage.

The examples use Python-like pseudocode, but the invariants described here do not depend on any specific language or runtime.

# Bug: retry uses the local value after a failed insert
def save_artifact_buggy(stage_id: str, artifact: bytes) -> bytes:
    try:
        db.execute(
            "INSERT INTO artifacts (stage_id, artifact) VALUES (%s, %s) ON CONFLICT DO NOTHING",
            (stage_id, artifact)
        )
        # Bug: no return value; caller continues with the local 'artifact'
    except Exception:
        pass
    return artifact  # Returns the local transient value, not the canonical one

# Fix: try to persist; if someone already did, read what actually won
def save_artifact_canonical(stage_id: str, artifact: bytes) -> bytes:
    inserted = db.fetchone(
        """
        INSERT INTO artifacts (stage_id, artifact)
        VALUES (%s, %s)
        ON CONFLICT (stage_id) DO NOTHING
        RETURNING artifact
        """,
        (stage_id, artifact)
    )
    if inserted:
        return inserted[0]  # We won the insert; our value is canonical
    # Someone else's artifact already owns this stage identity: read it
    row = db.fetchone(
        "SELECT artifact FROM artifacts WHERE stage_id = %s",
        (stage_id,)
    )
    return row[0]  # Canonical, persisted artifact — never the local transient value

Failure 2: the parallel job ordering trap

Now consider a workflow transition that creates two jobs: A updates an external system based on the canonical result, while B prepares the next stage. The implementation contains an implicit assumption that B will see the effects of A. But the scheduler does not know about that dependency. It treats A and B as two independent jobs that are ready to run, so their execution order is not guaranteed.

One possible sequence looks like this:

1. The workflow state change is committed.

2. A worker picks up job B.

3. B reads the state before the effects of A are visible.

4. B takes the wrong branch, fails, or prepares stale input.

5. Only later does A complete successfully.

Retrying the jobs or making their handlers idempotent does not fix the broken ordering. The pipeline may even finish eventually, but by then it has already made a decision based on incorrect state.

The order in which jobs are added to a queue does not create a happens-before relationship. That is the key rule here. Invariant: if job B requires the effects of job A, the A → B dependency must be recorded in durable workflow state as a real happens-before edge. Alternatively, the state change required by B and the creation of B itself must be tied together transactionally.

# Bug: sibling jobs with implicit ordering
def emit_jobs_buggy(stage_id: str):
    # Both jobs are emitted as independent siblings
    workflow.emit_job("update_external", stage_id)
    workflow.emit_job("prepare_next", stage_id)  # No ordering guarantee

# Fix: chain the dependent job from the successful predecessor
def emit_jobs_chained(stage_id: str):
    # A emits B only after A completes successfully
    workflow.emit_job("update_external", stage_id, on_success=["prepare_next"])

Failure 3: the persist-to-crash window trap

Consider a pipeline where an AI stage generates a draft, persists it as the canonical artifact, and then creates a job responsible for verifying it. The sequence may look like this:

1. The AI stage generates draft D.

2. D is durably persisted as the canonical draft.

3. The process crashes before the verification job is durably created.

4. On retry, the stage sees that D already exists.

5. The idempotent write changes nothing.

6. Because the verification job was created only after a successful initial INSERT, verification is never scheduled again.

The result is especially subtle: the canonical artifact exists, but the process has no way to move to the next step. The state is durable, locally correct, and incomplete at the same time. The pipeline is stuck.

The role of the outbox pattern in this scenario is worth calling out. The main point here is not reliable message delivery. It is atomically persisting two things: the state change and the intent to perform the next step. Invariant: persisting the artifact and creating — or guaranteeing the existence of — its continuation must be tied together transactionally. If that cannot be done in a single transaction, the system needs a reconciliation mechanism that can detect and recreate the missing continuation.

# Bug: persistence and continuation are not atomic
def process_draft_buggy(stage_id: str, draft: bytes):
    # Persist the draft
    db.execute(
        "INSERT INTO drafts (stage_id, draft) VALUES (%s, %s) ON CONFLICT DO NOTHING",
        (stage_id, draft)
    )
    # Schedule verification (may fail after persistence but before commit)
    workflow.emit_job("verify_draft", stage_id)

# Fix: transactional outbox couples persistence and continuation
def process_draft_atomic(stage_id: str, draft: bytes):
    with db.transaction():
        # Persist the draft
        db.execute(
            "INSERT INTO drafts (stage_id, draft) VALUES (%s, %s) ON CONFLICT DO NOTHING",
            (stage_id, draft)
        )
        # Persist the continuation event in the outbox table
        db.execute(
            "INSERT INTO outbox (event_type, payload) VALUES ('verify_draft', %s)",
            (json.dumps({"stage_id": stage_id}),)
        )
    # The outbox publisher will emit the verification job after commit

Regression test patterns for each failure

The most important rule behind these tests is simple: test system behavior at crash-recovery boundaries, not just the final business outcome. These are the points where the difference between "the operation is idempotent" and "the whole process is retry-safe" becomes visible.

Local value trap test — canonical artifact invariant

def test_canonical_artifact_on_retry():
    """
    Verifies that a retry after a crash uses the canonical, persisted artifact,
    not the local transient value.
    """
    stage_id = "test_run/stage_1"
    # First execution persists artifact A
    artifact_a = b"artifact A"
    save_artifact_canonical(stage_id, artifact_a)

    # Retry after a crash computes a different artifact B
    artifact_b = b"artifact B"
    result = save_artifact_canonical(stage_id, artifact_b)

    # Downstream must see A, not B
    assert result == artifact_a

Parallel job ordering trap test

def test_sibling_ordering():
    """
    Verifies that a dependent job does not run before its prerequisite.
    """
    # Workflow with two siblings where B depends on the effects of A
    workflow = Workflow()
    workflow.add_job("A", effects=["write_state_x"])
    workflow.add_job("B", depends_on=["A"], reads=["state_x"])

    # Run the workflow with a scheduler that may interleave execution
    # This test should fail if B runs before A completes
    result = workflow.run()
    assert result["B"]["state_x"] == result["A"]["state_x"]

Persist-to-crash window test — transactional continuation invariant

def test_crash_window_between_persist_and_continuation():
    """
    Crashes at every boundary between persistence, the outbox insert, commit, and claim.
    Verifies that exactly one canonical draft eventually exists
    and that verification becomes runnable.
    """
    for crash_point in ["after_persist", "after_outbox_insert", "after_commit", "after_claim"]:
        # Reset state
        db.execute("DELETE FROM drafts")
        db.execute("DELETE FROM outbox")

        # Run the pipeline with a crash at the specified point
        try:
            run_pipeline_with_crash(crash_point)
        except SimulatedCrash:
            pass

        # Retry the pipeline
        run_pipeline()

        # Verification: exactly one canonical draft exists
        drafts = db.fetchall("SELECT * FROM drafts")
        assert len(drafts) == 1

        # Verification: the verification job is runnable
        outbox_events = db.fetchall("SELECT * FROM outbox WHERE event_type = 'verify_draft'")
        assert len(outbox_events) == 1

Summary: three invariants for retry-safe pipelines

1. Canonical artifact invariant: one durable stage identity must map to exactly one canonical, persisted artifact. Every retry must continue from that artifact, not from a local value produced by a later attempt. A uniqueness constraint in the database is necessary, but not sufficient. You also need to guarantee that the value that actually won the write is the same value passed downstream through the pipeline.

1. Durable dependency invariant: if job B requires the effects of job A, that dependency must exist in durable system state. It should be an explicit happens-before edge in the workflow job graph, not something implied only by the order in which both jobs were added to the queue.

1. Atomic continuation invariant: committing a state change and recording the intent to perform the next step must be atomic. In practice, this usually means a job record or an outbox table entry written in the same transaction. This is a different property from durable dependency: a dependency defines ordering between jobs that already exist, while atomic continuation prevents a committed artifact from being left with no next step at all.

These three invariants, combined with the right regression tests, can catch retry-related bugs that easily slip through ordinary test suites. The main takeaway is simple: an idempotent database write does not make the whole process idempotent. You need to ensure that every attempt converges on the same canonical state and continues from that state — not merely that it hits the same database row.