Edge Buffering and Store-and-Forward for Zero-Data-Loss Ingestion

Edge buffering is the durability layer of Ingestion & Cleaning Workflows that keeps telemetry alive when the path from a gateway to the broker or time-series database breaks. Plant networks partition routinely — a switch reboots during a firmware push, a cellular backhaul drops during a storm, a TSDB cluster fails over — and every one of those events is invisible to the PLC still emitting a reading every 100 ms. Without a store-and-forward contract, that gap becomes a silent hole in the Availability denominator; with one, an outage becomes a bounded, replayable delay that a downstream engineer can distinguish from a real stoppage. This page defines the contract; the two recipes under it cover the concrete SQLite implementation and the sizing math for how much of an outage a gateway can absorb before it must shed data.

The core failure mode edge buffering exists to prevent is quiet loss: a gateway that publishes over MQTT with no local durability drops every message generated while the broker connection is down, and nothing downstream ever learns that those readings existed. A correctly designed buffer instead treats “broker unreachable” as an expected state, not an exception, and degrades through three phases — normal forwarding, durable local spill, and ordered replay — without operator intervention.

Three phases of edge store-and-forward: normal path, outage spill, and recovery replay In the normal path, a PLC feeds an edge process that publishes directly through the broker to the time-series database. In the outage path, the broker link is severed, so the edge process writes every record into a durable local spool instead of the network, with an X marking the broken link. In the recovery path, once the link returns, a replay worker drains the spool in order, performs an idempotent upstream write, and only then deletes the acknowledged row from the spool, closing the loop with zero data loss. 1 · NORMAL PATH — LINK UP PLC / sensor 100–250 ms scan Edge process publish direct MQTT broker QoS 1 Time-series DB continuous write 2 · OUTAGE PATH — LINK DOWN, SPILL TO DISK link severed PLC / sensor unaware of outage Edge process publish fails Durable local spool SQLite WAL · append-only every record persisted 3 · RECOVERY PATH — REPLAY ON RECONNECT Replay worker ordered drain Idempotent write broker → TSDB ack → delete row

Buffering strategies at a glance

Not every gateway needs the same durability budget. The table below maps the common edge-buffer implementations to how much of an outage they survive and what they cost in flash wear and code complexity.

Strategy Durability across power loss Ordering guarantee Typical outage covered Replay complexity
In-memory ring buffer none — lost on power cycle FIFO within process lifetime seconds to low minutes trivial (drain queue)
On-disk WAL (SQLite) full — fsync’d before ack strict, via monotonic row id hours to days moderate (cursor + ack/delete)
File-per-batch spool (rotating logs) full, but batch-grained coarse — batch order only hours to days moderate, but partial-batch replay is awkward
Broker-side persistent queue mirror depends on broker durability strict, broker-assigned offset until broker disk fills low code, but requires broker reachability to start

Most industrial gateways land on the second row: an embedded, transactional store that survives a power cycle without requiring a second network-attached service. That is the design this section develops in depth, and the concrete build is in Build a Zero-Data-Loss Edge Buffer with SQLite and Replay.

Core concept and design contract

Store-and-forward is not “cache the message and try again later” — that description omits every property that makes it safe under real outage patterns. The contract has four non-negotiable rules.

  • Durable append before acknowledgment. A record is only considered ingested once it has been fsync’d to the local spool, not once it has been handed to an in-memory queue. An edge process that acks a PLC read before the write is durable will lose that record on a power cut mid-write — the single most common cause of “why does this shift have a two-minute hole” tickets.
  • Ordered replay by source time, not by disk-write time. Because clock-drift correction has usually not yet run this early in the pipeline, the spool preserves the raw device timestamp and a monotonic sequence number so clock drift correction further downstream can still reconstruct a correct order after replay interleaves with live traffic.
  • Idempotent upstream writes. Replay after a long outage frequently overlaps with live ingestion resuming on the same asset, and a QoS 1 MQTT session redelivers on reconnect regardless. The write to the broker or TSDB must be keyed so a duplicate delivery is a no-op, following the same idempotency discipline used throughout async batch processing.
  • Bounded size with explicit shedding. An unbounded spool eventually fills the gateway’s flash and crashes the process that was supposed to protect the data. The buffer must have a hard ceiling, a spill/shed policy once that ceiling is approached, and instrumentation that makes the operator aware shedding occurred — the sizing math for that ceiling is worked out in Ring Buffer Sizing and Disk-Spill Thresholds at the Edge.

