Replaying a Buffered Backlog Without Duplicates #

The buffer did its job: three hours of readings survived a dead uplink. Now they have to reach the database without arriving twice, without arriving out of order, and without starving the live feed that has just resumed. Replay is where most fallback-buffer implementations actually fail — not in capturing data, which is easy, but in draining it safely. This guide builds the drain loop for the fallback buffering and offline caching stage: a state machine that survives a crash mid-batch, an ordering rule that keeps downstream stateful processing correct, and a rate limit that protects the live path.

The Three Properties a Drain Must Have #

Idempotent at the sink. Every replayed row must be absorbable. The natural key — (device_id, observed_at, metric) — makes a duplicate a no-op rather than a second measurement. Without this, every one of the other properties becomes far harder, because the drain has to guarantee exactly-once delivery on its own.

Ordered within a device. Downstream state — rolling baselines, gap detection, last-value caches — walks a sensor’s readings in time order. A drain that replays newest-first, or that interleaves buffered and live readings from the same sensor arbitrarily, produces a baseline computed from a scrambled series. Order globally by observed_at within each device and the problem disappears.

Bounded in its resource use. A three-hour backlog for 200 sensors at one-minute cadence is 36 000 rows. Written as fast as the database accepts them, that is a burst that competes with live writes and, on a shared instance, with everything else. The drain needs a throttle.

Production-Ready Implementation #

The buffer table carries an explicit state column rather than a boolean, because “sent” and “failed permanently” are different outcomes that need different handling:

-- sqlite 3.40+ (gateway-local buffer)
CREATE TABLE IF NOT EXISTS buffered_reading (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    device_id    TEXT    NOT NULL,
    observed_at  TEXT    NOT NULL,          -- ISO 8601 UTC
    metric       TEXT    NOT NULL,
    value        REAL    NOT NULL,
    state        TEXT    NOT NULL DEFAULT 'pending',   -- pending | sending | sent | failed
    attempts     INTEGER NOT NULL DEFAULT 0,
    last_error   TEXT,
    UNIQUE (device_id, observed_at, metric)
);

CREATE INDEX IF NOT EXISTS buffered_pending
    ON buffered_reading (state, device_id, observed_at);

The UNIQUE constraint deduplicates at capture time, so a device that re-sends during a flapping link does not inflate the backlog it will later have to drain.

The drain loop claims a batch, writes it, then marks it — in that order:

# python 3.11 · sqlite3 (stdlib) · psycopg[binary]==3.1.18
import sqlite3
import time

CLAIM_SQL = """
UPDATE buffered_reading
   SET state = 'sending', attempts = attempts + 1
 WHERE id IN (
     SELECT id FROM buffered_reading
      WHERE state IN ('pending', 'sending')
      ORDER BY device_id, observed_at
      LIMIT ?
 )
RETURNING id, device_id, observed_at, metric, value
"""


def drain_batch(buf: sqlite3.Connection, sink, batch_size: int = 500) -> int:
    """Move one batch from the buffer to the sink. Returns rows drained.

    Ordering: (device_id, observed_at) so each sensor's series replays in time
    order. Rows already in 'sending' are re-claimed — a crash between the write
    and the mark leaves them there, and the idempotent sink absorbs the repeat.
    """
    rows = buf.execute(CLAIM_SQL, (batch_size,)).fetchall()
    buf.commit()
    if not rows:
        return 0

    payload = [
        {"device_id": d, "observed_at": t, "metric": m, "value": v}
        for (_id, d, t, m, v) in rows
    ]
    try:
        sink.write_batch(payload)                       # ON CONFLICT DO NOTHING
    except Exception as exc:
        buf.executemany(
            "UPDATE buffered_reading SET state='pending', last_error=? WHERE id=?",
            [(str(exc)[:400], r[0]) for r in rows],
        )
        buf.commit()
        raise

    buf.executemany(
        "UPDATE buffered_reading SET state='sent' WHERE id=?", [(r[0],) for r in rows]
    )
    buf.commit()
    return len(rows)

