Designing a pipeline that can be re-run

An agent built you a nightly job. It pulls yesterday’s survey responses from a vendor API, runs each open-text answer through an LLM to tag sentiment, writes rows into Postgres, and a React dashboard reads the result. It worked when you tested it. Three weeks later the dashboard shows 1,400 responses for a day that had 700, and nobody can say which run did what.

Nothing went wrong in a dramatic way. What happened is that you have a script, not a pipeline. The difference is not size or sophistication. It is whether the thing can be re-run.

Stages, and why each one needs durable edges

Most data work decomposes into the same five jobs: extract (fetch raw input), transform (reshape and clean), enrich (add derived fields, including LLM steps), load (write to your own store) and publish (expose it downstream). Azure’s Pipes and Filters pattern makes the useful point about them: a stage is only independently retryable if it reads its input from durable storage and writes its output to durable storage.

That is the whole design constraint, and it is the one agents skip. Here is the version they emit:

# One function, nothing durable in the middle.
async def nightly_job() -> None:
    raw = await fetch_responses()            # lives only in memory
    cleaned = [clean(r) for r in raw]
    tagged = [await tag_sentiment(c) for c in cleaned]  # 40 minutes of LLM calls
    await insert_all(tagged)                 # one shot at the end

If tag_sentiment fails on item 680 of 700, you lose the fetch, the cleaning and 679 LLM calls. And because nothing was written, you cannot even inspect what it saw. The re-run starts from zero and pays for the tagging twice.

The fix is boring: give every stage a place to put its output.

async def nightly_job(run_id: str, day: date) -> None:
    # Each stage reads from and writes to storage, so each can be re-run alone.
    await extract_to_raw(run_id, day)        # writes raw payloads to R2 / a raw table
    await transform_raw(run_id)              # raw -> responses_clean
    await enrich_sentiment(run_id)           # fills sentiment where it is NULL
    await publish_marts(run_id)              # refreshes what the dashboard reads

Now enrich_sentiment can be re-run on its own and will only touch rows still missing a tag. That single property — a stage that knows what is left to do — is what makes a 40-minute LLM step survivable.

A stage that behaves like this is often described as pure-ish: its output depends only on its declared input plus a pinned config, and its one side effect is a single well-defined write. A stage that reads the wall clock, a global counter or “whatever happens to be in the uploads folder” cannot be re-run with confidence, because you can never rebuild the input it actually saw.

Idempotency and natural keys

Idempotency means doing something twice has the same effect as doing it once. It is the property that makes re-running safe, and in a pipeline it almost always comes down to how you write rows.

# What agents write: duplicates on every re-run.
await conn.execute(
    "INSERT INTO responses (respondent_id, question_id, answer) VALUES ($1,$2,$3)",
    r.respondent_id, r.question_id, r.answer,
)

The table has an auto-increment id. The database has no idea this row is the same one as last night’s, so a second run gives you a second copy. That is where the 1,400 came from.

-- Declare what identifies one unit of work, for all time.
ALTER TABLE responses
  ADD CONSTRAINT responses_natural_key
  UNIQUE (survey_id, respondent_id, question_id);

# Then the write is an upsert: run it 1 or 50 times, you get one row.
await conn.execute(
    """
    INSERT INTO responses (survey_id, respondent_id, question_id, answer, run_id)
    VALUES ($1,$2,$3,$4,$5)
    ON CONFLICT (survey_id, respondent_id, question_id)
    DO UPDATE SET answer = EXCLUDED.answer, run_id = EXCLUDED.run_id
    """,
    r.survey_id, r.respondent_id, r.question_id, r.answer, run_id,
)

Those identifying fields are the natural key: a key built from the data itself rather than handed out by the database. Pick the fields that genuinely identify one thing forever. If there are none — a scraped news item, say — hash the stable parts of the payload:

import hashlib

def natural_key(source: str, url: str, published: str) -> str:
    # Same input, same key, on any machine, on any run. Never include the fetch time.
    return hashlib.sha256(f"{source}|{url}|{published}".encode()).hexdigest()[:32]

Watch for a hash that includes datetime.now() or a random UUID. That is a very common agent mistake, and it makes the key useless: every run produces new keys, so every run duplicates everything while looking perfectly deduplicated.

Do not reach for SELECT-then-INSERT

The tempting alternative is to check first. It races: two workers can both read “absent” and both insert. Let the database’s unique constraint be the referee — it is the only thing that can decide atomically.

Checkpoints, and the ordering that matters

A checkpoint is a durable marker of progress so a re-run can skip finished work. Record the unit completed — a page cursor, a batch id, a high-water-mark timestamp — not a count, because counts drift the moment anything is retried.

The ordering is the part that bites:

# Wrong. A crash between these two lines loses the page silently, forever.
await save_cursor(run_id, page.next_cursor)
await upsert_rows(page.rows)

