Idempotent Batch Replays for Instrument Result Loads

Problem Statement

A worker dies halfway through committing a rack of results — the pod is evicted, the broker redelivers the unacknowledged batch, and the replay re-inserts the fifty results that already landed alongside the fifty that did not. Now the LIMS holds two potassium results for the same draw, a reviewer sees a phantom delta, and an auto-verification rule fires on a duplicate the analyzer only produced once. Crash recovery is not optional in a clinical pipeline; the broker will redeliver, and the only question is whether the second delivery is harmless. This page makes it harmless by deriving a deterministic dedup key from the result’s natural identity — accession, analyte, instrument, and the instrument’s own resulted_at timestamp — and committing through an upsert whose conflict target is that key, so replaying a batch converges on exactly the state a single clean run would have produced. It is the replay-safety guarantee that the bounded-concurrency drain relies on, and it lives under the Async Batch Processing stage.

Prerequisites

Before making a result load replay-safe, confirm the baseline:

  • Runtime: Python 3.11+, pydantic>=2.6 for the keyed result model, and an async datastore that supports an atomic upsert — PostgreSQL INSERT ... ON CONFLICT, or an equivalent MERGE. pytest>=8 with pytest-asyncio for the fixtures.
  • A stable natural key: the four fields that together identify a result independent of transport — accession number, analyte identity (LOINC), producing instrument, and the analyzer’s result timestamp. If any of these is assigned by the pipeline rather than the instrument, the key is not stable across a replay and dedup will fail.
  • At-least-once delivery: a broker that guarantees delivery but not exactly-once, which is every practical broker. Idempotency is what turns at-least-once into effectively-once at the datastore.
  • A concurrency ceiling: the bounded-concurrency drain for analyzer result batches, which can cancel a worker mid-commit — the exact event this page must tolerate.

Step-by-Step Implementation

Step 1: Derive a deterministic dedup key

The key must depend only on values the instrument produced, never on wall-clock time, a UUID minted at ingest, or the broker’s message id — any of which change on replay and defeat deduplication. Normalize each component (trim, case-fold the analyte code, render the timestamp in UTC at fixed precision) so two byte-different-but-semantically-identical deliveries hash the same. A SHA-256 over the canonical tuple gives a fixed-width, index-friendly key.

python
from __future__ import annotations

import datetime as dt
import hashlib

from pydantic import BaseModel, Field, field_validator


def _canon_ts(value: dt.datetime) -> str:
    """UTC, second precision — instrument timestamps rarely mean sub-second."""
    return value.astimezone(dt.timezone.utc).replace(microsecond=0).isoformat()


class InstrumentResult(BaseModel):
    accession: str
    analyte_loinc: str
    instrument_id: str
    resulted_at: dt.datetime
    value: float
    unit: str

    @field_validator("accession", "instrument_id")
    @classmethod
    def _strip(cls, v: str) -> str:
        v = v.strip()
        if not v:
            raise ValueError("identity field must be non-empty")
        return v

    @field_validator("analyte_loinc")
    @classmethod
    def _norm_loinc(cls, v: str) -> str:
        return v.strip().upper()

    @property
    def dedup_key(self) -> str:
        parts = "|".join(
            [self.accession, self.analyte_loinc, self.instrument_id, _canon_ts(self.resulted_at)]
        )
        return hashlib.sha256(parts.encode("utf-8")).hexdigest()

The key is a pure function of the model, so the same result always yields the same key regardless of when or how many times it is loaded — the foundation of replay safety.

Step 2: Give the datastore a unique constraint on the key

Idempotency cannot live only in application code; a race between two workers replaying the same message would both check-then-insert and both win. The database must enforce uniqueness so the second insert is rejected atomically. Declare the constraint on dedup_key and let the engine, not the application, be the arbiter.

sql
CREATE TABLE result_load (
    dedup_key     CHAR(64) PRIMARY KEY,          -- SHA-256 hex of the natural identity
    accession     TEXT        NOT NULL,
    analyte_loinc TEXT        NOT NULL,
    instrument_id TEXT        NOT NULL,
    resulted_at   TIMESTAMPTZ NOT NULL,
    value         DOUBLE PRECISION NOT NULL,
    unit          TEXT        NOT NULL,
    loaded_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
    load_count    INTEGER     NOT NULL DEFAULT 1  -- observability: how often replayed
);

