"""Reference solution — Exactly-Once Illusion: Windowed Deduplication. DO NOT READ BEFORE YOU HAVE RUN THE PROBLEM UNDER THE CLOCK. Teaching text: ../../../WARMUP.md § Chapter 9. Exactly-once DELIVERY is impossible. Sender sends, receiver processes, ack is lost -- the sender cannot distinguish "never arrived" from "arrived, ack lost", and adding round trips just moves the problem to the ack of the ack. That is the Two Generals problem, not a protocol deficiency. What IS achievable is exactly-once PROCESSING: at-least-once delivery plus an idempotent consumer. Which requires a STABLE idempotency key generated by the PRODUCER and unchanged across retries. Generate it at send time and every retry has a new key, so dedupe silently does nothing -- which is exactly why Stripe makes the client supply Idempotency-Key. For dedupe specifically, a Bloom filter's error points the DANGEROUS way. A false positive means "I think I've seen this" about a message you have not, so the filter DROPS A REAL MESSAGE, silently. Fine for analytics counting; company-ending for payments. So use it as a negative cache in front of an exact store: "definitely absent" is exact, "possibly present" consults the store. The filter then saves lookups instead of losing data. """ from __future__ import annotations import hashlib import math import time from collections import deque # --------------------------------------------------------------------------- # Gate 1 # --------------------------------------------------------------------------- class ExactDedupe: """Exact and unbounded. The correctness baseline, not a shippable design.""" __slots__ = ("_seen", "duplicates") def __init__(self): self._seen = set() self.duplicates = 0 def is_duplicate(self, key): if key in self._seen: self.duplicates += 1 return True self._seen.add(key) return False @property def seen_count(self): return len(self._seen) # --------------------------------------------------------------------------- # Gate 2 # --------------------------------------------------------------------------- class WindowedDedupe: """Exact within the window. Memory bounded by rate x window.""" __slots__ = ("window", "_clock", "_seen", "_order", "duplicates") def __init__(self, window_seconds, clock=time.monotonic): if window_seconds <= 0: raise ValueError("window must be positive") self.window = float(window_seconds) self._clock = clock self._seen = {} self._order = deque() # (timestamp, key) in arrival order self.duplicates = 0 def _evict(self, now): # Amortised O(1): the deque is in arrival order, so eviction only ever # walks the front. Scanning every key on every call would make the # bound worthless. cutoff = now - self.window while self._order and self._order[0][0] <= cutoff: _, key = self._order.popleft() if self._seen.get(key) is not None and self._seen[key] <= cutoff: del self._seen[key] def is_duplicate(self, key): now = self._clock() self._evict(now) if key in self._seen: self.duplicates += 1 return True self._seen[key] = now self._order.append((now, key)) return False def __len__(self): return len(self._seen) # --------------------------------------------------------------------------- # Gate 3 # --------------------------------------------------------------------------- class BloomFilter: """Constant memory. False positives, never false negatives.""" def __init__(self, capacity, error_rate=0.01): if capacity <= 0: raise ValueError("capacity must be positive") if not 0 < error_rate < 1: raise ValueError("error_rate must be in (0, 1)") self.capacity = capacity self.error_rate = error_rate # m = -n ln(p) / (ln 2)^2 k = (m/n) ln 2 self.m = max(8, int(math.ceil( -capacity * math.log(error_rate) / (math.log(2) ** 2)))) self.k = max(1, int(round(self.m / capacity * math.log(2)))) self._bits = bytearray((self.m + 7) // 8) self.count = 0 def _positions(self, item): # Kirsch-Mitzenmacher: two independent hashes simulate k of them, so # you pay for two digests instead of k. data = item.encode() if isinstance(item, str) else bytes(item) digest = hashlib.blake2b(data, digest_size=16).digest() h1 = int.from_bytes(digest[:8], "little") h2 = int.from_bytes(digest[8:], "little") | 1 # odd -> full period for i in range(self.k): yield (h1 + i * h2) % self.m def add(self, item): for pos in self._positions(item): self._bits[pos >> 3] |= 1 << (pos & 7) self.count += 1 def __contains__(self, item): return all(self._bits[p >> 3] & (1 << (p & 7)) for p in self._positions(item)) def current_fpr(self): """Degrades silently as you overfill — so expose it and alarm on it.""" return (1 - math.exp(-self.k * self.count / self.m)) ** self.k @property def bytes_used(self): return len(self._bits) class SafeDedupe: """Bloom as a NEGATIVE cache in front of an exact store. 'Definitely absent' is exact -> process, no lookup needed. 'Possibly present' -> consult the authoritative store. So the filter saves lookups; it never loses a message. This is exactly how an LSM engine uses per-file Bloom filters to skip files that cannot contain a key. """ def __init__(self, capacity, exact_store, error_rate=0.01): self.bloom = BloomFilter(capacity, error_rate) self._exact = exact_store self.lookups_avoided = 0 self.lookups_performed = 0 def is_duplicate(self, key): if key not in self.bloom: # exact answer: definitely new self.lookups_avoided += 1 self.bloom.add(key) self._exact.add(key) return False self.lookups_performed += 1 # maybe: check authoritatively if key in self._exact: return True self.bloom.add(key) self._exact.add(key) return False # --------------------------------------------------------------------------- # Gate 4 # --------------------------------------------------------------------------- class Reorderer: """Per-key sequence numbers with a bounded reorder buffer. Unbounded reordering means unbounded memory, so the window is not optional. And when the window forces you to give up on a missing sequence, the gap must be ANNOUNCED -- a silent gap is a correctness bug the consumer cannot see. """ def __init__(self, window, on_gap=None): if window <= 0: raise ValueError("window must be positive") self.window = window self._on_gap = on_gap self._next = {} # key -> next expected sequence self._buffer = {} # key -> {seq: payload} self.gaps = [] def offer(self, key, seq, payload): expected = self._next.setdefault(key, 0) buffered = self._buffer.setdefault(key, {}) if seq < expected or seq in buffered: return [] # duplicate or already out buffered[seq] = payload released = self._drain(key) if len(buffered) > self.window: # Give up on the missing sequences and jump to the lowest buffered # one, announcing what was skipped. lowest = min(buffered) missing = list(range(self._next[key], lowest)) if missing: self.gaps.append((key, missing)) if self._on_gap: self._on_gap(key, missing) self._next[key] = lowest released.extend(self._drain(key)) return released def _drain(self, key): out = [] buffered = self._buffer[key] while self._next[key] in buffered: out.append(buffered.pop(self._next[key])) self._next[key] += 1 return out def flush(self): """Emit everything still buffered, in order, announcing the gaps.""" out = [] for key in sorted(self._buffer): buffered = self._buffer[key] while buffered: lowest = min(buffered) missing = list(range(self._next[key], lowest)) if missing: self.gaps.append((key, missing)) if self._on_gap: self._on_gap(key, missing) self._next[key] = lowest out.extend(self._drain(key)) return out def pending(self, key=None): if key is not None: return len(self._buffer.get(key, {})) return sum(len(v) for v in self._buffer.values())