Idempotent Retries for Flight Schedule Ingestion

The exact task this guide solves is making the write side of schedule ingestion safe to retry, so that a batch delivered twice — after a timeout, a worker crash, or a broker redelivery — produces exactly the same database state as one delivered once. Retrying is unavoidable in an asynchronous pipeline, but a retried write that is not idempotent double-inserts duty events, which then double-count in every flight-time and cumulative-cap calculation downstream. The fix is a deterministic idempotency key, an upsert that collapses duplicates, and an inbox that records what has already been processed. This page builds all three. It completes the async batch processing workflows section and guarantees the flight data ingestion pipeline is safe under the retries the roster API client performs.

Prerequisites

Why at-least-once delivery forces idempotency

Asynchronous pipelines deliver at least once, not exactly once: a message can be redelivered whenever an acknowledgement is lost, even though the work succeeded. Exactly-once processing is achievable, but only by making the effect idempotent — writing in a way where applying the same message twice equals applying it once. Two ingredients deliver that: a stable key that identifies the event regardless of how many times it arrives, and a write that recognises the key and updates rather than duplicates. An inbox table then records processed keys so an entire batch can be short-circuited on redelivery.

Step 1 — Derive a deterministic idempotency key

The key must be a pure function of the event’s identity — never a random id or an arrival timestamp — so the same event always produces the same key. Hashing the natural identity fields yields a compact, stable key.

import hashlib


def idempotency_key(flight_id: str, event_kind: str, instant_utc: str) -> str:
    """Stable key from the event's natural identity — same event, same key."""
    material = f"{flight_id}|{event_kind}|{instant_utc}".encode("utf-8")
    return hashlib.sha256(material).hexdigest()

Verify: the same flight id, kind, and instant always produce the same digest; changing any field changes it. A random UUID would defeat idempotency by making every redelivery look new.

Step 2 — Upsert so a duplicate write is a no-op

A unique constraint on the key plus an ON CONFLICT clause turns a second write of the same event into an update of the existing row rather than a duplicate insert. The event is written once no matter how many times the batch is retried.

CREATE TABLE duty_event (
    idempotency_key text PRIMARY KEY,
    flight_id       text NOT NULL,
    event_kind      text NOT NULL,
    instant_utc     timestamptz NOT NULL,
    payload         jsonb NOT NULL
);

INSERT INTO duty_event (idempotency_key, flight_id, event_kind, instant_utc, payload)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (idempotency_key) DO UPDATE
    SET payload = EXCLUDED.payload,      -- latest wins, but row count stays one
        instant_utc = EXCLUDED.instant_utc;

Verify: inserting the same event twice leaves exactly one row; the second insert updates the payload in place, so a retried batch never inflates the event count that downstream flight-time sums depend on.

Step 3 — Short-circuit whole batches with an inbox

Re-running every upsert on a redelivered batch is correct but wasteful. An inbox table records the batch id once it has been fully processed, so a redelivery is recognised and skipped in a single lookup, and the inbox insert shares the transaction with the writes so the two can never disagree.

import asyncpg


async def process_batch(conn: asyncpg.Connection, batch_id: str, events: list[dict]) -> str:
    async with conn.transaction():
        already = await conn.fetchval(
            "SELECT 1 FROM ingest_inbox WHERE batch_id = $1", batch_id
        )
        if already:
            return "skipped_duplicate"
        for e in events:
            await conn.execute(UPSERT_SQL, e["key"], e["flight_id"],
                               e["kind"], e["instant_utc"], e["payload"])
        await conn.execute("INSERT INTO ingest_inbox (batch_id) VALUES ($1)", batch_id)
    return "processed"

Verify: the first delivery of a batch returns "processed" and writes its events; a redelivery returns "skipped_duplicate" without touching duty_event, because the inbox row committed in the same transaction as the writes.

Idempotent ingestion with inbox and upsert An incoming batch is checked against an inbox table. If the batch id is already recorded, it is skipped as a duplicate. Otherwise each event is upserted by its idempotency key, so duplicate events collapse to a single row, and the batch id is recorded in the inbox within the same transaction. Batch seen in inbox? Skip — duplicate Upsert by key ON CONFLICT DO UPDATE duty_event one row / key inbox += batch_id (same txn) yes no
A redelivered batch is recognised by the inbox and skipped; a new batch upserts each event by its idempotency key so duplicates collapse, and records its batch id in the same transaction as the writes.

Failure modes and troubleshooting

Frequently Asked Questions

Why not rely on exactly-once delivery?

Because exactly-once delivery is effectively impossible over an unreliable network — an acknowledgement can always be lost. What is achievable is exactly-once effect, by making writes idempotent so a redelivered message changes nothing. Idempotency is the practical substitute for a guarantee the transport cannot give.

What makes a good idempotency key?

A deterministic function of the event’s natural identity — the flight, event kind, and instant — so the same event always hashes to the same key. Anything random or time-based breaks the property by making redeliveries look like new events.

Why is an inbox needed if the upsert is already idempotent?

The upsert makes each event safe, but re-running every upsert in a large redelivered batch is wasteful. The inbox recognises an already-processed batch in one lookup and skips it, and committing the inbox row with the writes keeps the two consistent.

How does this relate to retrying the fetch side?

The roster API client retries reads, which are naturally idempotent. This page makes the writes equally safe to retry, so the whole pipeline — fetch and persist — can recover from any transient failure without corrupting data.

Back to Async Batch Processing Workflows.