Deduplicating Replayed Incident Messages After Broker Failover

At 02:14 a hurricane band takes down the primary message broker carrying the live incident feed. A standby promotes itself in under a second, and the operations floor barely notices — except that the Common Operating Picture (COP) now shows a structure-fire incident with a reported casualty count of six, when the field only ever reported three. Two engine-status updates that had already turned a unit “available” have flipped it back to “assigned”. Nothing malfunctioned: the standby broker did exactly what at-least-once delivery promises, re-delivering every message the failed broker had not yet seen acknowledged. The consumer that already applied those updates has now applied them a second time. This page solves that one narrow, dangerous failure mode — turning a burst of replayed messages after a broker failover into a stream where every incident effect lands exactly once, without ever silently dropping a genuine update.

Root Cause and Operational Impact

Message Queuing Telemetry Transport (MQTT) and every comparable broker default to at-least-once semantics for reliable topics: QoS 1. The broker persists an in-flight message and keeps re-sending it until the consumer returns a PUBACK. That acknowledgement is what lets the broker forget the message. When the primary fails over to a standby, the standby restores its state from the last persisted, replicated session — which almost always predates the acknowledgements that were in flight at the moment of failure. So the standby re-delivers messages the original consumer already processed. The transport did not lie; at-least-once means at-least-once, and a failover is precisely when the “more than once” case fires.

The impact is not cosmetic. An incident feed carries state transitions that are frequently non-idempotent by nature: increment a casualty tally, append a resource to an assignment, advance an incident to a new operational period. Apply any of those twice and the operational store diverges from ground truth. A double-counted casualty figure drives a mutual-aid request that pulls units off other incidents. A replayed “assigned” event re-commits an engine that dispatch had already released. Because the National Incident Management System (NIMS) and the Federal Emergency Management Agency (FEMA) both require that the decision record be reconstructable for after-action review, a COP that quietly absorbed duplicate updates is not defensible — and neither is one that dropped a real update while trying to suppress a fake one. The correct behaviour is exactly-once effect at the consumer, enforced with an idempotency key and a bounded dedup window, and it belongs in the same feed layer described in WebSocket & MQTT for Live Incident Feeds rather than trusted to the broker’s delivery guarantee alone.

Broker failover replay and consumer-side deduplication by idempotency key A publisher emits incident messages carrying producer-controlled sequence numbers to a primary broker at QoS 1. The primary delivers them to the consumer, which derives an idempotency key, applies the incident effect once, and stores the key in a bounded dedup window. The primary broker then fails and a standby promotes itself, restoring session state that predates the in-flight acknowledgements, so it replays the same messages. The consumer finds their keys already present in the dedup window, suppresses the replays, emits an audit record for each, and the Common Operating Picture receives every effect exactly once. Publisher Broker (primary → standby) Consumer + dedup window COP publish seq 41 · QoS 1 deliver seq 41 key A1 first-seen → apply · store key effect #1 primary fails · standby promotes session state predates PUBACK replay seq 41 (unacked) key A1 already seen suppress replay emit audit record no second effect publish seq 42 deliver seq 42 key A2 first-seen → effect #2

Tiered Resolution Strategy

Resolve the stream in ordered tiers, from the definitive fix down to a safe default that always leaves an audit trail. The governing rule: suppress a proven duplicate, but never discard a message you cannot prove is one — an incident update lost to over-aggressive deduplication is worse than a duplicate caught later.

  1. Derive a stable idempotency key (definitive). Hash producer-controlled fields — agency identifier, source device id, and a monotonic event sequence — into one key. Because the key comes from the payload, a broker replay of the identical bytes hashes to the identical key, while two genuinely distinct events never collide.
  2. Check a bounded dedup window. Look the key up in a store bounded by both time and entry count. A hit means the effect already happened and this delivery is a replay; a miss means first-seen. The window must outlive the longest possible failover replay delay.
  3. Apply the effect exactly once, atomically. Only first-seen keys mutate the operational store, and the effect plus the key are committed together so a crash can never leave the effect applied but the key unrecorded — which would re-open the duplicate on restart.
  4. Preserve per-key ordering. Track the highest sequence seen per source. A replay carrying a sequence at or below the last applied value for that source is rejected even if the key store was trimmed, so a late replay can never overwrite newer COP state.
  5. Suppress with a full audit record (safe default). Every dropped message emits an entry — key, source, sequence, reason — so an after-action reviewer can confirm each suppression removed a true duplicate and no genuine update vanished.

