116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
"""In-memory decision-event broker for the dashboard's live view.
|
|
|
|
This module is deliberately small and dependency-free: it keeps a bounded
|
|
ring buffer of recent routing decisions and lets asyncio-based subscribers
|
|
receive them in near real time. Every decision inserted via
|
|
``dispatcher.persist_route_decision`` is also published here, so the TUI can
|
|
show a live feed without polling ``/metrics``.
|
|
|
|
The broker is intentionally **not** durable: if no dashboard is connected, the
|
|
ring is all that survives. The durable source of truth remains the
|
|
``route_decisions`` SQLite table; this is just a volatile fan-out helper.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import queue
|
|
import threading
|
|
from collections import deque
|
|
from typing import Any, Final
|
|
|
|
DEFAULT_BUFFER_SIZE: Final[int] = 100
|
|
DEFAULT_QUEUE_SIZE: Final[int] = 100
|
|
|
|
# Sentinel value to signal a subscriber that has been evicted
|
|
# (its queue was full). The SSE stream generator detects this and
|
|
# terminates so the TUI can reconnect.
|
|
_EVICTED = object()
|
|
|
|
_buffer: deque[dict[str, Any]] = deque(maxlen=DEFAULT_BUFFER_SIZE)
|
|
_subscribers: set[queue.Queue[dict[str, Any]]] = set()
|
|
_subscribers_lock = threading.Lock()
|
|
|
|
|
|
def publish_decision(decision: dict[str, Any]) -> None:
|
|
"""Append ``decision`` to the ring buffer and fan it out to subscribers.
|
|
|
|
If a subscriber's queue is full or dead it is silently removed, because
|
|
the broker must never fail the caller (a routing decision should never
|
|
be slowed or broken by a dashboard consumer).
|
|
|
|
When a subscriber is evicted for being full, a sentinel value is written
|
|
to its queue so the consuming SSE generator can detect the closure and
|
|
terminate rather than blocking forever.
|
|
"""
|
|
_buffer.append(decision)
|
|
|
|
dead: set[queue.Queue[dict[str, Any]]] = set()
|
|
with _subscribers_lock:
|
|
for subscriber in _subscribers:
|
|
try:
|
|
subscriber.put_nowait(decision)
|
|
except queue.Full:
|
|
dead.add(subscriber)
|
|
_subscribers.difference_update(dead)
|
|
|
|
# Signal evicted subscribers so their SSE streams terminate.
|
|
# The decision was never queued (queue was full), so place the sentinel
|
|
# directly — it will displace the old item but is enough to wake up the
|
|
# consumer and signal that the connection should be re-established.
|
|
for subscriber in dead:
|
|
try:
|
|
subscriber.put_nowait(_EVICTED)
|
|
except queue.Full:
|
|
# Queue is still full — replace the old item with the sentinel.
|
|
# Put non-blocking puts will never block if we're here, but the
|
|
# queue is genuinely full so we just discard the old item.
|
|
try:
|
|
subscriber.get_nowait() # discard the oldest item
|
|
subscriber.put_nowait(_EVICTED)
|
|
except queue.Empty:
|
|
pass
|
|
|
|
|
|
def subscribe(
|
|
replay: bool = True,
|
|
max_queue: int = DEFAULT_QUEUE_SIZE,
|
|
) -> queue.Queue[dict[str, Any]]:
|
|
"""Create a thread-safe subscriber queue.
|
|
|
|
By default the current ring-buffer contents are preloaded so a new
|
|
connection immediately sees recent decisions. The caller must arrange to
|
|
remove the queue with :func:`unsubscribe` when it disconnects.
|
|
"""
|
|
new_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=max_queue)
|
|
if replay:
|
|
for decision in _buffer:
|
|
try:
|
|
new_queue.put_nowait(decision)
|
|
except queue.Full:
|
|
break
|
|
with _subscribers_lock:
|
|
_subscribers.add(new_queue)
|
|
return new_queue
|
|
|
|
|
|
def unsubscribe(subscriber: queue.Queue[dict[str, Any]]) -> None:
|
|
"""Remove ``subscriber`` from the fan-out set."""
|
|
with _subscribers_lock:
|
|
_subscribers.discard(subscriber)
|
|
|
|
|
|
def recent_decisions(limit: int = DEFAULT_BUFFER_SIZE) -> list[dict[str, Any]]:
|
|
"""Return up to ``limit`` items from the ring buffer, newest last."""
|
|
return list(_buffer)[-limit:]
|
|
|
|
|
|
def clear() -> None:
|
|
"""Drop every buffered decision and subscriber.
|
|
|
|
Used by tests to ensure a clean broker state; production code should
|
|
almost never call this.
|
|
"""
|
|
with _subscribers_lock:
|
|
_subscribers.clear()
|
|
_buffer.clear()
|