Event deduplication window

Deep dive

How it works

In: event id + arrival time · Out: accept (process) or drop (ack and ignore) · State: ring of per-bucket id lists plus one union set · Cost: one hash lookup per event, O(1) expected; expiry is amortized O(1) per id

1. Choose W above the maximum redelivery delay (retry cap, visibility timeout, outage length) and a bucket width b; the ring holds N = W / b buckets.
2. For each event, advance the clock: for every elapsed bucket push a new empty bucket at the front.
3. When the ring exceeds N buckets, pop the oldest and delete its ids from the union set.
4. Look the id up in the union set. Found: drop the event and still ack it, so the broker stops redelivering.
5. Not found: process it, insert the id into the set and the newest bucket.
6. An id is remembered for between W - b and W seconds. Reordered or late events need no special case: lookup ignores order, and a past-stamped event goes to the newest bucket.

If producers give no id, hash the canonical payload plus a producer sequence or timestamp; otherwise identical but distinct events collapse.

Atomicity. Check, effect and record must not be separable by a crash. In-memory state is lost on restart, so a redelivered event is processed again. With a database, insert the id (unique key) in the same transaction as the effect, as in the inbox step of the outbox pattern. With only an external effect (SMS), choose per risk: record first and accept a loss, or send first and accept a duplicate.

Partitioning. All copies of an id must reach one consumer: partition the stream by id or entity key (partitioned log), so each consumer holds only its slice.

Code

use std::collections::{HashSet, VecDeque};

/// Remembers every accepted id for at least (buckets-1) * bucket_secs and at most buckets * bucket_secs.
pub struct DedupWindow {
    bucket_secs: u64,
    max_buckets: usize,
    head: u64,                    // index (time / bucket_secs) of the newest bucket
    buckets: VecDeque<Vec<u64>>,  // ids per time bucket, newest at the front
    seen: HashSet<u64>,           // union of all buckets: one lookup per event
}

impl DedupWindow {
    pub fn new(bucket_secs: u64, max_buckets: usize) -> Self {
        let mut buckets = VecDeque::new();
        buckets.push_front(Vec::new());
        DedupWindow { bucket_secs, max_buckets, head: 0, buckets, seen: HashSet::new() }
    }

    fn advance(&mut self, now: u64) {
        let idx = now / self.bucket_secs;
        if idx.saturating_sub(self.head) >= self.max_buckets as u64 {
            self.seen.clear();                       // idle longer than the window
            self.buckets.clear();
            self.buckets.push_front(Vec::new());
            self.head = idx;
        }
        while self.head < idx {
            self.buckets.push_front(Vec::new());
            self.head += 1;
            if self.buckets.len() > self.max_buckets {
                for id in self.buckets.pop_back().unwrap() { self.seen.remove(&id); }
            }
        }
    }

    /// true = first sighting, process it; false = duplicate, drop and ack it.
    /// `now` is arrival time; an event stamped in the past lands in the newest bucket.
    pub fn first_seen(&mut self, id: u64, now: u64) -> bool {
        self.advance(now);
        if !self.seen.insert(id) { return false; }
        self.buckets[0].push(id);
        true
    }

    pub fn held(&self) -> usize { self.seen.len() }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn window_semantics() {
        let mut w = DedupWindow::new(1, 60);          // 60 one-second buckets
        assert!(w.first_seen(7, 0));
        assert!(!w.first_seen(7, 30));                // duplicate inside the window: dropped
        assert!(!w.first_seen(7, 59));
        assert!(w.first_seen(7, 60));                 // bucket of t=0 rotated out: accepted again
        for id in [5, 3, 4] { assert!(w.first_seen(id, 61)); }
        for id in [3, 5, 4] { assert!(!w.first_seen(id, 62)); } // reordered duplicates dropped
    }

    #[test]
    fn memory_bounded_by_rate_times_window() {
        let (mut w, rate) = (DedupWindow::new(1, 60), 1000u64);
        for t in 0..600 { for i in 0..rate { assert!(w.first_seen(t * rate + i, t)); } } // all distinct: all accepted
        assert!(w.held() as u64 <= rate * 60);
    }
}