Within ISA-95 terms, the buffer sits at the boundary between Level 1/2 (the PLC and its scan cycle) and Level 3 (the MES/historian path), and it is the component that makes that boundary tolerant of a Level 3 outage without corrupting Level 2 production counts. A gateway that cannot buffer durably effectively couples the reliability of the shop floor sensor to the reliability of a WAN link it does not control — an architecture that fails the first time IT patches the core switch.

A fifth, quieter rule follows from the first four: the buffer must be observable while it is doing nothing. A spool that is empty and healthy looks identical, from the outside, to a spool that was silently disabled by a misconfigured path or a permissions error, unless it actively reports its own depth, oldest-unacknowledged-row age, and last-successful-replay timestamp. Gateways deployed to hundreds of sites cannot be SSH’d into individually to confirm the buffer is functioning; the metrics named here are what feed a fleet-wide dashboard that turns “is store-and-forward actually working on line 14” into a query instead of a site visit.

Implementation

The reference implementation below is a minimal but production-honest store-and-forward core: an asyncio writer appends every record to a WAL-mode SQLite database before returning, and a separate replay task drains unacknowledged rows in order, writes them idempotently upstream, and only deletes a row after the upstream write is confirmed. Both tasks run concurrently inside the same gateway process so ingestion never blocks on network state.

import asyncio
import json
import logging
import sqlite3
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any

logger = logging.getLogger("edge_buffer")


def _connect(db_path: Path) -> sqlite3.Connection:
    """Open a WAL-mode connection tuned for durable, concurrent access."""
    conn = sqlite3.connect(db_path, isolation_level=None, check_same_thread=False)
    conn.execute("PRAGMA journal_mode=WAL")     # readers do not block the writer
    conn.execute("PRAGMA synchronous=FULL")     # fsync on every commit — durability over speed
    conn.execute("PRAGMA busy_timeout=5000")    # wait rather than fail under lock contention
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS spool (
            seq INTEGER PRIMARY KEY AUTOINCREMENT,
            asset_id TEXT NOT NULL,
            device_ts REAL NOT NULL,
            payload TEXT NOT NULL,
            enqueued_at REAL NOT NULL
        )
        """
    )
    return conn


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


class EdgeSpool:
    """Durable append-only store-and-forward buffer over WAL-mode SQLite."""

    def __init__(self, db_path: Path):
        self._conn = _connect(db_path)
        self._lock = asyncio.Lock()

    async def enqueue(self, record: TelemetryRecord) -> None:
        """Durably persist a record. Returns only after the write is fsync'd."""
        payload = json.dumps({
            "metric_name": record.metric_name,
            "value": record.value,
        })
        async with self._lock:
            # sqlite3 is synchronous; run the blocking write off the event loop.
            await asyncio.to_thread(
                self._conn.execute,
                "INSERT INTO spool (asset_id, device_ts, payload, enqueued_at) "
                "VALUES (?, ?, ?, ?)",
                (record.asset_id, record.device_ts, payload, time.time()),
            )

    async def replay_batch(self, limit: int = 500) -> list[tuple[int, dict[str, Any]]]:
        """Return the oldest unacknowledged rows, ordered for deterministic replay."""
        async with self._lock:
            rows = await asyncio.to_thread(
                self._conn.execute,
                "SELECT seq, asset_id, device_ts, payload FROM spool "
                "ORDER BY seq ASC LIMIT ?",
                (limit,),
            )
            rows = rows.fetchall()
        return [
            (seq, {"asset_id": asset_id, "device_ts": device_ts, **json.loads(payload)})
            for seq, asset_id, device_ts, payload in rows
        ]

    async def acknowledge(self, seqs: list[int]) -> None:
        """Delete only rows the upstream write has confirmed, never before."""
        if not seqs:
            return
        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 depth(self) -> int:
        async with self._lock:
            cur = await asyncio.to_thread(self._conn.execute, "SELECT COUNT(*) FROM spool")
            return cur.fetchone()[0]

