Fallback sync architecture for disconnected POS systems

A crash-resilient offline buffer for pharmacy POS systems that prevents duplicate Schedule II dispensing and audit drift when a disconnected terminal reconnects and replays its queue.

The dangerous moment for a disconnected pharmacy point-of-sale (POS) terminal is not the outage itself — it is the reconnect. When the network returns and a buffered queue replays upstream, the two failure modes that produce DEA findings are duplicate dispensing posts (the same Schedule II decrement applied twice because an acknowledgment was lost mid-flight) and offline audit drift (locally-captured events that arrive out of order, with a clock skew, or with a payload that no longer hashes to its original value). Either one fractures the perpetual-inventory count and breaks the chain of custody that 21 CFR § 1304.21 requires. This page solves exactly that problem: a crash-resilient, idempotent offline buffer whose replay is provably exactly-once. It operates within the parent topic, Fallback Routing for Offline Sync, and the broader Core Architecture & DEA Compliance Frameworks.

Prerequisites & environment

Python 3.11+ — the buffer uses dataclasses(frozen=True, slots=True) so a captured transaction cannot be mutated between enqueue and replay.
Standard library for the durable core: sqlite3 (in WAL mode), hashlib, hmac, json, uuid, logging. The only third-party dependency is requests (with urllib3.util.retry.Retry) for the upstream POST.
A local disk the POS process can fsync to. Network-mounted or ephemeral container storage defeats crash resilience — the buffer must survive a hard power loss.
Regulatory context you should already hold: 21 CFR § 1304.21 (recordkeeping for dispensing), 21 CFR § 1304.04 (record retention), 21 CFR § 1311.30 (electronic record integrity), and HIPAA § 164.312(b) (audit controls). Logs must record that a dispense occurred and its NDC and quantity, but must never embed patient PHI.
The idempotency key is generated at the terminal, before the outage, and is stable across every retry. Generating it at send time — or per attempt — reintroduces the duplicate-post bug this page exists to kill.

Before a payload is buffered, its NDC is normalized to the canonical 11-digit form defined in NDC-11 vs NDC-10 Parsing Standards, and its schedule is resolved by the DEA Schedule II–V Classification Mapping engine. Normalizing after the outage is unsafe: the FDA reference table may have refreshed while the terminal was offline, silently shifting a code’s canonical form and breaking idempotency.

Disconnect detection & the replay trigger

Relying on HTTP status codes alone is insufficient where intermittent packet loss, DNS failures, or TLS renegotiation can corrupt a partial payload. Use a layered probe and trip into OFFLINE_ROUTING deterministically. The following diagnostic query isolates the highest-risk controlled-substance backlog before a replay so Schedule II records can be reconciled first:

sql

SELECT
  COUNT(*)                                            AS total_pending,
  MIN(created_at_utc)                                 AS oldest_pending,
  SUM(CASE WHEN schedule = 'II' THEN 1 ELSE 0 END)    AS c2_pending,
  MAX(retry_count)                                    AS worst_retry
FROM offline_queue
WHERE sync_status = 'PENDING';

Trip into fallback mode when the oldest pending transaction exceeds 120 seconds, or when local queue depth approaches its storage ceiling — uncontrolled queue growth during a prolonged outage is itself an audit-boundary risk.

Implementation: the idempotent offline buffer

The buffer is a SQLite-backed FIFO with three integrity guarantees: a UNIQUE idempotency key (collision-proof at the storage layer), a SHA-256 hash of the canonical payload (tamper-evidence per 21 CFR § 1311.30), and a monotonic id (FIFO replay order). WAL mode plus synchronous=NORMAL keeps writes durable across a crash without blocking the dispensing thread.

python

import hashlib
import hmac
import json
import logging
import sqlite3
import uuid
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Dict, List, Optional

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# HIPAA-compliant audit logger — records that an event occurred, never patient PHI.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[logging.FileHandler("/var/log/pharmacy/sync_audit.log")],
)
logger = logging.getLogger("deafallback.sync_engine")


