Instrument Control

Syncing Instrument Metadata to a LIMS over REST

Pushing enriched acquisition metadata from a lab host into a Laboratory Information Management System over a REST/HTTP API is where reproducible records go to die quietly: a lost 200 response duplicates a sample, a concurrent edit clobbers an ETag, and a LIMS outage silently drops a shift’s worth of runs. This guide makes that hop reliable with a client-generated record UUID, an idempotency key, and a local durable outbox, extending the enrichment stage described in Metadata Injection Workflows within the wider Data Capture, Validation & Metadata Sync architecture.

Scope: At-Least-Once REST Delivery from an Acquisition Host

The problem starts once a metadata record has already been validated and serialized — the deterministic payload produced by the injector — and must reach a LIMS reachable only over HTTP. Two properties of that transport dominate the design. First, HTTP over a lab network is unreliable in a specific way: the request can succeed on the server while the response is lost to a dropped connection, a proxy timeout, or the acquisition host being power-cycled mid-flight. The client cannot distinguish “the LIMS never saw it” from “the LIMS committed it but I never heard back,” so any retry must assume the record may already exist. Second, the acquisition loop must never block on the LIMS being up; a plate reader running overnight cannot pause because an appliance is mid-reboot.

Both constraints point to the same shape: an at-least-once pipeline built on a local durable outbox and made safe by idempotent writes. The acquisition thread does one cheap thing — append the validated record to a SQLite table and return. A separate draining worker owns every network concern: authentication, retry, status classification, and concurrency control. This is the durable-queue fallback that the parent guide names in the abstract; here it is the primary path, not a degraded one. The scope explicitly excludes payload construction (covered upstream), regulated provenance fields (see Injecting ALCOA+ Provenance Metadata for 21 CFR Part 11), and the clock discipline that stamps each record (see Deterministic UTC Timestamping for Multi-Instrument Runs). We assume Python 3.10+, a LIMS REST API that honours an Idempotency-Key header and conditional If-Match/ETag requests, and bearer-token auth.

Idempotency Keys, Status Classification, and the Outbox Drain Bound

Idempotency is enforced with two client-generated identifiers that travel with the record from the moment it is enqueued. The record UUID is the resource identity: the write is a PUT /records/{uuid}, so re-sending the same UUID targets the same server row rather than creating a new one — a PUT to a client-chosen key is naturally idempotent in a way a bare POST is not. The idempotency key is a per-attempt-safe token in the Idempotency-Key header that lets the LIMS collapse duplicate deliveries of the same logical write: if the server already committed that key, it replays the stored 200/201 instead of applying the body twice. The UUID gives you a stable address; the idempotency key gives you exactly-once effect on top of at-least-once delivery.

Whether a failed attempt is retried or abandoned is decided entirely by the HTTP status class, and getting that mapping wrong is how outboxes either grow forever or discard good data. The classification is the HTTP dialect of Error Code Categorization: 2xx is success and captures the returned ETag; 429 and 5xx are transient (rate-limit or server-side) and must be retried with a bounded, growing delay; a 409 Conflict signals an ETag mismatch and needs reconciliation before a retry can succeed; and any other 4xx — 400 malformed, 422 schema rejection, 403 scope — is a permanent client fault that must be quarantined to a dead-letter state and alerted, never retried, because the same bytes will fail identically forever. The one exception is 401, handled as a token-refresh-then-retry rather than a dead-letter.

The retry delay itself is the same bounded exponential curve derived for serial timeout handling — delay = min(base · 2^attempts, max_delay) — so it is not re-derived here; the outbox simply stores a next_attempt_at timestamp per row and the worker respects it. Optimistic concurrency uses the ETag as a version token: every read of the LIMS record returns an ETag, and the next write sends it back as If-Match. If another writer changed the record in between, the server rejects the write with 409 and the worker must refetch before retrying. This is what keeps two hosts editing the same sample record from silently overwriting each other. Finally, every payload carries an envelope version (X-Envelope-Version) so a LIMS schema change produces an explicit 422 you can quarantine and migrate, rather than a coerced, subtly-wrong record.

