Build a Zero-Data-Loss Edge Buffer with SQLite and Replay

This is the concrete build for the contract described in Edge Buffering and Store-and-Forward: a gateway process that never drops a reading, whether the broker is one hop away or unreachable for six hours. The scenario is specific — a single-board industrial gateway polling a handful of PLCs, with no guarantee of continuous WAN connectivity, that must feed an OEE pipeline where a missing interval reads as lost production rather than as a known, recoverable gap. Every step below produces working code you can run against a real SQLite file; none of it is pseudocode, and the ordering of the steps matters — skipping the pragma tuning in step 1 or the acknowledgment ordering in step 4 reintroduces exactly the data-loss window this recipe exists to close.

Record state flow: enqueue, spool, replay, idempotent write, acknowledge A linear state diagram. Enqueue durably writes a record into the spool. The spool holds it until a replay worker selects it. The replay worker performs an idempotent upstream write. On success, an acknowledgment deletes the row from the spool. On failure, the record returns to the spool and is retried after a backoff delay instead of being lost. 1. Enqueue fsync before ack 2. Spool WAL-mode SQLite 3. Replay worker ordered by seq 4. Upstream write idempotent key 5. Acknowledge delete row write failed → backoff → retry, row stays spooled durability boundary is step 1; the recovery boundary is step 5

1. Schema and WAL pragmas

The spool is a single SQLite table opened with write-ahead logging and full synchronous durability. These two pragmas are the entire durability guarantee, so get them right before writing any application code.

import sqlite3
from pathlib import Path


def open_spool(db_path: Path) -> sqlite3.Connection:
    """Open (or create) the durable spool database with production pragmas."""
    conn = sqlite3.connect(db_path, isolation_level=None, check_same_thread=False)
    conn.execute("PRAGMA journal_mode=WAL")      # concurrent readers, single writer
    conn.execute("PRAGMA synchronous=FULL")      # fsync every commit; survives OS crash
    conn.execute("PRAGMA busy_timeout=5000")     # block up to 5s under lock contention
    conn.execute("PRAGMA wal_autocheckpoint=1000")  # checkpoint every ~1000 WAL pages
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS spool (
            seq         INTEGER PRIMARY KEY AUTOINCREMENT,
            asset_id    TEXT NOT NULL,
            metric_name TEXT NOT NULL,
            device_ts   REAL NOT NULL,
            value       REAL NOT NULL,
            enqueued_at REAL NOT NULL
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_spool_asset_ts ON spool(asset_id, device_ts)"
    )
    return conn

journal_mode=WAL lets the enqueue path keep appending while a background checkpoint flushes older pages, so writers are not blocked by SQLite’s own housekeeping. synchronous=FULL is the line that actually buys durability: it forces an fsync before a transaction is reported committed, so a power loss immediately after enqueue() returns still leaves the record on disk. synchronous=NORMAL is roughly twice as fast on spinning media and noticeably faster on flash, but in WAL mode it can lose the most recent commits after an operating-system crash (not after a mere process crash) — acceptable for some fleets, not for one where a lost commit means an unrecoverable production count.

2. Atomic enqueue on receive

Every record — not just ones generated during an outage — passes through the same enqueue call. Treating the durable write as the front door, rather than as a fallback bolted onto a happy-path publish, is what prevents a code path from existing where a record can be acknowledged to the PLC-facing side without first being durable.

import asyncio
import time
from dataclasses import dataclass


@dataclass(slots=True)
class Reading:
    asset_id: str
    metric_name: str
    device_ts: float
    value: float