Step 3: Commit through an upsert with the key as conflict target

The write is an INSERT ... ON CONFLICT (dedup_key). On first load it inserts; on replay the conflict clause fires and the row is left untouched except for a load_count bump that makes replays observable rather than invisible. Crucially, the clinical columns are not overwritten on conflict — a replay must never mutate an already-committed result, only confirm it.

python
from typing import Protocol


class Pool(Protocol):
    async def execute(self, sql: str, *args: object) -> str: ...


UPSERT = """
INSERT INTO result_load
    (dedup_key, accession, analyte_loinc, instrument_id, resulted_at, value, unit)
VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (dedup_key) DO UPDATE
    SET load_count = result_load.load_count + 1
RETURNING (xmax = 0) AS inserted
"""


async def upsert_result(pool: Pool, r: InstrumentResult) -> bool:
    """Return True if this was a first insert, False if a replay was absorbed."""
    row = await pool.execute(
        UPSERT, r.dedup_key, r.accession, r.analyte_loinc,
        r.instrument_id, r.resulted_at, r.value, r.unit,
    )
    return bool(row)

The DO UPDATE on conflict (rather than DO NOTHING) is deliberate: DO NOTHING returns no row, making it impossible to distinguish “absorbed a replay” from “insert skipped for another reason,” whereas the xmax = 0 trick reports whether the row was freshly inserted, which the audit trail needs.

Step 4: Replay a whole batch and reconcile the counts

Batch replay is now just re-running the load. Because each upsert_result is individually idempotent, the batch is idempotent: run it once and every result inserts; run it again after a crash and every already-committed result is absorbed while only the genuinely missing tail inserts. The returned inserted flags let the caller reconcile — a healthy replay reports zero new inserts for the portion that completed before the crash.

python
class ReplayReport(BaseModel):
    total: int = Field(ge=0)
    inserted: int = Field(ge=0)
    absorbed: int = Field(ge=0)


async def load_batch(pool: Pool, batch: list[InstrumentResult]) -> ReplayReport:
    inserted = 0
    for r in batch:
        if await upsert_result(pool, r):
            inserted += 1
    return ReplayReport(
        total=len(batch), inserted=inserted, absorbed=len(batch) - inserted
    )
Idempotent replay: a redelivered batch converges via a keyed upsert without duplicate results Two deliveries of the same batch are shown. The first delivery reaches a crash partway through, so only its earlier results commit before the worker dies and the batch is left unacknowledged. The broker redelivers the full batch as a second delivery. Each result derives a deterministic dedup key from accession, analyte, instrument, and resulted_at, and is committed through an upsert with the dedup key as the conflict target. Results already committed by the first delivery hit the conflict clause and are absorbed with only a load_count increment, while the tail that never committed is inserted. The result table ends in exactly the state one clean run would have produced, with no duplicate rows. First delivery 50 results committed crash mid-batch nack Broker redelivers full batch, second time at-least-once Derive dedup_key sha256(accession|analyte| instrument|resulted_at) Upsert on conflict ON CONFLICT (dedup_key) key exists new key Absorbed (replay) row untouched load_count += 1 Inserted (tail) missing results only no duplicate rows
The dedup key is a pure function of the instrument's own identity fields, so a redelivered batch upserts on conflict — absorbing what already committed and inserting only the missing tail.

Verification & Testing

The property to prove is convergence: loading a batch twice leaves the datastore identical to loading it once. Use an in-memory fake that enforces the same unique-key semantics as the database.

python
import pytest


class FakePool:
    def __init__(self) -> None:
        self.rows: dict[str, int] = {}  # dedup_key -> load_count

    async def execute(self, sql, *args) -> str:
        key = args[0]
        if key in self.rows:
            self.rows[key] += 1
            return ""            # conflict path: no row returned -> absorbed
        self.rows[key] = 1
        return "inserted"        # first insert -> truthy


