Approach: use an in-memory hash table for O(1) presence checks and a time-ordered structure to expire entries when TTL passes. Provide thread-safety and periodic cleanup to remove stale IDs. For resilience and scale, move the store to a shared external system (Redis) using atomic operations (SETNX/EXPIRE or Lua), add persistence, and partition keys across nodes.Python implementation (thread-safe, in-memory with TTL and O(1) checks):python
import time
import threading
from collections import deque
class InMemoryDedupe:
def __init__(self, ttl_seconds=300, cleanup_interval=1.0):
self.ttl = ttl_seconds
self.store = {} # id -> expiry_time
self.expiry_queue = deque() # (expiry_time, id) ordered by insertion
self.lock = threading.Lock()
self._stop = False
self._thread = threading.Thread(target=self._cleanup_loop, daemon=True)
self._thread.start()
self.cleanup_interval = cleanup_interval
def is_duplicate(self, msg_id):
"""Return True if msg_id seen (and not expired). Otherwise record it and return False."""
now = time.time()
with self.lock:
expiry = self.store.get(msg_id)
if expiry and expiry > now:
return True
# record new id
expiry_time = now + self.ttl
self.store[msg_id] = expiry_time
self.expiry_queue.append((expiry_time, msg_id))
return False
def _cleanup_loop(self):
while not self._stop:
now = time.time()
with self.lock:
while self.expiry_queue and self.expiry_queue[0][0] <= now:
_, mid = self.expiry_queue.popleft()
# only remove if expiry matches (avoid race where same id reinserted)
if self.store.get(mid, 0) <= now:
self.store.pop(mid, None)
time.sleep(self.cleanup_interval)
def stop(self):
self._stop = True
self._thread.join()