What bounds the outbox is a queueing argument. Let acquisition enqueue records at rate λ (records/s) and let the drain worker sustain throughput μ = c / t_req, where c is worker concurrency and t_req the mean successful round-trip. The backlog B(t) obeys a fill-and-drain balance, and during a LIMS outage of duration D it grows linearly, then drains only if μ > λ:

The design rule falls straight out of the denominator: provision μ with enough headroom over peak λ that μ − λ clears the worst tolerable outage backlog λD inside your recovery-time objective. A worker that can only just keep up (μ ≈ λ) never recovers from an outage — t_drain diverges — and the SQLite file grows without bound until the disk fills.

Durable outbox drain and HTTP status classification Top row, left to right: Enrich and enqueue on the acquisition thread appends a validated record to the Outbox, a SQLite table holding PENDING rows with a record UUID and idempotency key. A drain Worker claims due rows and issues PUT /records/{uuid} with Idempotency-Key and If-Match headers to the LIMS REST API. The LIMS response drops into a Classify status node that fans out to four outcomes: a 2xx success marks the row SENT and stores the new ETag; a 429 or 5xx transient response schedules a backoff retry that loops back to the outbox with attempts incremented; a 409 conflict triggers an ETag refetch then retry; any other 4xx is a permanent fault routed to a dead-letter quarantine. The 401 case refreshes the token and retries. Enrich + enqueue Outbox (SQLite) Drain worker LIMS REST acquisition thread PENDING · uuid · key claims due rows PUT /records/{uuid} Classify HTTP status If-Match · Idempotency-Key backoff · attempts++ 2xx → SENT 429 / 5xx → retry 409 → reconcile 4xx → dead-letter store new ETag next_attempt_at refetch ETag quarantine + alert
Records land in a durable SQLite outbox; the drain worker owns every network concern and routes each HTTP response into success, bounded retry, ETag reconciliation, or a permanent dead-letter path.

The outbox drain and status-classification path: 2xx commits and stores the ETag, 429/5xx reschedules with backoff, 409 reconciles the ETag before retrying, and non-retryable 4xx is quarantined.

A Durable Outbox and Idempotent LIMS Client

The implementation splits cleanly: LimsOutbox owns the SQLite durability and the retry bookkeeping, LimsClient.upsert_record() owns one idempotent HTTP attempt and its classification, and a small drain loop ties them together. It uses requests, but every call sits behind a method boundary, so a stdlib http.client/urllib transport drops in unchanged. The digest carried in each row is the transport-integrity tag from Checksum & CRC Validation, so the worker can detect a payload that was corrupted at rest before it ever hits the wire.

from __future__ import annotations

import sqlite3
import time
import uuid
from dataclasses import dataclass
from enum import Enum

import requests

_SCHEMA = """
CREATE TABLE IF NOT EXISTS outbox (
    record_uuid     TEXT PRIMARY KEY,
    idempotency_key TEXT NOT NULL,
    resource_path   TEXT NOT NULL,
    payload         BLOB NOT NULL,
    etag            TEXT,
    state           TEXT NOT NULL DEFAULT 'PENDING',   -- PENDING | SENT | DEAD
    attempts        INTEGER NOT NULL DEFAULT 0,
    next_attempt_at REAL NOT NULL DEFAULT 0,
    last_error      TEXT
);
"""


class Outcome(str, Enum):
    SUCCESS = "success"      # 2xx: committed, capture new ETag
    RETRY = "retry"          # 429/5xx/401: transient, reschedule with backoff
    CONFLICT = "conflict"    # 409: ETag mismatch, refetch then retry
    QUARANTINE = "quarantine"  # other 4xx: permanent client fault, dead-letter


@dataclass(frozen=True)
class BackoffPolicy:
    base_delay: float = 0.5      # seconds; doubled per attempt (bounded)
    max_delay: float = 60.0      # cap on any single wait
    max_attempts: int = 8        # dead-letter after this many transient failures

    def delay_for(self, attempts: int) -> float:
        """Bounded exponential delay; identical to the serial-backoff curve."""
        return min(self.base_delay * (2 ** attempts), self.max_delay)