@dataclass(frozen=True, slots=True)
class TransactionPayload:
    """Immutable once captured — the idempotency_key is assigned at the terminal,
    before the outage, and is stable across every replay attempt."""
    ndc: str               # canonical 11-digit NDC, normalized pre-buffer
    quantity: int
    schedule: str          # 'II'..'V', resolved pre-buffer
    prescription_id: str
    pharmacist_npi: str
    dispensed_at_utc: str  # ISO-8601 UTC, set at dispense time, not send time
    idempotency_key: str


class OfflineTransactionBuffer:
    """Crash-resilient FIFO queue with cryptographic integrity (WAL mode)."""

    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path, timeout=30)
        self.conn.execute("PRAGMA journal_mode=WAL;")
        self.conn.execute("PRAGMA synchronous=NORMAL;")
        self._init_schema()

    def _init_schema(self) -> None:
        self.conn.executescript(
            """
            CREATE TABLE IF NOT EXISTS offline_queue (
                id              INTEGER PRIMARY KEY AUTOINCREMENT,
                idempotency_key TEXT UNIQUE NOT NULL,
                payload_hash    TEXT NOT NULL,
                payload_json    TEXT NOT NULL,
                schedule        TEXT NOT NULL,
                sync_status     TEXT NOT NULL DEFAULT 'PENDING',
                created_at_utc  TEXT NOT NULL,
                retry_count     INTEGER NOT NULL DEFAULT 0
            );
            CREATE INDEX IF NOT EXISTS idx_status ON offline_queue(sync_status);
            """
        )
        self.conn.commit()

    @staticmethod
    def _canonical(payload: TransactionPayload) -> str:
        # Deterministic serialization — sorted keys, no whitespace — so the hash
        # is reproducible at replay time for the integrity check.
        return json.dumps(asdict(payload), sort_keys=True, separators=(",", ":"))

    def enqueue(self, payload: TransactionPayload) -> bool:
        """Persist a transaction. Returns False on idempotency collision (already
        buffered) — a duplicate scan or a retried local capture is absorbed here."""
        canonical = self._canonical(payload)
        payload_hash = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
        try:
            self.conn.execute(
                """INSERT INTO offline_queue
                   (idempotency_key, payload_hash, payload_json, schedule, created_at_utc)
                   VALUES (?, ?, ?, ?, ?)""",
                (payload.idempotency_key, payload_hash, canonical,
                 payload.schedule, payload.dispensed_at_utc),
            )
            self.conn.commit()
            logger.info("enqueued schedule=%s key=%s", payload.schedule, payload.idempotency_key)
            return True
        except sqlite3.IntegrityError:
            logger.info("dedup_local key=%s already buffered", payload.idempotency_key)
            return False

    def fetch_pending(self, limit: int = 50) -> List[Dict]:
        # Schedule II first, then strict FIFO — C2 backlog reconciles ahead of C3-V.
        cur = self.conn.execute(
            """SELECT id, idempotency_key, payload_hash, payload_json, schedule, retry_count
               FROM offline_queue WHERE sync_status = 'PENDING'
               ORDER BY (schedule = 'II') DESC, id ASC LIMIT ?""",
            (limit,),
        )
        cols = ["id", "idempotency_key", "payload_hash", "payload_json", "schedule", "retry_count"]
        return [dict(zip(cols, row)) for row in cur.fetchall()]

    def mark(self, record_id: int, status: str) -> None:
        self.conn.execute(
            "UPDATE offline_queue SET sync_status = ? WHERE id = ?", (status, record_id)
        )
        self.conn.commit()

    def bump_retry(self, record_id: int) -> None:
        self.conn.execute(
            "UPDATE offline_queue SET retry_count = retry_count + 1 WHERE id = ?", (record_id,)
        )
        self.conn.commit()