Every record passes through enqueue before it is considered safely received — including records generated while the broker is perfectly healthy. That uniformity is deliberate: it means the same code path handles the normal case and the outage case, so there is no separate “outage mode” to test and no seam where a code change silently reintroduces a data-loss window.

import random


async def replay_loop(spool: EdgeSpool, publish_upstream, *, poll_interval: float = 1.0) -> None:
    """Continuously drain the spool with exponential backoff on upstream failure."""
    backoff = 1.0
    while True:
        batch = await spool.replay_batch(limit=500)
        if not batch:
            await asyncio.sleep(poll_interval)
            continue
        try:
            # publish_upstream must be idempotent: keyed on (asset_id, device_ts, metric_name)
            await publish_upstream(batch)
            await spool.acknowledge([seq for seq, _ in batch])
            backoff = 1.0                        # reset after a clean flush
        except Exception:
            logger.exception("upstream publish failed; retrying with backoff")
            jitter = random.uniform(0.0, 0.5)
            await asyncio.sleep(min(backoff, 30.0) + jitter)
            backoff *= 2

The upstream publish_upstream callable is expected to write with an idempotent key — typically (asset_id, metric_name, device_ts) on an INSERT ... ON CONFLICT DO NOTHING — because a replay batch and a live-ingestion batch can legitimately overlap once the link returns, and because MQTT QoS 1 already guarantees at-least-once delivery on top of that. A gateway using MQTT QoS 1 for discrete state transitions already has to solve deduplication once; the spool’s replay path reuses exactly that mechanism rather than inventing a second one.

Edge cases and failure modes

Real gateways fail in ways a happy-path implementation does not anticipate. Each of the following has a specific, testable defense.

  • Power loss mid-write. PRAGMA synchronous=FULL guarantees SQLite does not report a commit as complete until the page is on stable storage, so a power cut either loses nothing committed or rolls back an in-flight transaction cleanly on the next open — it never leaves a half-written row. synchronous=NORMAL is faster but, in WAL mode, can lose the most recent commits after an OS crash (though not after a mere application crash); use FULL unless the measured fsync cost is unacceptable for the sample rate.
  • Replay storms after a long outage. A gateway reconnecting after an eight-hour outage may have tens of thousands of queued rows. Draining them all in one burst can overwhelm the broker or trip TSDB rate limiting. Replay in bounded batches (as above) and apply the same backpressure discipline used for live asyncio MQTT consumers so a catch-up flush cannot starve concurrently arriving live telemetry.
  • Disk full. A spool with no ceiling eventually consumes all flash on the gateway, at which point SQLite writes start failing and the enqueue path itself becomes the outage. Enforce a hard row-count or byte-size ceiling and shed the oldest, lowest-priority records once it is reached — worked out precisely in ring buffer sizing and disk-spill thresholds.
  • Clock jump during the outage. If the gateway’s own clock steps (NTP correction, manual reset) while the link is down, rows enqueued before and after the jump can appear out of source-time order even though they are in correct insertion order. Retain both device_ts (raw, uncorrected) and seq (monotonic insertion order); replay by seq for delivery order but let downstream clock drift correction reconcile device_ts afterward — never try to reorder the spool itself by a clock that may be wrong.
  • Partial acknowledgment. If the process crashes between a successful upstream write and the local DELETE, the same rows are replayed again on restart. This is expected and correct as long as the upstream write is idempotent; it is a duplicate-delivery problem, not a data-loss problem, and it must never be “fixed” by acking before the upstream write is confirmed.
  • Schema drift across a firmware update. A gateway firmware upgrade that changes payload shape mid-outage means old spooled rows and new live rows can differ in schema. Version the payload envelope and validate on replay exactly as the normalized telemetry schema does on the live path — reject unparseable spooled rows to a quarantine table rather than crashing the replay loop.

Verification and testing

The only trustworthy proof that store-and-forward works is to actually sever the connection and count records afterward. A unit test proves the local mechanics; a chaos test proves the end-to-end contract.