Production Python Implementation

The consumer below carries the full resolution path: idempotency-key derivation, a bounded dedup window with time and size limits, per-source ordering enforcement, exactly-once effect application, structured logging, explicit exception handling, and an immutable audit record for every suppressed replay. The dedup window and ordering map are shown in-process for clarity; in production, back them with a durable store (Redis with a TTL, or a small table) so the window survives a consumer restart. Thresholds are parameters, not literals, so they can be committed alongside the feed contract. Senior-engineer assumptions apply: message payloads are already decoded dictionaries, and the effect applier is injected so this component stays transport- and schema-agnostic.

python 
from __future__ import annotations

import hashlib
import logging
from collections import OrderedDict
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Callable, Optional

logger = logging.getLogger("incidentgis.dedup")


class Outcome(str, Enum):
    APPLIED = "applied"
    DUPLICATE = "duplicate_suppressed"
    STALE = "stale_out_of_order"
    MALFORMED = "malformed_rejected"


@dataclass
class IncidentMessage:
    agency: str
    source_id: str          # producer-controlled device / feed id
    sequence: int           # monotonic per (agency, source_id)
    payload: dict


@dataclass
class AuditEntry:
    """Immutable record of one dedup decision, emitted to the audit trail."""
    key: str
    source: str
    sequence: int
    outcome: str
    recorded_at: str = field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )


class ReplayDeduplicator:
    """Enforce exactly-once effect for an at-least-once incident feed.

    ``apply_effect`` is called only for first-seen, in-order messages. Every
    decision is logged and appended to ``audit_log`` so a corrected feed can be
    reconstructed against the exact window that produced it.
    """

    def __init__(
        self,
        apply_effect: Callable[[IncidentMessage], None],
        window_seconds: float = 600.0,
        max_entries: int = 100_000,
    ) -> None:
        self._apply_effect = apply_effect
        self.window_seconds = window_seconds
        self.max_entries = max_entries
        # key -> epoch seconds seen; OrderedDict gives O(1) FIFO eviction.
        self._seen: "OrderedDict[str, float]" = OrderedDict()
        # (agency, source_id) -> highest applied sequence.
        self._high_water: dict[tuple[str, str], int] = {}
        self.audit_log: list[AuditEntry] = []

    @staticmethod
    def idempotency_key(msg: IncidentMessage) -> str:
        """Stable key from producer-controlled fields only — never broker metadata."""
        raw = f"{msg.agency}|{msg.source_id}|{msg.sequence}".encode("utf-8")
        return hashlib.sha256(raw).hexdigest()

    def _evict(self, now: float) -> None:
        """Bound the window by both age and size so it cannot grow without limit."""
        cutoff = now - self.window_seconds
        while self._seen:
            key, ts = next(iter(self._seen.items()))
            if ts < cutoff or len(self._seen) > self.max_entries:
                self._seen.popitem(last=False)
            else:
                break

    def _audit(self, key: str, msg: IncidentMessage, outcome: Outcome) -> None:
        entry = AuditEntry(
            key=key, source=f"{msg.agency}/{msg.source_id}",
            sequence=msg.sequence, outcome=outcome.value,
        )
        self.audit_log.append(entry)
        logger.info("dedup_decision", extra={"audit": asdict(entry)})

    def process(self, msg: IncidentMessage) -> Outcome:
        now = datetime.now(timezone.utc).timestamp()
        try:
            # Guard: a malformed message can never advance state silently.
            if not msg.agency or not msg.source_id or msg.sequence < 0:
                self._audit("<none>", msg, Outcome.MALFORMED)
                return Outcome.MALFORMED

            key = self.idempotency_key(msg)
            self._evict(now)

            # Tier 2: already in the window → this is a failover replay.
            if key in self._seen:
                self._seen.move_to_end(key)
                self._audit(key, msg, Outcome.DUPLICATE)
                return Outcome.DUPLICATE

            # Tier 4: ordering guard catches replays evicted from the window.
            src = (msg.agency, msg.source_id)
            last = self._high_water.get(src)
            if last is not None and msg.sequence <= last:
                self._audit(key, msg, Outcome.STALE)
                return Outcome.STALE

            # Tier 3: first-seen and in-order → record key + apply effect together.
            # Recording the key before the effect keeps a retry idempotent even
            # if apply_effect raises; on failure we roll the key back below.
            self._seen[key] = now
            try:
                self._apply_effect(msg)
            except Exception:
                del self._seen[key]      # allow a genuine redelivery to retry
                raise
            self._high_water[src] = msg.sequence
            self._audit(key, msg, Outcome.APPLIED)
            return Outcome.APPLIED

        except Exception as exc:
            # Never let one bad message stall the feed; surface it for replay.
            logger.error("dedup_process_failed", exc_info=exc,
                         extra={"source": f"{msg.agency}/{msg.source_id}"})
            raise

