Deduplicating Bursty Bin-Sensor Reconnects in Python

This page solves one specific production failure: a bin sensor drops off the network, buffers readings locally, and on reconnect replays the whole buffer in a burst — so the same fill-level reading is counted twice, a bin looks serviced when it is full, and the route the solver builds is wrong.

Battery-powered LoRaWAN and NB-IoT bin sensors are designed to buffer while offline and flush on reconnect; duplicate delivery is the normal case, not the exception. The naive fix — dedupe on timestamp — is worse than nothing, because two genuinely distinct readings from the same device can legitimately carry the same coarse timestamp, and collapsing them silently deletes real data. Correct deduplication needs a stable event key that identifies one logical reading regardless of how many times it is delivered, plus a bounded memory of keys already seen. This deduplicator is the idempotency stage of the Bin Sensor API Sync layer, running immediately after the vendor frames are decoded in Parsing IoT Fill-Level Sensor Payloads and before anything reaches the solver.

Environment & Data Prerequisites

The deduplicator uses the Python 3.10+ standard library — hashlib, json, logging, collections, and datetime — for hashing and the bounded seen-set, plus Pydantic to model and coerce the decoded reading:

python -m pip install pydantic==2.9.2 pytest==8.2.0

Each item is one decoded fill-level reading from a single bin sensor. The deduplicator reasons over exactly these fields:

Field	Unit / type	Rule
`device_id`	str	stable per-sensor identifier
`seq`	int	device-local counter; monotonic but resets on battery swap
`ts`	datetime (tz-aware UTC)	coarse, sometimes shared across readings
`fill_pct`	float, percent	0.0 … 100.0 measured fill level

Two rules matter before any code. First, seq alone is not a safe key across the sensor’s whole life, because a battery swap resets the counter — so the key must also bind the reading’s content. Second, ts must never be the sole key: coarse-clock sensors legitimately emit two distinct readings inside the same second, and deduping on timestamp would silently delete one.

Step-by-Step Implementation

Step 1 — Model the reading and compute a stable idempotency key

The reading is a Pydantic model so malformed frames are rejected at the boundary. The idempotency key binds device_id, seq, and a content hash of the measurement, so a replayed frame produces an identical key while two same-second-but-different readings produce different keys. The content hash uses SHA-256 over a canonical, sorted JSON encoding so the digest is stable regardless of field ordering in the source frame.

import json
import hashlib
import logging
from collections import OrderedDict
from datetime import datetime, timezone, timedelta
from pydantic import BaseModel, Field


class SensorReading(BaseModel):
    device_id: str = Field(min_length=1)
    seq: int
    ts: datetime
    fill_pct: float = Field(ge=0.0, le=100.0)


def idempotency_key(reading: SensorReading) -> str:
    """A stable key: device + seq + a content hash of the measurement.

    Binding the content hash means a battery-swap seq reset cannot alias two
    different readings, and two distinct readings sharing a timestamp get
    different keys because their content differs.
    """
    content = json.dumps(
        {"ts": reading.ts.isoformat(), "fill_pct": reading.fill_pct},
        sort_keys=True,
        separators=(",", ":"),
    )
    digest = hashlib.sha256(content.encode("utf-8")).hexdigest()[:16]
    return f"{reading.device_id}:{reading.seq}:{digest}"

Step 2 — A bounded TTL seen-set

The seen-set remembers keys already admitted. To stay bounded in memory it is an OrderedDict mapping key to admission time, capped at both a maximum age (TTL) and a maximum size (LRU eviction of the oldest key). A plain unbounded set would grow for the life of the process — the exact leak this page’s Common Errors section warns against.

class TTLSeenSet:
    def __init__(self, ttl: timedelta = timedelta(hours=6), max_size: int = 100_000) -> None:
        self._ttl = ttl
        self._max_size = max_size
        self._seen: "OrderedDict[str, datetime]" = OrderedDict()

    def _evict_expired(self, now: datetime) -> None:
        cutoff = now - self._ttl
        while self._seen:
            oldest_key = next(iter(self._seen))
            if self._seen[oldest_key] < cutoff:
                self._seen.pop(oldest_key)
            else:
                break

    def check_and_add(self, key: str, now: datetime) -> bool:
        """Return True if the key is new (admit); False if already seen (suppress)."""
        self._evict_expired(now)
        if key in self._seen:
            return False                       # duplicate — first-write-wins
        self._seen[key] = now
        self._seen.move_to_end(key)
        while len(self._seen) > self._max_size:
            self._seen.popitem(last=False)     # LRU bound: drop the oldest key
        return True