class LimsOutbox:
    """Durable at-least-once queue of metadata records awaiting LIMS delivery."""

    def __init__(self, db_path: str) -> None:
        self._conn = sqlite3.connect(db_path, isolation_level=None)
        self._conn.execute("PRAGMA journal_mode=WAL;")  # crash-safe appends
        self._conn.executescript(_SCHEMA)

    def enqueue(self, resource_path: str, payload: bytes) -> str:
        """Append a validated record; returns its client-generated UUID."""
        record_uuid = str(uuid.uuid4())
        self._conn.execute(
            "INSERT INTO outbox(record_uuid, idempotency_key, resource_path, payload)"
            " VALUES (?, ?, ?, ?)",
            (record_uuid, str(uuid.uuid4()), resource_path, payload),
        )
        return record_uuid

    def claim_due(self, now: float, limit: int = 32) -> list[sqlite3.Row]:
        """Rows that are PENDING and whose backoff window has elapsed."""
        self._conn.row_factory = sqlite3.Row
        return self._conn.execute(
            "SELECT * FROM outbox WHERE state='PENDING' AND next_attempt_at<=?"
            " ORDER BY next_attempt_at LIMIT ?",
            (now, limit),
        ).fetchall()

    def mark_sent(self, record_uuid: str, etag: str | None) -> None:
        self._conn.execute(
            "UPDATE outbox SET state='SENT', etag=?, last_error=NULL"
            " WHERE record_uuid=?",
            (etag, record_uuid),
        )

    def reschedule(self, record_uuid: str, attempts: int,
                   next_at: float, error: str, etag: str | None = None) -> None:
        self._conn.execute(
            "UPDATE outbox SET attempts=?, next_attempt_at=?, last_error=?,"
            " etag=COALESCE(?, etag) WHERE record_uuid=?",
            (attempts, next_at, error[:500], etag, record_uuid),
        )

    def mark_dead(self, record_uuid: str, error: str) -> None:
        self._conn.execute(
            "UPDATE outbox SET state='DEAD', last_error=? WHERE record_uuid=?",
            (error[:500], record_uuid),
        )

    def backlog(self) -> int:
        return self._conn.execute(
            "SELECT COUNT(*) FROM outbox WHERE state='PENDING'"
        ).fetchone()[0]


class LimsClient:
    """Issues one idempotent conditional upsert and classifies the response."""

    def __init__(self, base_url: str, token_provider, timeout: float = 10.0) -> None:
        self._base = base_url.rstrip("/")
        self._token_provider = token_provider  # callable -> fresh bearer token
        self._timeout = timeout
        self._session = requests.Session()

    def upsert_record(self, row: sqlite3.Row) -> tuple[Outcome, str | None, str]:
        """PUT the record idempotently; return (outcome, new_etag, detail)."""
        headers = {
            "Authorization": f"Bearer {self._token_provider()}",
            "Content-Type": "application/json",
            "Idempotency-Key": row["idempotency_key"],
            "X-Envelope-Version": "2",
        }
        if row["etag"]:                       # optimistic concurrency guard
            headers["If-Match"] = row["etag"]
        url = f"{self._base}{row['resource_path']}/{row['record_uuid']}"
        try:
            resp = self._session.put(
                url, data=row["payload"], headers=headers, timeout=self._timeout,
            )
        except requests.RequestException as exc:
            # No response ever arrived: safe to retry, PUT+key make it idempotent.
            return Outcome.RETRY, None, f"transport: {exc}"

        if resp.status_code < 300:
            return Outcome.SUCCESS, resp.headers.get("ETag"), str(resp.status_code)
        if resp.status_code == 409:
            return Outcome.CONFLICT, resp.headers.get("ETag"), "etag mismatch"
        if resp.status_code in (401, 429) or resp.status_code >= 500:
            return Outcome.RETRY, None, f"transient {resp.status_code}"
        return Outcome.QUARANTINE, None, f"permanent {resp.status_code}: {resp.text[:200]}"


def drain_once(outbox: LimsOutbox, client: LimsClient,
               policy: BackoffPolicy = BackoffPolicy()) -> None:
    """Process every due row exactly once through classification + backoff."""
    now = time.time()
    for row in outbox.claim_due(now):
        outcome, etag, detail = client.upsert_record(row)
        if outcome is Outcome.SUCCESS:
            outbox.mark_sent(row["record_uuid"], etag)
            continue
        attempts = row["attempts"] + 1
        if outcome is Outcome.QUARANTINE or attempts >= policy.max_attempts:
            outbox.mark_dead(row["record_uuid"], detail)
            continue
        # CONFLICT stores the refetched ETag so the retry sends a fresh If-Match.
        next_at = now + policy.delay_for(attempts)
        outbox.reschedule(row["record_uuid"], attempts, next_at, detail, etag)