The state IN ('pending', 'sending') in the claim is the crash-recovery mechanism: a process that died between the sink write and the mark leaves rows stuck in sending, and the next pass picks them up rather than leaving them stranded. That re-claim is only safe because the sink is idempotent — which is why the two design decisions cannot be separated.

The throttle keeps replay from crowding out live traffic:

def drain(buf, sink, *, budget_rows_per_sec: float = 400, batch_size: int = 500) -> int:
    """Drain the backlog at a bounded rate, yielding capacity to the live path."""
    total = 0
    while True:
        started = time.monotonic()
        n = drain_batch(buf, sink, batch_size)
        if n == 0:
            return total
        total += n
        target = n / budget_rows_per_sec
        elapsed = time.monotonic() - started
        if elapsed < target:
            time.sleep(target - elapsed)

Expressing the throttle as rows per second rather than a fixed sleep means the drain automatically uses more of the window when the sink is fast and backs off when it is slow.

Parameter Tuning Guide #

Two numbers deserve attention. The replay budget must stay below the sink’s spare capacity — if live ingest uses 60% of the write throughput, a replay budget above 40% makes the live path the thing that falls behind. And the retention row is a policy decision, not a technical one: when the buffer fills, dropping the oldest readings keeps the recent picture intact, while dropping the newest keeps the historical record complete. For regulatory reporting the second is sometimes correct, and it should be a conscious choice rather than a default.

Verification and Testing #

The crash test is the one that matters, because it exercises the exact window the state machine exists to cover.

# python 3.11 · pytest==8.2.0
class FlakySink:
    """Writes successfully, then raises before the caller can mark the rows sent."""

    def __init__(self):
        self.written = []
        self.crash_after_write = True

    def write_batch(self, rows):
        self.written.extend(rows)
        if self.crash_after_write:
            self.crash_after_write = False
            raise ConnectionResetError("sink connection dropped after write")


def test_crash_between_write_and_mark_replays_without_duplicating(buf):
    seed(buf, count=10)
    sink = FlakySink()

    with pytest.raises(ConnectionResetError):
        drain_batch(buf, sink, batch_size=10)

    drain_batch(buf, sink, batch_size=10)          # second pass re-claims and completes

    assert len(sink.written) == 20                 # the sink saw them twice...
    assert len({(r["device_id"], r["observed_at"]) for r in sink.written}) == 10   # ...as 10 readings
    assert pending_count(buf) == 0


def test_replay_preserves_per_device_time_order(buf):
    seed_interleaved(buf, devices=3, per_device=50)
    sink = RecordingSink()
    while drain_batch(buf, sink, batch_size=17):
        pass
    for device in {r["device_id"] for r in sink.written}:
        times = [r["observed_at"] for r in sink.written if r["device_id"] == device]
        assert times == sorted(times)

In production, the numbers to watch are the pending count and its first derivative. A pending count that falls steadily is a healthy drain; one that plateaus means the replay budget equals the arrival rate; one that rises during replay means the drain is losing to live ingest and the budget needs raising or the outage will never clear.

Gotchas #

Marking rows sent before writing them. The ordering looks harmless and loses data on every crash. Write first. Always.

Draining oldest-first globally instead of per device. Global time ordering across devices sounds tidier and is unnecessary — no downstream consumer cares about the interleaving of two different sensors. Ordering per device is what matters, and it lets the claim query use the index.

No cap on attempts. A row that fails permanently — a malformed value that the sink rejects with a constraint violation — will be re-claimed forever, blocking the batch behind it. Move rows past a threshold (five attempts is generous) to failed and surface the count.

Deleting rows instead of marking them sent. Deletion during a drain fragments the SQLite file and makes the crash window unrecoverable, because a deleted row cannot be re-claimed. Mark them, then delete sent rows in a separate vacuum pass when the buffer is idle.

FAQ #

Related #

• Fallback Buffering & Offline Caching for Environmental IoT Sensors — the buffering stage this replay drains
• Building a Local SQLite Fallback Buffer for Remote Sensors — the buffer schema and write path this guide reads from
• MQTT QoS Levels and Duplicate Sensor Messages — the same idempotent-sink argument, applied to transport duplicates