From d423c47ca612c2adb82acae7c08db7692dcfb16b Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 14 Oct 2026 23:31:54 +0200 Subject: [PATCH] Socket snapshot carries the nodes waiting for a machine, so a reconnecting browser restores their queued dots --- backend/fluksio/api/routes/flows.py | 4 ++++ backend/tests/flow/test_placement.py | 20 +++++++++++++++++++ frontend/src/components/Flow/useFlowSocket.ts | 10 ++++++++++ 3 files changed, 34 insertions(+) diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index d4f2a6f..2150ba8 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -907,6 +907,8 @@ def snapshot_payload( two have to agree, so they build the message here rather than each their own. ``emits`` is what the bus counted while nobody was listening — a client that reconnects between two pages would otherwise start from zero. + ``queued`` is the nodes waiting for a machine, for the same reason: the bus + announces a wait once, and a client that missed it would draw the node idle. ``only`` bounds it to a set of message names, which is what a wall panel gets: the values its own dashboards draw, and none of the rest — no node @@ -925,6 +927,7 @@ def snapshot_payload( "paused": [], "logs": [], "emits": {}, + "queued": [], } return { "type": "snapshot", @@ -934,6 +937,7 @@ def snapshot_payload( "paused": controller.paused_flows(), "logs": list(event_bus.recent_logs), "emits": dict(event_bus.emits), + "queued": controller.placer.snapshot()["waiting"] if controller.placer else [], } diff --git a/backend/tests/flow/test_placement.py b/backend/tests/flow/test_placement.py index 73dabf5..27705ed 100644 --- a/backend/tests/flow/test_placement.py +++ b/backend/tests/flow/test_placement.py @@ -193,6 +193,26 @@ def test_a_node_waits_for_a_busy_machine_and_says_so(loop): assert placer.snapshot()["waiting"] == [] +def test_a_reconnecting_client_is_told_what_is_waiting(tmp_path): + """The wait is announced once, so the socket's snapshot carries it again. + + A panel is sent none of it, the same as node status. + """ + from fluksio.api.routes.flows import snapshot_payload + from fluksio.flow.controller import FlowController + from fluksio.flow.store import FlowStore + + placer = Placer(local=ResourceAccountant(cpus=2)) + placer._announce("study.b", "r1", time.monotonic(), "waiting for 2 cpu(s)") + controller = FlowController(FlowStore(tmp_path / "flows"), placer=placer) + + queued = snapshot_payload(controller)["queued"] + assert [(e["node"], e["reason"]) for e in queued] == [ + ("study.b", "waiting for 2 cpu(s)") + ] + assert snapshot_payload(controller, set())["queued"] == [] + + def test_a_worker_attaching_wakes_what_was_waiting_for_one(loop): """A node waiting inside one machine's books could not hear this.""" hub = RemoteWorkerHub() diff --git a/frontend/src/components/Flow/useFlowSocket.ts b/frontend/src/components/Flow/useFlowSocket.ts index 43b1a0f..7f5c364 100644 --- a/frontend/src/components/Flow/useFlowSocket.ts +++ b/frontend/src/components/Flow/useFlowSocket.ts @@ -43,6 +43,8 @@ type FlowEvent = logs?: LogLine[] /** Emission counts per qualified node. Absent on older instances. */ emits?: Record + /** Nodes waiting for a machine. Absent on older instances. */ + queued?: { node: string; reason: string; seconds: number }[] } | { type: "message_value" @@ -275,6 +277,14 @@ function open(url: string) { // Missing against an instance older than this bundle; the graph // then starts from zero the way it always did. liveStore.setEmits(message.emits ?? {}) + // `node_queued` is announced once per wait, so a socket that opens + // mid-wait learns of it here or draws the node idle. + for (const entry of message.queued ?? []) { + liveStore.setStatus(entry.node, { + status: "queued", + detail: entry.reason, + }) + } break case "message_value": liveStore.setValue(message.name, {