# Right. Commit the work, then claim it.
await upsert_rows(page.rows)          # durable first
await save_cursor(run_id, page.next_cursor)   # only now is the claim true

Checkpointing first turns a crash into silent data loss. Checkpointing afterwards turns the same crash into a duplicate — and because your writes are idempotent upserts, that duplicate costs nothing. This is the general shape of the trade: prefer the failure mode your idempotency already absorbs.

Run records: how you find out what happened

A run record is one row per execution. It is the single highest-value thing you can add to an agent-written job, because it converts “the job seems fine” into evidence.

CREATE TABLE pipeline_runs (
  run_id        TEXT PRIMARY KEY,
  pipeline      TEXT NOT NULL,
  params        JSONB NOT NULL,       -- the day/window this run covered
  code_version  TEXT NOT NULL,        -- git sha
  config_hash   TEXT NOT NULL,        -- prompt + model + settings
  started_at    TIMESTAMPTZ NOT NULL,
  finished_at   TIMESTAMPTZ,
  status        TEXT NOT NULL,        -- running | succeeded | partial | failed
  rows_read     INT DEFAULT 0,
  rows_written  INT DEFAULT 0,
  rows_failed   INT DEFAULT 0
);

Stamp run_id on every row you write and every log line you emit. In Application Insights that means one field you can filter on to see a single run end to end; in your own tables it means you can answer “which run produced this number?” without archaeology.

At-least-once, exactly-once, and the honest version

Schedulers and queues retry anything they are unsure about, so a unit of work may be processed more than once. That is at-least-once delivery, and it is what Cloudflare Queues, Azure Service Bus and almost every cron-plus-worker arrangement actually give you. Exactly-once means each unit takes effect once and only once.

You do not get exactly-once from a setting. In practice it is at-least-once delivery plus idempotent writes, so duplicate processing has no visible effect. If a docstring claims exactly-once and the load step is a plain INSERT with a generated id, the claim is decoration.

Partial failure and dead letters

Over 4,000 documents, a dozen will time out, return malformed JSON or trip a content filter. That is not an exception, it is Tuesday. Two instincts are both wrong: failing the whole run throws away 3,988 good results, and logging a warning means a stakeholder finds the gap before you do.

Handle units independently and write the failures down:

failed = 0
for doc in batch:
    try:
        result = await extract_fields(doc)          # LLM step
        await upsert_extraction(natural_key(doc), result, run_id)
    except (ValidationError, TimeoutError, LLMError) as exc:
        await insert_dead_letter(
            key=natural_key(doc), run_id=run_id,
            payload=doc.model_dump(),               # Pydantic v2: the exact input
            error=repr(exc), attempts=MAX_ATTEMPTS,
        )
        failed += 1
await finish_run(run_id, status="partial" if failed else "succeeded", rows_failed=failed)

A dead-letter table is only worth having if each entry can be re-run alone, which means it holds the natural key, the exact input payload, the error and the attempt count. Then put its row count on a dashboard and alert when it grows. A dead-letter table nobody looks at is a slower way of losing data.

Making a run reproducible

Six months later somebody disputes a number. To defend it you need three things.

The LLM stage is where this is usually missing. If prompts live inline in code that has since been edited, and the model was requested by a floating alias, you cannot reconstruct what the run asked — only approximate it.

# Prompts as versioned data, not string literals buried in a function.
PROMPT_VERSION = "sentiment-v3"
PROMPTS = {"sentiment-v3": (Path(__file__).parent / "prompts/sentiment-v3.txt").read_text()}

await upsert_extraction(
    key=key, result=result, run_id=run_id,
    prompt_version=PROMPT_VERSION,
    model=settings.model_id,   # a pinned, dated model id from config - never a floating alias
)

Snapshotting inputs also buys you backfills. When you add a field or fix a bug, you re-run the enrich and load stages over stored raw data, rather than begging the vendor for a historical export. A pipeline that consumes its input destructively cannot be backfilled at all.

What to check in your own code

• Grep for INSERT INTO with no ON CONFLICT or MERGE, on any table a scheduled job writes to. Each one is a duplicate waiting for a retry.
• Check every dedup hash for now(), utcnow(), uuid4() or a fetch timestamp in its input. If one is there, the hash guarantees duplicates instead of preventing them.
• Find where the checkpoint or cursor is saved and confirm the data write happens before it, not after.
• Ask whether one stage can be run on its own. If the only entry point is a single main() that does everything, there are no durable edges.
• Look for a pipeline_runs table or equivalent. If there isn’t one, you have no way to answer “which run wrote this row?”.
• Find where failed units go. If the answer is logger.warning, add a dead-letter table with the payload, and a count you can alert on.
• Check that prompts and model ids are versioned data recorded per run, not literals in a file that changes.
• Try to describe, out loud, what happens if the job is triggered twice at once. If you can’t, that is the next thing to test in staging.