import pytest


@pytest.mark.asyncio
async def test_spool_replays_all_records_exactly_once(tmp_path):
    spool = EdgeSpool(tmp_path / "spool.db")
    for i in range(50):
        await spool.enqueue(TelemetryRecord("PLC-9F2A", device_ts=1000.0 + i,
                                             value=float(i), metric_name="rpm"))
    assert await spool.depth() == 50

    delivered: list[dict] = []

    async def fake_publish(batch):
        delivered.extend(payload for _, payload in batch)

    batch = await spool.replay_batch(limit=100)
    await fake_publish(batch)
    await spool.acknowledge([seq for seq, _ in batch])

    assert await spool.depth() == 0
    assert len(delivered) == 50
    assert [d["device_ts"] for d in delivered] == sorted(d["device_ts"] for d in delivered)

The chaos test that actually earns confidence runs against a real deployment target: sever the gateway’s network interface for a fixed interval while a synthetic PLC emulator publishes at a known rate, restore connectivity, and reconcile the count that arrived at the TSDB against the count the emulator sent. Schedule this as a recurring drill, not a one-time acceptance test — buffer code that is never exercised under real failure rots silently as dependencies change.

-- Reconcile expected vs received counts for a chaos-test window
SELECT
    expected.asset_id,
    expected.expected_count,
    COALESCE(received.received_count, 0) AS received_count,
    expected.expected_count - COALESCE(received.received_count, 0) AS lost
FROM (
    SELECT asset_id, COUNT(*) AS expected_count
    FROM emulator_send_log
    WHERE sent_at BETWEEN :window_start AND :window_end
    GROUP BY asset_id
) AS expected
LEFT JOIN (
    SELECT asset_id, COUNT(*) AS received_count
    FROM telemetry_raw
    WHERE device_ts BETWEEN :window_start AND :window_end
    GROUP BY asset_id
) AS received ON received.asset_id = expected.asset_id
WHERE expected.expected_count - COALESCE(received.received_count, 0) <> 0;

A zero-row result is the pass condition. Any non-zero lost value means either the spool dropped a durable write or the replay loop terminated before draining fully — both are release blockers, not tuning issues.

Performance and scale considerations

WAL-mode SQLite comfortably sustains several thousand single-row inserts per second on typical industrial gateway flash (eMMC or industrial SD), but synchronous=FULL makes every commit pay an fsync, so throughput is bounded by the storage medium’s fsync latency, not by SQLite itself. Batch multiple sensor readings into a single transaction when the sample rate exceeds a few hundred records per second — wrapping ten inserts in one BEGIN...COMMIT amortizes the fsync cost across all ten without weakening the durability guarantee, since the whole batch either commits or does not.

Write amplification matters on flash-based gateways: WAL mode writes to the WAL file and later checkpoints back into the main database file, roughly doubling total bytes written during heavy replay. Tune wal_autocheckpoint so checkpoints happen on a predictable cadence rather than blocking a large replay burst, and monitor flash wear on gateways expected to run for years unattended — a spool sized for a two-hour outage that instead becomes a permanent high-frequency write target will shorten eMMC lifespan measurably faster than the sensor stream alone would.

Replay throughput after a long outage is typically limited by the upstream broker or TSDB’s acceptance rate, not by the local read from SQLite. Size replay_batch limits so a single batch clears comfortably inside the network round-trip and timeout budget, mirroring the batch-sizing guidance in async batch processing. Persist the corrected, replayed stream into a time-series database keyed on the same idempotent composite key used for live writes, so a replay that races a reconnect never double-counts a shift’s production.

Emit spool_depth_rows, spool_oldest_row_age_seconds, replay_batches_flushed_total, and replay_publish_failures_total from every gateway to whatever metrics pipeline the fleet already uses (Prometheus node exporter over a low-bandwidth link is common on cellular-connected sites). A spool depth that climbs steadily during otherwise-healthy hours is the earliest signal of a slow upstream degradation — a TSDB approaching a compaction window, a broker running low on disk — and catching it from spool_depth_rows before the outage is total turns an incident into a capacity-planning ticket.

Continue exploring

Related in this section