stroblmeandClaude Opus 5
1069247085
A three-node cascade publishes 13-16 events and each one crossed to the
event loop on its own. They are one `call_soon_threadsafe` now — whatever
was published between two turns of the loop goes over together — and every
subscriber still receives every event, oldest still dropped first when one
falls behind.
The socket end of the same path:
- **any frame from the client ended its stream.** `receive_text` was
awaited once, outside the loop, so a keepalive — or anything else a
client decided to say — satisfied it and was read as the client going
away. It is recreated per iteration; only a disconnect ends the stream.
- events go out in one frame per wave (`{"type": "batch", "events": [...]}`,
capped at 64), serialised once with orjson rather than per client with
the stdlib's `json.dumps` through `send_json`. The client unpacks a batch
and still understands single frames, so an older engine behind a newer
bundle keeps working.
- authenticating and building the snapshot happen on a thread. Both were on
the event loop: one is a database round trip, the other reads the whole
of state, per connect and again per `dashboard_changed` per panel.
`Pipeline.values()` — what that snapshot is — no longer SCANs the whole
Redis namespace. It scanned five bookkeeping keys for every message to find
the messages; `RedisState` keeps a set of the names beside them and answers
from it. Maintained wherever a message is written, so a seeded value or a
deleted flow keeps it exact.
On the client, while in the same file:
- a `node_health` event invalidates the flow's detail. The canvas draws
health from the server-derived `issues`, so a node going down or
recovering only showed on mount, navigation or a rebuild. The store had
a health map of its own that nothing ever read; it and `useNodeHealth`
are gone rather than wired up, since the server's view is the one the
canvas already uses.
- a reconnect invalidates the five key families this socket feeds instead
of the entire cache, and the backoff is jittered. The usual reason a
socket dropped is the engine restarting, so every tab and every wall
panel refetched everything, together, at the moment it was least able to
answer.
- a frame that will not parse costs the frame, not the connection. It was
the one unguarded `JSON.parse` in the app; an exception there escaped to
`window.onerror` and left whatever it had already applied behind.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M6hPWS6YEbT1P8LxhhFb2T
110 lines
4.3 KiB
Python
110 lines
4.3 KiB
Python
"""Event bus bridging the engine's worker threads to async subscribers.
|
|
|
|
Nodes execute in a thread pool; websocket clients live on the event loop.
|
|
Publishers are therefore thread-safe and never block: a subscriber that cannot
|
|
keep up loses its oldest queued events rather than stalling the engine.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import threading
|
|
from collections import deque
|
|
from collections.abc import AsyncIterator
|
|
from contextlib import asynccontextmanager
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
QUEUE_SIZE = 256
|
|
LOG_HISTORY = 400
|
|
|
|
|
|
class EventBus:
|
|
"""Fan-out of engine events to any number of async subscribers."""
|
|
|
|
def __init__(self) -> None:
|
|
self._loop: asyncio.AbstractEventLoop | None = None
|
|
self._subscribers: set[asyncio.Queue[dict[str, Any]]] = set()
|
|
# Events published since the last flush was scheduled. A cascade
|
|
# publishes 13-16 of them and each used to cost its own
|
|
# `call_soon_threadsafe`; they now cost one between two turns of the
|
|
# loop, which is what a wave is.
|
|
self._pending: list[dict[str, Any]] = []
|
|
self._pending_lock = threading.Lock()
|
|
self._flush_scheduled = False
|
|
# Kept whether or not anyone is listening, so opening the log panel
|
|
# shows what just happened rather than an empty box.
|
|
self.recent_logs: deque[dict[str, Any]] = deque(maxlen=LOG_HISTORY)
|
|
# How often each qualified node has emitted, for the same reason: a
|
|
# client that reconnects between two pages would otherwise start from
|
|
# zero and draw a busy graph as idle. A session count, not a metric —
|
|
# the rollups in `metrics.py` are what survives a restart.
|
|
self.emits: dict[str, int] = {}
|
|
|
|
def bind(self, loop: asyncio.AbstractEventLoop) -> None:
|
|
"""Attach the bus to the running event loop (called once at startup)."""
|
|
self._loop = loop
|
|
|
|
def publish(self, event: dict[str, Any]) -> None:
|
|
"""Publish an event from any thread."""
|
|
kind = event.get("type")
|
|
if kind == "node_log":
|
|
# deque.append is atomic, so worker threads need no lock here.
|
|
self.recent_logs.append(event)
|
|
elif kind == "node_executed" and event.get("outputs"):
|
|
# The condition the canvas animates on, so both ends agree on what
|
|
# counts as an emission. Two worker threads racing here can lose an
|
|
# increment, which is a count nobody is going to miss.
|
|
node = str(event.get("node") or "")
|
|
self.emits[node] = self.emits.get(node, 0) + 1
|
|
loop = self._loop
|
|
if loop is None or not self._subscribers:
|
|
return
|
|
with self._pending_lock:
|
|
self._pending.append(event)
|
|
if self._flush_scheduled:
|
|
# A flush is already on its way and has not run yet, so it
|
|
# will find this event too.
|
|
return
|
|
self._flush_scheduled = True
|
|
try:
|
|
loop.call_soon_threadsafe(self._dispatch)
|
|
except RuntimeError:
|
|
# Loop already closed — shutting down.
|
|
with self._pending_lock:
|
|
self._pending.clear()
|
|
self._flush_scheduled = False
|
|
|
|
def _dispatch(self) -> None:
|
|
"""Hand everything published since the last turn to every subscriber."""
|
|
with self._pending_lock:
|
|
batch = self._pending
|
|
self._pending = []
|
|
self._flush_scheduled = False
|
|
if not batch:
|
|
return
|
|
for queue in self._subscribers:
|
|
for event in batch:
|
|
if queue.full():
|
|
# Drop the oldest so a slow client never blocks the engine.
|
|
try:
|
|
queue.get_nowait()
|
|
except asyncio.QueueEmpty:
|
|
pass
|
|
queue.put_nowait(event)
|
|
|
|
@asynccontextmanager
|
|
async def subscribe(self) -> AsyncIterator[asyncio.Queue[dict[str, Any]]]:
|
|
"""Yield a queue receiving every event published while subscribed."""
|
|
queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=QUEUE_SIZE)
|
|
self._subscribers.add(queue)
|
|
try:
|
|
yield queue
|
|
finally:
|
|
self._subscribers.discard(queue)
|
|
|
|
|
|
event_bus = EventBus()
|