diff --git a/backend/src/cyclone/pubsub.py b/backend/src/cyclone/pubsub.py new file mode 100644 index 0000000..7047e84 --- /dev/null +++ b/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() diff --git a/backend/tests/test_pubsub.py b/backend/tests/test_pubsub.py new file mode 100644 index 0000000..23923b2 --- /dev/null +++ b/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]