Idempotent by default: why every pipeline step must be replay-safe

A step is idempotent if running it twice leaves the world in the same state as running it once. It sounds like a small property. In practice it decides whether you can retry, replay and recover without fear.

Why replay safety matters

Pipelines fail in the middle. A network call times out after the write succeeded. A worker is restarted during a deployment. A scheduler fires the same run twice because of a clock change. Whatever you do, a step will sometimes run again.

If your step is not safe to repeat, you now have two bad options: never retry and accept gaps, or retry and accept duplicates. Neither is acceptable for data that people rely on. If the step is safe to repeat, the whole system becomes simple. Retries are free, a crashed run just continues, and "run it again to be sure" is a valid operational answer.

This is why Synchronoid treats replay safety as a rule rather than a tip. Runs are durable and resume from the last finished step, and that only works because steps are expected to be repeatable.

Pattern 1: natural keys

The most reliable defence against duplicates is an identity that does not depend on when or how often you ran. Use a key that comes from the data itself, such as an order number, a registry id, or a meter id plus a timestamp. Do not use an auto-incrementing counter or a random id generated during the run.

# Bad: a new id on every run
def store(ctx, items):
    for item in items:
        ctx.db.insert({"id": new_uuid(), **item})

# Good: the key comes from the data
def store(ctx, items):
    for item in items:
        ctx.db.upsert(key=item["order_no"], value=item)

Pattern 2: upserts instead of inserts

Once you have a natural key, writes become "set this record to this state". In Postgres this is an INSERT ... ON CONFLICT, and most warehouses have an equivalent MERGE.

INSERT INTO staged.orders (order_no, status, amount, run_id)
VALUES ($1, $2, $3, $4)
ON CONFLICT (order_no) DO UPDATE
SET status = EXCLUDED.status,
    amount = EXCLUDED.amount,
    run_id = EXCLUDED.run_id
WHERE staged.orders.status IS DISTINCT FROM EXCLUDED.status
   OR staged.orders.amount IS DISTINCT FROM EXCLUDED.amount;

Note the WHERE clause. Without it, a replay rewrites every row and produces a flood of fake "updates". With it, a repeated run changes nothing, which is exactly what idempotent means.

Pattern 3: content hashes

Some sources give you no stable key, or you want to know whether a payload has changed at all. Hash the content and use the hash as the identity.

import hashlib, json
import synchronoid as sy

def content_hash(payload: dict) -> str:
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode()).hexdigest()

@sy.step
def store_snapshot(ctx, payload):
    h = content_hash(payload)
    if ctx.raw.exists(h):
        return ctx.raw.ref(h)      # already stored, nothing to do
    return ctx.raw.put(h, payload)

The important detail is canonical serialization: sort keys and fix separators, or the same data will produce different hashes. Content-addressed storage also gives you free deduplication in the raw layer.

Pattern 4: write the result, then mark it done

For steps that call an external system, order matters. Make the external effect safe to repeat, for example by sending an idempotency key with the request, and only then record success.

@sy.step(retry=sy.Retry(max_attempts=5, backoff="exponential"))
def notify_partner(ctx, change):
    key = f"{ctx.run_id}:{change['id']}"
    ctx.http.post("https://partner.example.com/hooks/orders", json=change, headers={"Idempotency-Key": key})

Anti-patterns

Appending without a key

"Just append the new rows" is the most common source of duplicates. If you must append, add a unique constraint so that a replay fails loudly instead of silently doubling your data.

Using the current time as identity

A key like now() or a run-start timestamp differs on every attempt. Use the time that belongs to the data, such as the reading time of a meter.

Counters and "last processed" offsets in memory

State that lives in a worker's memory disappears when the worker does. Keep progress in durable storage, and make it part of the step's output.

Delete-then-insert without a transaction

If a crash happens between the two statements, you have lost data. Use a transaction, or better, an upsert.

Hidden side effects

Sending an email or posting to a chat channel inside a data step means a retry sends it twice. Move side effects to a dedicated step with its own idempotency key, or emit a change event and let a consumer deduplicate.

Testing replay safety

The best test is simple: run the step twice and compare. Synchronoid makes this a command:

$ syn runs replay run_01J9Z4 --verify
Replaying 3 steps against a scratch copy of staged...
  fetch_orders        ok   0 rows changed
  normalize_orders    ok   0 rows changed
  publish_orders      ok   0 rows changed
Replay verified: no differences.

If a replay changes anything, the step is not idempotent, and the output tells you which one. We recommend running this in continuous integration for every pipeline. You can find the retry and replay settings in the retries and replays guide.

The payoff

Idempotent steps are not just a defence against failure. They make backfills trivial, because reprocessing a month of data is the same operation as processing a day. They make deployments calmer, because a restart is never an incident. And they make orchestration honest: the platform can retry on your behalf because it knows retrying is safe.

Make every step repeatable, and the rest of the system gets easier.