Step 3 — The deduplicator with first-write-wins and audit fields

The deduplicator drives the two pieces above. It counts what it received, what it admitted, and what it suppressed, and it keeps the first copy of any key — later replays are discarded, never overwritten, so a corrupted retransmission cannot clobber a good reading.

class JSONFormatter(logging.Formatter):
    """Emit each log record as a single-line JSON object for dedupe audit."""

    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "logged_at": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "event": record.getMessage(),
            "payload": getattr(record, "payload", None),
        }
        return json.dumps(entry)


audit_logger = logging.getLogger("waste_ops.dedupe")
if not audit_logger.handlers:
    _h = logging.StreamHandler()
    _h.setFormatter(JSONFormatter())
    audit_logger.addHandler(_h)
audit_logger.setLevel(logging.INFO)


class ReconnectDeduplicator:
    def __init__(self, seen: TTLSeenSet | None = None) -> None:
        self._seen = seen or TTLSeenSet()

    def process_burst(self, readings: list[SensorReading]) -> list[SensorReading]:
        unique: list[SensorReading] = []
        suppressed = 0
        now = datetime.now(timezone.utc)
        for reading in readings:
            key = idempotency_key(reading)
            if self._seen.check_and_add(key, now):
                unique.append(reading)          # first write wins
            else:
                suppressed += 1                 # exact replay, discarded
        audit_logger.info(
            "burst_deduplicated",
            extra={"payload": {
                "received": len(readings),
                "unique": len(unique),
                "duplicates_suppressed": suppressed,
            }},
        )
        return unique

Step 4 — Run a replayed burst

The example burst contains a genuine replay (seq 41 sent twice) and a same-timestamp-but-distinct reading (two readings at the same coarse second with different fill levels). The deduplicator collapses the replay and keeps both distinct readings.

if __name__ == "__main__":
    dedup = ReconnectDeduplicator()
    burst = [
        SensorReading(device_id="bin-A17", seq=41, ts=datetime(2026, 7, 16, 9, 0, 0, tzinfo=timezone.utc), fill_pct=62.5),
        SensorReading(device_id="bin-A17", seq=41, ts=datetime(2026, 7, 16, 9, 0, 0, tzinfo=timezone.utc), fill_pct=62.5),  # replay
        SensorReading(device_id="bin-A17", seq=42, ts=datetime(2026, 7, 16, 9, 0, 0, tzinfo=timezone.utc), fill_pct=71.0),  # same ts, distinct
        SensorReading(device_id="bin-B04", seq=8,  ts=datetime(2026, 7, 16, 9, 0, 5, tzinfo=timezone.utc), fill_pct=15.0),
    ]
    kept = dedup.process_burst(burst)
    print(f"received {len(burst)}, kept {len(kept)}")   # received 4, kept 3

Compliance Output

Each burst emits one JSON line to the append-only dedupe audit. The run above serializes as:

{"logged_at": "2026-07-16T09:00:11.204731+00:00", "level": "INFO", "event": "burst_deduplicated", "payload": {"received": 4, "unique": 3, "duplicates_suppressed": 1}}

Each field earns its place in the audit trail:

received — the raw count delivered in the burst, including replays. Retaining it distinguishes a chatty reconnect (high received, high duplicates_suppressed) from real new data, which matters when a sensor’s replay rate is itself a health signal.
unique — the number of distinct readings admitted downstream. This is the count that reaches the solver and the service record, so it is the figure a municipal collection report reconciles against expected bin visits.
duplicates_suppressed — replays collapsed under first-write-wins. Logging the suppression count, rather than silently dropping, keeps the audit truthful: an auditor can prove the pipeline saw the replays and made a deliberate idempotent decision rather than losing data.