class DurableEnqueue:
    def __init__(self, conn: sqlite3.Connection):
        self._conn = conn
        self._lock = asyncio.Lock()

    async def enqueue(self, r: Reading) -> None:
        """Blocks until the record is fsync'd; never buffers only in memory."""
        async with self._lock:
            await asyncio.to_thread(
                self._conn.execute,
                "INSERT INTO spool (asset_id, metric_name, device_ts, value, enqueued_at) "
                "VALUES (?, ?, ?, ?, ?)",
                (r.asset_id, r.metric_name, r.device_ts, r.value, time.time()),
            )

    async def enqueue_batch(self, readings: list[Reading]) -> None:
        """Batch several readings into one transaction to amortize fsync cost."""
        if not readings:
            return
        rows = [(r.asset_id, r.metric_name, r.device_ts, r.value, time.time())
                for r in readings]
        async with self._lock:
            await asyncio.to_thread(self._executemany, rows)

    def _executemany(self, rows: list[tuple]) -> None:
        self._conn.executemany(
            "INSERT INTO spool (asset_id, metric_name, device_ts, value, enqueued_at) "
            "VALUES (?, ?, ?, ?, ?)",
            rows,
        )

Use enqueue_batch whenever the polling loop naturally produces several readings per tick (a multi-tag Modbus scan, for example): wrapping ten inserts in one transaction pays one fsync instead of ten, without weakening the guarantee, because SQLite commits the whole batch atomically or not at all.

3. Async replay worker with acked deletion

The replay worker runs concurrently with enqueue, continuously draining the oldest unacknowledged rows in insertion order. It never deletes a row until the upstream call has confirmed the write, which is the property that makes a crash mid-replay safe: on restart, the worker simply resumes from the same unacknowledged rows and the idempotent upstream write absorbs the duplicate.

import logging

logger = logging.getLogger("replay_worker")


class ReplayWorker:
    def __init__(self, conn: sqlite3.Connection, publish_upstream, *,
                 batch_size: int = 500, idle_poll: float = 1.0):
        self._conn = conn
        self._publish = publish_upstream
        self._batch_size = batch_size
        self._idle_poll = idle_poll
        self._lock = asyncio.Lock()

    async def _fetch_batch(self) -> list[tuple]:
        async with self._lock:
            cur = await asyncio.to_thread(
                self._conn.execute,
                "SELECT seq, asset_id, metric_name, device_ts, value FROM spool "
                "ORDER BY seq ASC LIMIT ?",
                (self._batch_size,),
            )
            return cur.fetchall()

    async def _ack(self, seqs: list[int]) -> None:
        placeholders = ",".join("?" * len(seqs))
        async with self._lock:
            await asyncio.to_thread(
                self._conn.execute, f"DELETE FROM spool WHERE seq IN ({placeholders})", seqs
            )

    async def run(self) -> None:
        backoff = 1.0
        while True:
            rows = await self._fetch_batch()
            if not rows:
                await asyncio.sleep(self._idle_poll)
                continue
            payload = [
                {"asset_id": a, "metric_name": m, "device_ts": t, "value": v}
                for _, a, m, t, v in rows
            ]
            try:
                await self._publish(payload)
                await self._ack([seq for seq, *_ in rows])
                backoff = 1.0
            except Exception:
                logger.exception("replay publish failed for %d rows; backing off", len(rows))
                await asyncio.sleep(min(backoff, 30.0))
                backoff *= 2

Notice the worker never branches on “are we in an outage right now.” It behaves identically whether the spool has one row waiting from a momentary hiccup or fifty thousand rows from an overnight WAN failure — the batching and backoff logic is the only thing standing between a normal catch-up flush and a replay storm that overwhelms the broker.

4. Idempotent upstream write + backoff

The upstream publish function is the other half of the safety property. Because replay can race live ingestion — a reconnect can deliver a burst of spooled records at nearly the same moment the PLC resumes live publishing — the write must key on the full identity of a reading and treat a duplicate as a successful no-op rather than an error.

import aiohttp


async def publish_upstream(records: list[dict], session: aiohttp.ClientSession) -> None:
    """Idempotent upstream write: composite key dedupes replay/live overlap."""
    async with session.post(
        "https://tsdb.internal/api/v1/ingest",
        json={
            "records": records,
            "conflict_strategy": "ignore",   # server-side ON CONFLICT DO NOTHING
            "key_fields": ["asset_id", "metric_name", "device_ts"],
        },
        timeout=aiohttp.ClientTimeout(total=15),
    ) as resp:
        if resp.status == 429:
            raise RuntimeError("upstream rate-limited; caller applies backoff")
        if resp.status >= 500:
            raise RuntimeError(f"upstream server error: {resp.status}")
        if resp.status not in (200, 201, 204):
            raise RuntimeError(f"unexpected upstream status: {resp.status}")