The audit_log is the load-bearing output. Persisted as a committed, append-only artifact, it lets a post-incident reviewer replay every decision and confirm that each suppression removed a true duplicate — the reproducibility guarantee that keeps a deduplicated feed defensible and interoperable with downstream conflict resolution in multi-agency edits.

Validation Checklist

Verify every item before deploying the deduplicator to a live incident feed.

  • The idempotency key is derived from producer-controlled fields only — never from broker-assigned message ids or timestamps that change on replay.
  • The dedup window is bounded by both time and entry count, and window_seconds exceeds the maximum failover replay delay plus MQTT session expiry.
  • The window and per-source high-water marks are backed by a durable store so a consumer restart does not re-open every recent duplicate.
  • The key is recorded together with the effect atomically; a crash cannot leave the effect applied but the key unrecorded, or vice versa.
  • A replay whose key was already evicted is still caught by the per-source ordering guard and never overwrites newer COP state.
  • The first message from a new source (no high-water mark yet) is applied without raising.
  • Malformed messages are audited and rejected, never applied silently.
  • Every suppression appears in audit_log and routes to the incident logging sink, not stdout.

Edge Cases and Gotchas

  • Non-monotonic producer sequences. The whole scheme rests on a sequence that only ever increases per source. A device that resets its counter on reboot will emit low sequences that the ordering guard rejects as stale. Namespace the sequence with a boot-session id, or fold a producer restart epoch into the key so a legitimate post-reboot message is first-seen rather than a false duplicate.
  • Window shorter than the replay delay. If a broker persists an unacknowledged message for longer than window_seconds, its key is evicted before the replay arrives and the ordering guard becomes the only line of defence. Size the window against the broker’s real session-expiry and retry configuration, not an optimistic guess.
  • Effect that is itself non-idempotent on retry. Recording the key before apply_effect and rolling it back on failure keeps retries safe only if the effect is transactional. If the applier writes to an external system that partially commits, wrap it so the key and the effect share one transaction, or the rollback re-opens the duplicate.
  • Fan-out to multiple consumers. Where several consumers share a subscription for throughput, an in-process window on each one misses duplicates delivered to a different instance after failover. The dedup store must be shared across the consumer group, which is one more reason to keep it in a durable, central backend rather than local memory.
  • Legitimate resend versus replay. A field app that re-publishes after its own reconnect looks identical to a broker replay — and that is correct: both carry the same producer sequence, so both hash to the same key and are suppressed. Only a genuinely new event with a new sequence should ever apply, which is why the key must never include the transport’s delivery attempt.

Frequently Asked Questions

Why does an MQTT broker failover produce duplicate messages? At QoS 1, at-least-once delivery, the broker holds an in-flight message until the consumer acknowledges it. If the broker fails over to a standby before that acknowledgement is durably recorded, the standby re-delivers every unacknowledged message from its persisted session state. The consumer that already processed the original now receives an identical copy, so the same incident update is applied twice unless it is deduplicated.

Why not just use MQTT 5 QoS 2 for exactly-once delivery? QoS 2 guarantees exactly-once delivery of a packet between one client and one broker, but it does not survive a failover to a broker that never saw the four-way handshake, and it says nothing about a producer that re-publishes after its own reconnect. Exactly-once effect has to be enforced at the consumer with an idempotency key and a dedup window, which also protects against duplicates the transport layer can never see.

How large should the deduplication window be? It must be at least as long as the maximum time a message can survive in a failed-over broker’s persisted session plus the worst-case producer replay delay — typically several minutes for MQTT session expiry. Size it as a committed parameter, bound it by both time and entry count so it cannot grow without limit, and back it with a durable store so the window is not lost when the consumer itself restarts.

Up: WebSocket & MQTT for Live Incident Feeds

Other guides in WebSocket & MQTT for Live Incident Feeds

Related
Building a Live Incident Dashboard with Python and Leaflet
→
Related
Handling MQTT Reconnect Storms During Wildfire Surge
→