Socket snapshot carries the nodes waiting for a machine, so a reconnecting browser restores their queued dots
This commit is contained in:
@@ -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 [],
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user