feat(pubsub): EventBus with drop-oldest overflow, per-kind fan-out

backend/src/cyclone/pubsub.py

@@ -0,0 +1,58 @@
+"""In-process pub/sub for live-tail event streaming.
+
+``EventBus`` is a tiny, async-first fan-out broker. Publishers call
+``publish(kind, payload)`` and never block; if a subscriber's per-kind
+queue is full, the oldest event is dropped so the slow consumer cannot
+stall the producer. Subscribers receive events through an async
+iterator that yields ``{**payload, "_kind": kind}``.
+
+The bus is intentionally not thread-safe: it is designed to run on a
+single asyncio event loop, which matches the FastAPI/uvicorn
+deployment model used elsewhere in this project.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncIterator
+
+
+class EventBus:
+    """Per-kind fan-out bus with drop-oldest overflow per subscriber."""
+
+    def __init__(self, max_queue_size: int = 256) -> None:
+        self._max_queue_size = max_queue_size
+        self._subscribers: dict[str, list[asyncio.Queue[dict]]] = {}
+
+    async def publish(self, kind: str, payload: dict) -> None:
+        """Fan ``payload`` out to every subscriber of ``kind``.
+
+        Never blocks. If a subscriber queue is full, the oldest queued
+        event is discarded to make room.
+        """
+        event = {**payload, "_kind": kind}
+        for queue in list(self._subscribers.get(kind, ())):
+            self._enqueue_or_drop_oldest(queue, event)
+
+    def _enqueue_or_drop_oldest(
+        self, queue: asyncio.Queue[dict], event: dict
+    ) -> None:
+        try:
+            queue.put_nowait(event)
+        except asyncio.QueueFull:
+            # Drop the oldest event to make room, then enqueue.
+            queue.get_nowait()
+            queue.task_done()
+            queue.put_nowait(event)
+
+    def subscribe(self, kinds: list[str]) -> AsyncIterator[dict]:
+        """Return an async iterator delivering events for any of ``kinds``."""
+        queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=self._max_queue_size)
+        for kind in kinds:
+            self._subscribers.setdefault(kind, []).append(queue)
+        return self._iterator(queue)
+
+    async def _iterator(self, queue: asyncio.Queue[dict]) -> AsyncIterator[dict]:
+        while True:
+            yield await queue.get()
+            queue.task_done()

backend/tests/test_pubsub.py

@@ -0,0 +1,99 @@
+"""Tests for the in-process EventBus.
+
+These exercise the drop-oldest overflow strategy, per-kind fan-out, and
+the non-blocking publish contract. They are async because the bus
+exposes an async iterator for subscribers.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+
+import pytest
+
+from cyclone.pubsub import EventBus
+
+
+@pytest.mark.asyncio
+async def test_publish_no_subscribers_is_noop():
+    bus = EventBus()
+    # No subscribers at all — must not raise.
+    await bus.publish("x", {"a": 1})
+
+
+@pytest.mark.asyncio
+async def test_publish_to_n_subscribers_delivers_to_all():
+    bus = EventBus()
+    queues = [bus.subscribe(["claim"]) for _ in range(3)]
+
+    await bus.publish("claim", {"id": "C1"})
+
+    received = [await asyncio.wait_for(q.__anext__(), timeout=0.5) for q in queues]
+    for event in received:
+        assert event == {"id": "C1", "_kind": "claim"}
+
+
+@pytest.mark.asyncio
+async def test_subscriber_only_receives_matching_kinds():
+    bus = EventBus()
+    queue = bus.subscribe(["a"])
+
+    await bus.publish("b", {"n": 1})
+    await bus.publish("a", {"n": 2})
+
+    event = await asyncio.wait_for(queue.__anext__(), timeout=0.5)
+    assert event == {"n": 2, "_kind": "a"}
+
+
+@pytest.mark.asyncio
+async def test_slow_subscriber_does_not_block_publish():
+    bus = EventBus(max_queue_size=2)
+    queue = bus.subscribe(["e"])
+
+    async def emit():
+        for i in range(3):
+            await bus.publish("e", {"i": i})
+
+    start = time.monotonic()
+    await emit()
+    elapsed = time.monotonic() - start
+
+    # Publishing must not block even though the subscriber queue is full
+    # after the second event; the drop-oldest policy keeps publishes non-blocking.
+    assert elapsed < 0.05, f"publish blocked for {elapsed:.3f}s"
+
+    # The slow subscriber retains only the last two events.
+    seen = []
+    for _ in range(2):
+        seen.append(await asyncio.wait_for(queue.__anext__(), timeout=0.5))
+    assert [e["i"] for e in seen] == [1, 2]
+
+
+@pytest.mark.asyncio
+async def test_queue_overflow_drops_oldest():
+    bus = EventBus(max_queue_size=2)
+    queue = bus.subscribe(["e"])
+
+    for i in range(3):
+        await bus.publish("e", {"i": i})
+
+    seen = []
+    for _ in range(2):
+        seen.append(await asyncio.wait_for(queue.__anext__(), timeout=0.5))
+    assert [e["i"] for e in seen] == [1, 2]
+
+
+@pytest.mark.asyncio
+async def test_multiple_concurrent_subscribers_each_get_all_events():
+    bus = EventBus()
+    queues = [bus.subscribe(["claim"]) for _ in range(2)]
+
+    for i in range(5):
+        await bus.publish("claim", {"i": i})
+
+    for q in queues:
+        seen = []
+        for _ in range(5):
+            seen.append(await asyncio.wait_for(q.__anext__(), timeout=0.5))
+        assert [e["i"] for e in seen] == [0, 1, 2, 3, 4]