def _mk(analyte: str) -> InstrumentResult:
    return InstrumentResult(
        accession="A100", analyte_loinc=analyte, instrument_id="chem-1",
        resulted_at=dt.datetime(2026, 7, 1, 14, 30, tzinfo=dt.timezone.utc),
        value=4.2, unit="mmol/L",
    )


@pytest.mark.asyncio
async def test_replay_is_idempotent():
    pool = FakePool()
    batch = [_mk("2823-3"), _mk("2951-2"), _mk("2075-0")]

    first = await load_batch(pool, batch)
    second = await load_batch(pool, batch)

    assert first.inserted == 3 and first.absorbed == 0
    assert second.inserted == 0 and second.absorbed == 3   # nothing new on replay
    assert len(pool.rows) == 3                             # no duplicate rows
    assert all(count == 2 for count in pool.rows.values()) # each seen twice


def test_dedup_key_is_stable_under_noise():
    a = _mk("2823-3")
    b = InstrumentResult(accession=" A100 ", analyte_loinc="2823-3 ".strip().lower(),
                         instrument_id="chem-1", resulted_at=a.resulted_at,
                         value=4.2, unit="mmol/L")
    assert a.dedup_key == b.dedup_key  # whitespace/case differences collapse

Expected: both tests pass. test_replay_is_idempotent confirms the second load inserts nothing and creates no duplicate rows; test_dedup_key_is_stable_under_noise confirms the normalizers make cosmetically different-but-identical payloads share a key. A third fixture should feed a result with a genuinely different resulted_at and assert a new row is inserted — proving the key discriminates real re-runs (an actual re-test) from transport replays.

Compliance Note

Duplicate results are a data-integrity defect with direct patient-safety consequences: a phantom second potassium can trigger a spurious delta flag or a duplicated critical-value page, and CLIA §493.1291 requires that the reported result accurately reflect what the analyzer produced — once. Enforcing uniqueness at the datastore, not merely in application logic, satisfies the 21 CFR Part 11 expectation that the electronic record be attributable and not silently altered by system behavior; a replay confirms the existing record rather than overwriting it. The load_count column turns replays into an observable, auditable signal instead of an invisible event, and because the dedup key is a documented, deterministic function of the instrument’s own identity fields, an inspector can recompute it for any row and confirm the record maps to exactly one real analyzer result. Pair the upsert with an append-only audit entry recording the first-insert-versus-absorbed outcome so the load history is reconstructable.

Troubleshooting

Replays still create duplicate rows despite the dedup key.

Root cause: the key includes a value that changes on replay — an ingest-time UUID, now(), or the broker message id — so the two deliveries hash differently. Fix: build the key only from instrument-produced identity fields (accession, analyte, instrument, resulted_at), and confirm no pipeline-assigned field leaks into dedup_key.

Two concurrent workers both insert the same result.

Root cause: idempotency is enforced only in application code with a check-then-insert, which races. Fix: put a PRIMARY KEY or UNIQUE constraint on dedup_key and use INSERT ... ON CONFLICT so the database rejects the second insert atomically — the arbiter must be the engine, not the coroutine.

A legitimate re-test gets swallowed as a replay.

Root cause: the analyzer re-ran the specimen and produced a new value, but the key ignores the new resulted_at, so the upsert absorbs it. Fix: keep resulted_at in the key at the instrument’s real precision; a genuine re-test carries a different result timestamp and therefore a different key, which correctly inserts a new row rather than masking the repeat.

Timestamps that are the same instant produce different keys.

Root cause: one delivery carries a naive local timestamp and another carries the UTC offset, or sub-second jitter differs. Fix: canonicalize with _canon_ts — convert to UTC and truncate to second precision before hashing — so the same clinical instant always canonicalizes to the same string.

You cannot tell whether a batch replay was healthy.

Root cause: using ON CONFLICT DO NOTHING, which returns no row and hides whether a replay was absorbed. Fix: use DO UPDATE ... SET load_count = load_count + 1 RETURNING (xmax = 0) so each call reports inserted-versus-absorbed; a healthy post-crash replay shows zero new inserts for the completed portion and non-zero for the missing tail.

Part of: Async Batch Processing.