class SyncEngine:
    """Exactly-once replay: integrity check, then idempotent upstream POST."""

    def __init__(self, api_base_url: str, auth_token: str, buffer: OfflineTransactionBuffer):
        self.api_base = api_base_url.rstrip("/")
        self.buffer = buffer
        self.session = requests.Session()
        self.session.headers.update(
            {"Authorization": f"Bearer {auth_token}", "Content-Type": "application/json"}
        )
        # Exponential back-off with jitter so a reconnecting fleet does not stampede
        # the central pharmacy management system (PMS).
        retry = Retry(
            total=5, backoff_factor=1.5, backoff_jitter=0.5,
            status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["POST"],
        )
        self.session.mount("https://", HTTPAdapter(max_retries=retry))

    def _verify_integrity(self, payload_json: str, expected_hash: str) -> bool:
        actual = hashlib.sha256(payload_json.encode("utf-8")).hexdigest()
        return hmac.compare_digest(actual, expected_hash)  # constant-time

    def sync_batch(self) -> Dict[str, int]:
        result = {"synced": 0, "quarantined": 0, "deferred": 0}
        for rec in self.buffer.fetch_pending():
            # 1) Tamper-evidence: a record that no longer hashes to its stored value
            #    never reaches the ledger (21 CFR § 1311.30).
            if not self._verify_integrity(rec["payload_json"], rec["payload_hash"]):
                self.buffer.mark(rec["id"], "QUARANTINED")
                result["quarantined"] += 1
                logger.error("hash_mismatch key=%s quarantined", rec["idempotency_key"])
                continue
            try:
                # 2) Idempotency key in the header lets the PMS reject a replayed
                #    post whose ack was lost — the duplicate-dispense guard.
                resp = self.session.post(
                    f"{self.api_base}/api/v1/transactions/sync",
                    data=rec["payload_json"],
                    headers={"X-Idempotency-Key": rec["idempotency_key"]},
                    timeout=15,
                )
                # 200 = newly committed, 409 = already seen upstream. BOTH are success.
                if resp.status_code in (200, 201, 409):
                    self.buffer.mark(rec["id"], "SYNCED")
                    result["synced"] += 1
                    if rec["schedule"] == "II":
                        logger.info("c2_reconciled key=%s code=%s",
                                    rec["idempotency_key"], resp.status_code)
                else:
                    resp.raise_for_status()
            except requests.exceptions.RequestException as exc:
                self.buffer.bump_retry(rec["id"])
                result["deferred"] += 1
                logger.warning("deferred key=%s err=%s", rec["idempotency_key"], exc)
        logger.info("batch_done %s", result)
        return result

The exactly-once property comes from one design decision: HTTP 409 from the PMS is treated as success, not failure. If a post commits upstream but the acknowledgment is lost to the network, the next replay re-sends the same X-Idempotency-Key, the PMS recognizes it, returns 409 Conflict, and the buffer marks the record SYNCED instead of re-queuing it. The decrement is applied once and exactly once.

Verification & testing

Prove the no-duplicate guarantee directly: enqueue a transaction, simulate an acknowledged-but-lost reply on the first attempt and a 409 on the replay, and assert the record ends SYNCED with no second decrement.

python

def test_lost_ack_does_not_double_post():
    buf = OfflineTransactionBuffer(":memory:")
    tx = TransactionPayload(
        ndc="00093-0058-01", quantity=30, schedule="II",
        prescription_id="RX-88231", pharmacist_npi="1972648394",
        dispensed_at_utc="2026-06-28T14:03:11Z",
        idempotency_key="term07-9f3c1a2b-0001",
    )
    assert buf.enqueue(tx) is True
    assert buf.enqueue(tx) is False          # local dedup — same key absorbed
    pending = buf.fetch_pending()
    assert len(pending) == 1                  # one record, not two

    # Integrity must hold for the untouched payload.
    rec = pending[0]
    eng = SyncEngine("https://pms.example", "token", buf)
    assert eng._verify_integrity(rec["payload_json"], rec["payload_hash"]) is True

    # Tamper one byte → integrity fails → would be quarantined, not posted.
    assert eng._verify_integrity(rec["payload_json"] + " ", rec["payload_hash"]) is False