Because the ledger is append-only and every burst records its own three counts, received must always equal unique + duplicates_suppressed; a run where those do not reconcile is a signal the Schema Validation Pipelines layer rejected malformed frames upstream, and the DOT/FMCSA rule mapping reference resolves any completeness gap into the field a compliance reviewer reads.

Verification

The tests prove the two properties that matter: a replayed burst yields the same unique set as a single delivery (idempotency), and two distinct readings sharing a timestamp are both kept (no timestamp-only collapse).

from datetime import datetime, timezone


def _reading(seq: int, fill: float, second: int = 0) -> SensorReading:
    return SensorReading(
        device_id="bin-A17", seq=seq,
        ts=datetime(2026, 7, 16, 9, 0, second, tzinfo=timezone.utc),
        fill_pct=fill,
    )


def test_replayed_burst_is_idempotent():
    dedup = ReconnectDeduplicator()
    original = [_reading(41, 62.5), _reading(42, 71.0), _reading(43, 80.0)]
    first_pass = dedup.process_burst(original)
    # sensor reconnects and replays the identical buffer
    replay_pass = dedup.process_burst(original)
    assert len(first_pass) == 3
    assert replay_pass == []                     # every key already seen — all suppressed


def test_same_timestamp_distinct_readings_both_kept():
    dedup = ReconnectDeduplicator()
    burst = [_reading(41, 62.5), _reading(42, 71.0)]   # same second, different fill and seq
    kept = dedup.process_burst(burst)
    assert len(kept) == 2                        # distinct content hashes: neither collapsed


def test_exact_replay_within_one_burst_collapses():
    dedup = ReconnectDeduplicator()
    burst = [_reading(41, 62.5), _reading(41, 62.5)]   # identical frame twice
    kept = dedup.process_burst(burst)
    assert len(kept) == 1                        # first-write-wins

Common Errors

Two legitimate readings disappear whenever the sensor’s clock is coarse. Root cause: the dedupe key was the timestamp (or device_id + ts) alone, so two distinct readings emitted inside the same second aliased to one key and one was discarded. Fix: bind the reading’s content into the key as Step 1 does — device_id + seq + content_hash — so distinct measurements produce distinct keys even when their timestamps match exactly.

The worker’s memory grows without bound over a multi-day run. Root cause: the seen-set was a plain set() that accumulates every key forever, so a fleet emitting millions of readings leaks steadily until it is killed. Fix: use a bounded structure with both a TTL and a size cap, as TTLSeenSet in Step 2 does — expired keys are evicted by age and the oldest keys are dropped by LRU once the size cap is hit, so memory stays flat.

A late replay is admitted as new after clock skew reorders the burst. Root cause: the key incorporated a wall-clock ingestion time or a re-derived timestamp instead of the reading’s own stable fields, so the same logical reading hashed differently on its second delivery. Fix: build the key only from device-stable, content-stable fields (Step 1), never from ingestion time; the reading’s own seq and measurement content are invariant across deliveries, so the key is too.

FAQ

Why first-write-wins instead of last-write-wins?

On a reconnect the buffered copy and its replay are meant to be identical, so the choice is usually moot — but when a retransmission is corrupted in flight, first-write-wins keeps the copy that already passed schema validation rather than letting a damaged duplicate overwrite it. Preserving the first admitted reading also keeps the audit trail monotonic: a key, once decided, never changes meaning.

What TTL should the seen-set use?

Set the TTL comfortably longer than the longest offline buffering window your sensors support, so a reading replayed after a long outage is still recognized as a duplicate. Six hours is a safe default for daily-route bins; extend it for sensors that can buffer across a weekend. The size cap is the memory backstop for when the TTL alone would hold too many keys, and LRU eviction keeps the most recently active devices covered.

Up: Bin Sensor API Sync

Bin Sensor API Sync — the parent layer that reconciles bin-sensor state against the ingestion store.
Parsing IoT Fill-Level Sensor Payloads — decoding the vendor frames that arrive in these reconnect bursts.
Async Batch Processing — the flow-controlled stage the deduplicated readings feed into.
Schema Validation Pipelines — the type-and-containment contract that rejects malformed frames before this deduplicator sees them.