On the receiving side, the endpoint (or the TSDB itself) must implement the corresponding INSERT ... ON CONFLICT (asset_id, metric_name, device_ts) DO NOTHING — this recipe’s replay worker cannot make the write safe by itself; it can only guarantee that it tries the write until the write itself reports success. An ambiguous failure (timeout after the request left the gateway) is treated as a failure and retried, which is safe precisely because retrying an idempotent write is a no-op rather than a duplicate insert.

5. Recovery and reconciliation

Recovery is not a separate code path — it is what the replay worker does continuously — but it does need a verification step after any real outage so an operator can confirm nothing was actually lost. Query the spool depth and oldest row age as a live health signal, and reconcile counts against the upstream store once the spool is empty.

-- Health check: run on every gateway heartbeat
SELECT
    COUNT(*)                                   AS unacked_rows,
    MIN(enqueued_at)                           AS oldest_unacked_at,
    (strftime('%s','now') - MIN(enqueued_at))  AS oldest_age_seconds
FROM spool;

async def reconcile_after_outage(spool_conn: sqlite3.Connection, tsdb_count_fn,
                                  asset_id: str, window_start: float, window_end: float) -> bool:
    """Confirm zero net loss for an asset over the outage window."""
    cur = spool_conn.execute(
        "SELECT COUNT(*) FROM spool WHERE asset_id = ? AND device_ts BETWEEN ? AND ?",
        (asset_id, window_start, window_end),
    )
    still_spooled = cur.fetchone()[0]
    if still_spooled:
        return False   # replay is not finished yet; reconcile again later

    upstream_count = await tsdb_count_fn(asset_id, window_start, window_end)
    expected_count = ...  # from the emulator log or PLC scan-cycle expectation
    return upstream_count >= expected_count

Run this reconciliation as a scheduled job after any outage exceeding a configured threshold (for example, five minutes), and alert if it does not converge to True within a bounded time after the spool reports zero unacknowledged rows.

Gotchas & anti-patterns

  • Acking before the upstream write confirms. The single most common mistake in a hand-rolled buffer: deleting the spooled row as soon as it is handed to a publish call, rather than after that call returns success. This reintroduces silent loss on every crash or timeout during publish.
  • One INSERT transaction per reading at high sample rates. At more than a few hundred readings per second, per-row fsync cost dominates. Batch into enqueue_batch transactions sized to the actual burst pattern.
  • Replaying without a batch ceiling. Draining the entire backlog in a single unbounded call after a long outage can exceed request timeouts and, worse, starve concurrently arriving live telemetry. Always bound batch_size and interleave with bounded queue backpressure.
  • Trusting device time to order replay. A gateway clock can jump during an outage. Order replay by the spool’s own monotonic seq, not by device_ts, and let downstream correction reconcile true time afterward.
  • No ceiling on spool growth. A spool with no size limit will eventually fill the gateway’s flash. Pair this recipe with an explicit sizing policy — see ring buffer sizing and disk-spill thresholds — before deploying to a fleet.

Quick reference

Pragma / setting Purpose Recommended value
journal_mode Concurrent read/write without blocking the writer WAL
synchronous Durability across power loss vs. OS crash FULL (durability-first)
busy_timeout Wait instead of erroring under lock contention 5000 ms
wal_autocheckpoint Bound WAL file growth during heavy replay 1000 pages
Idempotent key Dedupe replay/live overlap at the upstream write (asset_id, metric_name, device_ts)
Ack ordering Never lose a row on crash mid-replay delete only after upstream confirms
Replay batch size Bound replay-storm impact on the broker 500 rows (tune to timeout budget)

Related in this section