A clean replay should emit an audit line you can grep during a DEA inspection — that the C2 event reconciled, with its idempotency key, and never the patient identity:

text

2026-06-28 14:09:02 | INFO | deafallback.sync_engine | c2_reconciled key=term07-9f3c1a2b-0001 code=409
2026-06-28 14:09:02 | INFO | deafallback.sync_engine | batch_done {'synced': 1, 'quarantined': 0, 'deferred': 0}

The code=409 here is the proof: the record had already committed during a prior attempt, the replay was absorbed, and no second Schedule II decrement was written.

Gotchas & compliance pitfalls

Per-attempt idempotency keys. Generating the key inside the retry loop (or with uuid4() at send time) defeats the entire mechanism — every replay looks new to the PMS and double-posts. Mint it once, at the terminal, and freeze it in the payload.
Re-normalizing the NDC after the outage. If the FDA reference table refreshed while the terminal was offline, late normalization can change the canonical form, change the hash, and trip a false quarantine. Normalize and classify before enqueue; see the rules in NDC-11 vs NDC-10 Parsing Standards.
Treating 409 as an error. A 409 is the success signal for a lost-ack replay. Routing it to raise_for_status() re-queues a record that already committed, producing the exact duplicate this design prevents.
json.dumps non-determinism. Without sort_keys=True and fixed separators, two serializations of the same object differ, the replay hash will not match, and clean records get quarantined. Hash the canonical form only.
Clock skew on dispensed_at_utc. Use a UTC timestamp captured at dispense time on the terminal, not at replay time. A terminal whose clock drifted offline must reconcile against the PMS clock during RECONCILING, or the ledger ordering in your audit-boundary filter will disagree with wall-clock reality.
Silent quarantine. A QUARANTINED record is a recordkeeping gap until a human resolves it. Wire quarantine count into the same alerting path your retry and error-handling logic uses — a non-zero quarantine total is a 21 CFR § 1304.04 exposure, not a metric to log and forget.

Frequently Asked Questions

How is “exactly-once” guaranteed if the network can drop an acknowledgment?

It is guaranteed at two layers. The local UNIQUE constraint on idempotency_key absorbs duplicate captures before they ever leave the terminal, and the stable key sent in X-Idempotency-Key lets the PMS recognize a replayed post and answer 409. Because the buffer marks both 200 and 409 as SYNCED, a lost acknowledgment costs one redundant request, never a second inventory decrement.

Should Schedule II transactions replay before Schedule III–V?

Yes. fetch_pending orders schedule = 'II' first so the highest-diversion-risk backlog reconciles ahead of lower schedules, shrinking the window in which a C2 dispense exists only on local disk. The classification that assigns each record its schedule is owned by the DEA Schedule II–V Classification Mapping engine.

What happens to a record that fails its hash check on replay?

It is moved to QUARANTINED and never posted, satisfying the electronic-record integrity expectation of 21 CFR § 1311.30. Quarantine is a tamper-evidence signal, so it must raise an operator alert rather than being silently retried.

Fallback Routing for Offline Sync — parent topic: network-state detection, the routing state machine, and queue boundaries.
DEA Schedule II–V Classification Mapping — how each buffered record’s schedule is resolved before enqueue.
NDC-11 vs NDC-10 Parsing Standards — canonical NDC normalization that keeps the replay hash stable.
Defining audit boundaries for controlled substances — which reconciled events enter the immutable ledger.
Error handling & retry mechanisms — back-off, jitter, and quarantine alerting for the wider ingestion pipeline.