S075 · Message Queues — SQS, RabbitMQ, Kafka as Queue

#!/usr/bin/env python3
"""tiny_queue.py — SQS-style queue in one file.
 
Features:
  - at-least-once delivery
  - per-message visibility timeout (invisible while being processed)
  - dead-letter queue after N failed receives
  - idempotent consumer using a seen-set
 
Run:  python tiny_queue.py
No dependencies. Uses threads to simulate producer + workers.
"""
from __future__ import annotations
 
import random
import threading
import time
import uuid
from collections import deque
from dataclasses import dataclass, field
 
 
@dataclass
class Message:
    body: dict
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    receive_count: int = 0
    invisible_until: float = 0.0
 
 
class Queue:
    def __init__(self, name: str, visibility_timeout: float = 0.5,
                 max_receives: int = 3, dlq: "Queue | None" = None):
        self.name = name
        self.msgs: deque[Message] = deque()
        self.visibility_timeout = visibility_timeout
        self.max_receives = max_receives
        self.dlq = dlq
        self.lock = threading.Lock()
 
    def send(self, body: dict) -> str:
        with self.lock:
            m = Message(body=body)
            self.msgs.append(m)
            return m.id
 
    def receive(self) -> Message | None:
        """Return the first visible message; make it invisible."""
        now = time.time()
        with self.lock:
            for m in self.msgs:
                if m.invisible_until <= now:
                    m.invisible_until = now + self.visibility_timeout
                    m.receive_count += 1
                    if m.receive_count > self.max_receives and self.dlq:
                        # Move to DLQ
                        self.msgs.remove(m)
                        self.dlq.send({**m.body, "_orig_id": m.id})
                        return None
                    return m
            return None
 
    def delete(self, msg_id: str) -> bool:
        with self.lock:
            for m in list(self.msgs):
                if m.id == msg_id:
                    self.msgs.remove(m)
                    return True
            return False
 
    def depth(self) -> int:
        with self.lock:
            return len(self.msgs)
 
 
# --- Idempotent consumer ---
 
class IdempotentProcessor:
    """Simulates a downstream side effect (e.g. sending an email or
    inserting into a database). Records the set of message ids it has
    already handled so duplicates are safely ignored."""
    def __init__(self, name: str, fail_rate: float = 0.0):
        self.name = name
        self.seen: set[str] = set()
        self.processed: list[str] = []
        self.fail_rate = fail_rate
        self.lock = threading.Lock()
 
    def handle(self, msg: Message) -> bool:
        # Simulate transient failure BEFORE the dedup check — so we get
        # legitimate redeliveries.
        if random.random() < self.fail_rate:
            return False
        with self.lock:
            key = msg.body.get("idempotency_key", msg.id)
            if key in self.seen:
                # Duplicate — already processed. Ack silently.
                return True
            self.seen.add(key)
            self.processed.append(key)
            return True
 
 
# --- Worker + producer ---
 
def worker(q: Queue, proc: IdempotentProcessor, stop: threading.Event) -> None:
    while not stop.is_set():
        m = q.receive()
        if m is None:
            time.sleep(0.05)
            continue
        # Simulate variable processing time.
        time.sleep(random.uniform(0.05, 0.2))
        if proc.handle(m):
            q.delete(m.id)  # ack
        # else: don't delete, message becomes visible again after timeout
 
 
def producer(q: Queue, n: int) -> None:
    for i in range(n):
        q.send({"order_id": f"ord-{i:04d}",
                "idempotency_key": f"ord-{i:04d}",  # idempotency by order_id
                "amount": random.randint(10, 500)})
        time.sleep(0.02)
 
 
def main() -> None:
    dlq = Queue("orders-dlq", max_receives=999)
    q = Queue("orders", visibility_timeout=0.3, max_receives=3, dlq=dlq)
    proc = IdempotentProcessor("email-sender", fail_rate=0.3)
 
    stop = threading.Event()
    workers = [threading.Thread(target=worker, args=(q, proc, stop),
                                daemon=True) for _ in range(3)]
    for w in workers:
        w.start()
 
    print("producing 50 messages...")
    producer(q, 50)
 
    # Let workers drain
    print("draining...")
    while q.depth() > 0:
        time.sleep(0.1)
    time.sleep(1.0)  # let in-flight retries settle
    stop.set()
 
    print(f"\nqueue depth: {q.depth()}")
    print(f"DLQ depth:   {dlq.depth()}")
    print(f"unique processed by consumer: {len(proc.seen)}")
    print(f"total handle() calls: {len(proc.processed)}")
    print(f"(handle() calls > 50 = duplicates suppressed by idempotency)")
 
 
if __name__ == "__main__":
    main()