Complexity and constants

• Per event: one SipHash lookup plus one push, expected O(1); each id is removed once at rotation.
• Hash set (hashbrown: 7/8 load, 9 bytes per slot, power-of-two growth) about 14 bytes per id plus 8 in the bucket list: about 22 bytes per id steady state. Measured peak RSS for 5.4 M held ids: 201.5 MB (about 37 bytes per id, with doubling transients and Vec slack).
• Bloom filter at 1% false positives: -ln(0.01) / (ln 2)^2 = 9.59 bits per id, k = 7 (computed from the standard formula, not measured): about 1.2 bytes per id.
• The cost driver is the cache miss on a multi-megabyte table, not hashing.

Efficiency

• Cost: measured on Apple M4 Pro: about 45-55 ns per check plus insert plus amortized expiry over 10 M events at 100 k events/s, 60 s window, 10% redeliveries, 5.4 M ids held at peak; bound by cache misses on the set.
• Levers:
  • Rotate whole buckets instead of per-entry TTL timers — no timer or heap per id; expiry is a sequential walk of one Vec (this page's code) — always.
  • Store a 64-bit hash of the id, not the string — 8 bytes instead of tens; collision odds about n^2 / 2^65 (birthday bound), order of 1e-6 for 5 M ids — when ids are long.
  • Bloom filter, or a rotating pair of filters — about 1.2 bytes per id instead of 22, at 1% wrongly dropped new events (computed) — when losing 1 in 100 is acceptable or a cheap second check follows.
  • Cuckoo filter instead of Bloom — supports delete, so single ids can expire (Fan et al. 2014) — when ids expire individually.
  • Per-producer sequence numbers — O(producers) state instead of O(rate x W) — when each producer numbers its events.
• Naive → standard → state of the art:
  • Unbounded HashSet<String> · memory grows forever.
  • Per-entry TTL map (Redis SET key 1 NX EX w) · about one network round trip per event, one key plus timer per id.
  • Bucket ring with hashed ids, partitioned by id, or per-producer sequences · tens of ns per event, bytes per id or per producer.

Implementation notes

• Use a fast hasher (or the id itself when it is a random UUID half): lookups drop to the cache-miss floor.
• Rotate on a monotonic clock; a wall-clock jump must not wipe or freeze the window.
• Persist ids with the offset commit (or in a unique-key table) if restart gaps can exceed the redelivery delay.
• Dropping must still ack, or the duplicate returns forever and fills a dead-letter queue.

Pitfalls

• Window shorter than the redelivery delay: a late retry is processed twice. Size W from the retry cap.
• Record-then-crash: id stored, effect never ran, redelivery dropped. Commit both together.
• Bloom false positives drop real events; never use alone for money.
• Not routing by id: copies reach different consumers and both pass.

Variants and alternatives

In production

• Amazon SQS FIFO: MessageDeduplicationId, 5-minute deduplication window.
• Azure Service Bus: duplicate detection on MessageId; window default 10 minutes (20 seconds to 7 days).
• NATS JetStream: Nats-Msg-Id header, 2-minute duplicate window by default.
• Apache Kafka: the idempotent producer deduplicates retries per producer id and sequence number.
• Google Pub/Sub: exactly-once delivery within a region, keyed by message id.
• Apache Flink: ROW_NUMBER() deduplication; webhook receivers and notifiers claim ids with Redis SET key value NX EX seconds.

Implementations

Sources

• Amazon SQS Developer Guide, Message deduplication ID. docs
• Microsoft, Duplicate detection, Azure Service Bus. docs
• NATS docs, JetStream model deep dive. docs
• Apache Kafka documentation, Design. docs
• Google Cloud, Exactly-once delivery, Pub/Sub. docs
• Apache Flink documentation, Deduplication. docs
• Bloom, Space/time trade-offs in hash coding with allowable errors, CACM 1970. paper
• Fan et al., Cuckoo Filter, CoNEXT 2014. paper