Two decisions carry the correctness. The upsert_record transport-exception branch returns RETRY, not failure, precisely because a dropped response is the lost-200 case — retrying is safe only because the PUT+Idempotency-Key combination guarantees the server collapses the duplicate. And a CONFLICT reschedule stores the ETag the server returned on the 409, so the next attempt sends the current version in If-Match and stops fighting a stale one.

Validating Idempotent Replay Against the Outbox Table

Verify two things: that a lost response never produces a duplicate, and that the outbox reaches a terminal state for every record. The first is a server-side replay test — send the same UUID and idempotency key twice and assert the LIMS created exactly one row. The second is a table invariant: after the worker drains, no row may sit in PENDING unless its next_attempt_at is in the future.

def test_idempotent_replay_creates_one_record(outbox, client, lims_probe):
    uuid_ = outbox.enqueue("/records", b'{"sample_id":"S-1","value":42}')

    drain_once(outbox, client)          # first delivery: server commits
    row = _row(outbox, uuid_)
    assert row["state"] == "SENT"

    # Simulate the lost-200: re-arm the row and drain again with the SAME key.
    outbox._conn.execute(
        "UPDATE outbox SET state='PENDING', next_attempt_at=0 WHERE record_uuid=?",
        (uuid_,))
    drain_once(outbox, client)

    assert lims_probe.count("/records", uuid_) == 1   # replayed, not duplicated
    assert outbox.backlog() == 0                       # no rows stuck PENDING

On a live LIMS, confirm the same property by inspecting the outbox directly: SELECT state, attempts, last_error FROM outbox after a batch should show SENT rows with attempts reflecting real transient retries, DEAD rows carrying a permanent 4xx in last_error, and no PENDING row older than one backoff window. A row’s attempts climbing to max_attempts with a 5xx in last_error is the signal that the LIMS is genuinely down, not that the record is bad — exactly the distinction the drain worker preserves by dead-lettering 4xx but retrying 5xx.

Failure Modes Specific to REST Metadata Sync

Duplicate submission after a lost 200. The LIMS commits the write, the response is lost to a proxy timeout, the worker retries, and without idempotency you now have two sample records. The signature is duplicate rows in the LIMS sharing every field but the primary key. The fix is structural, not defensive: the client-generated UUID makes the retry target the same resource, and the Idempotency-Key makes the server replay its stored result. If duplicates still appear, the LIMS is not honouring the key — verify with the replay test above before trusting the endpoint.

ETag mismatch under concurrent writers. Two acquisition hosts (or a host and a manual LIMS edit) touch the same record; the second write arrives with a stale If-Match and the LIMS returns 409. The tell is a row whose attempts climbs while last_error stays etag mismatch. Retrying the same body with the same stale ETag loops forever — that is why the worker stores the server’s returned ETag on the 409 and refetches before retrying. If a record genuinely has two authoritative writers, the 409 is telling you to merge at the application layer, not to keep overwriting.

Unbounded outbox growth during a LIMS outage. While the LIMS is unreachable, every attempt classifies as RETRY, PENDING count rises, and the SQLite file grows at λ records per second. If μ has no headroom over λ, the backlog λD never drains after recovery and the disk fills. Watch outbox.backlog() as a first-class metric and wire it to Threshold Tuning & Alerting; a backlog rising past what μ − λ can clear inside your recovery window is the early warning, long before the disk does.

Bearer token expiry mid-batch. A long drain outlives the token TTL, and every remaining record starts returning 401. Because 401 is classified as RETRY, not QUARANTINE, the records are not lost — but a token provider that returns the same expired token on every call turns the whole batch into a slow march to the max_attempts dead-letter. The token_provider callable must refresh on expiry (cache with a TTL and re-mint), so the next attempt in the same drain carries a live token. Diagnose by correlating a wave of 401s in last_error with the token lifetime; a clustered onset at a round interval is expiry, not authorization failure.

← Back to Metadata Injection Workflows

References