Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
app/backend/tests/flow/test_queue.py
T
stroblmeandClaude Opus 5 1069247085 Coalesce the event bus, and fix the socket that ended on a client frame 
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
2026-08-29 20:08:50 +02:00

629 lines
20 KiB
Python
Raw Permalink Blame History

"""The work queue, and what the execution service does with it."""
import threading
import time
from fluksio.flow import executor
from fluksio.flow import queue as queue_module
from fluksio.flow.executor import ExecutionService
from fluksio.flow.messages import DType, MessageSpec
from fluksio.flow.nodes import Node
from fluksio.flow.pipeline import Pipeline
from fluksio.flow.queue import MemoryWorkQueue, WorkItem
from fluksio.flow.state import MemoryState
def test_items_come_back_in_the_order_they_went_in():
queue = MemoryWorkQueue()
for i in range(3):
queue.add(WorkItem(kind="cascade", node=f"f.n{i}", flow="f"))
claimed = queue.claim(10, 10)
assert [item.node for item in claimed] == ["f.n0", "f.n1", "f.n2"]
# Every item gets an id, which is what idempotency markers hang off.
assert all(item.entry_id for item in claimed)
def test_claiming_an_empty_queue_waits_and_gives_up():
queue = MemoryWorkQueue()
started = time.monotonic()
assert queue.claim(1, 50) == []
assert time.monotonic() - started >= 0.04
def test_a_delayed_item_stays_put_until_it_is_due():
queue = MemoryWorkQueue()
queue.add_delayed(WorkItem(kind="cascade", node="f.n", flow="f"), time.time() + 60)
assert queue.claim(1, 10) == []
assert queue.move_due(time.time()) == 0
assert queue.move_due(time.time() + 61) == 1
assert [i.node for i in queue.claim(1, 10)] == ["f.n"]
def test_the_queue_says_when_the_soonest_delayed_item_is_due():
"""What lets the engine sleep to a deadline rather than poll for it."""
queue = MemoryWorkQueue()
assert queue.next_due() is None
queue.add_delayed(WorkItem(kind="cascade", node="f.late", flow="f"), 500.0)
queue.add_delayed(WorkItem(kind="cascade", node="f.soon", flow="f"), 100.0)
assert queue.next_due() == 100.0
queue.move_due(200.0)
assert queue.next_due() == 500.0
def test_scheduling_says_a_deadline_moved():
"""A delay scheduled for 200ms must cut short a sleep to the next second."""
queue = MemoryWorkQueue()
woken = threading.Event()
queue.on_delayed = woken.set
queue.add_delayed(WorkItem(kind="cascade", node="f.n", flow="f"), time.time() + 0.2)
assert woken.is_set()
def test_a_due_timer_is_claimed_before_the_backlog():
"""Lateness is what a timer is measured by; queued work waits on no clock."""
queue = MemoryWorkQueue()
for i in range(3):
queue.add(WorkItem(kind="cascade", node=f"f.queued{i}", flow="f"))
queue.add_delayed(WorkItem(kind="cascade", node="f.timer", flow="f"), 100.0)
queue.move_due(200.0)
assert [i.node for i in queue.claim(10, 10)][0] == "f.timer"
def test_a_delay_fires_without_waiting_out_the_poll():
"""The whole point: a 200ms delay is 200ms late, not up to a second."""
queue = MemoryWorkQueue()
service = ExecutionService(queue)
dispatched: list[float] = []
service._dispatch = lambda item: dispatched.append(time.monotonic()) # type: ignore[method-assign]
service.start()
try:
started = time.monotonic()
queue.add_delayed(
WorkItem(kind="cascade", node="f.n", flow="f"), time.time() + 0.2
)
# Claimed by the consumer thread, which is what calls _dispatch.
deadline = time.monotonic() + 2.0
while not dispatched and time.monotonic() < deadline:
time.sleep(0.01)
finally:
service.stop()
assert dispatched, "the deferred item never ran"
late = dispatched[0] - started - 0.2
# A fixed one-second poll made this up to 1.0s; the budget is generous
# because CI is not a real-time machine.
assert late < 0.3, f"fired {late:.3f}s late"
def test_parked_work_comes_back_oldest_first():
queue = MemoryWorkQueue()
for i in range(3):
queue.park("heating", WorkItem(kind="cascade", node=f"f.n{i}", flow="heating"))
assert [i.node for i in queue.unpark("heating")] == ["f.n0", "f.n1", "f.n2"]
# Unparking empties it, so a second resume does not replay the same work.
assert queue.unpark("heating") == []
def test_a_deleted_flow_leaves_nothing_parked():
queue = MemoryWorkQueue()
queue.park("gone", WorkItem(kind="cascade", node="gone.n", flow="gone"))
queue.clear_flow("gone")
assert queue.unpark("gone") == []
def test_claimed_work_counts_as_in_flight_until_it_is_acknowledged():
"""The health tile's "in flight" reads zero without this."""
queue = MemoryWorkQueue()
queue.add(WorkItem(kind="cascade", node="f.n", flow="f"))
assert queue.stats()["pending"] == 0
# The stream length was never a backlog, so the key is gone from both queues.
assert "depth" not in queue.stats()
(item,) = queue.claim(1, 10)
assert queue.stats()["pending"] == 1
queue.ack(item)
assert queue.stats()["pending"] == 0
def test_work_waiting_to_be_claimed_is_the_backlog():
"""`pending` is what is running; an engine hours behind reports it as idle."""
queue = MemoryWorkQueue()
queue.add(WorkItem(kind="cascade", node="f.n", flow="f"))
assert queue.stats()["backlog"] == 1
assert queue.stats()["pending"] == 0
(item,) = queue.claim(1, 10)
assert queue.stats()["backlog"] == 0
assert queue.stats()["pending"] == 1
queue.ack(item)
assert queue.stats()["backlog"] == 0
def test_a_sustained_backlog_says_the_engine_is_behind():
"""A flow enqueuing faster than the pool drains produced no signal at all."""
events: list[dict] = []
queue = MemoryWorkQueue()
service = ExecutionService(queue)
service._publish = events.append # type: ignore[method-assign]
for _ in range(executor.BACKLOG_DEGRADED):
queue.add(WorkItem(kind="cascade", node="f.n", flow="f"))
for _ in range(executor.BACKLOG_STRIKES - 1):
service._check_backlog()
assert events == []
service._check_backlog()
assert [e["type"] for e in events] == ["engine_degraded"]
assert service.stats()["behind"] is True
# Said once, not once every five seconds for as long as it lasts.
service._check_backlog()
assert len(events) == 1
# And a drained queue clears it, so the next backlog is announced again.
queue.claim(executor.BACKLOG_DEGRADED, 10)
service._check_backlog()
assert service.stats()["behind"] is False
def _pipeline_with_a_consumer() -> tuple[Pipeline, Node, MemoryState, list]:
"""A source whose message a consumer records."""
seen: list[float] = []
def consume(reading, params):
seen.append(reading)
return {"doubled": reading * 2}
source = Node(
f=lambda params: None,
provides=[MessageSpec(name="reading", port="reading", dtype=DType.FLOAT)],
name="source",
)
consumer = Node(
f=consume,
requires=[MessageSpec(name="reading", port="reading", dtype=DType.FLOAT)],
provides=[MessageSpec(name="doubled", port="doubled", dtype=DType.FLOAT)],
name="consumer",
)
source.assign_flow("f", "source")
consumer.assign_flow("f", "consumer")
state = MemoryState()
queue = MemoryWorkQueue()
pipeline = Pipeline(nodes=[source, consumer], state=state, work_queue=queue)
return pipeline, source, state, seen
def test_a_trigger_is_journaled_rather_than_run_on_the_spot():
pipeline, source, state, seen = _pipeline_with_a_consumer()
source.inject({"reading": 3.0})
# Nothing ran yet: the value is in the queue, not in state.
assert seen == []
assert "f.reading" not in state
service = ExecutionService(pipeline._queue)
service.bind(pipeline)
for item in pipeline._queue.claim(10, 10):
service._run_item(item)
assert seen == [3.0]
assert state["f.doubled"] == 6.0
def test_work_for_a_paused_flow_is_held_and_released_on_resume():
pipeline, source, state, seen = _pipeline_with_a_consumer()
service = ExecutionService(pipeline._queue)
service.bind(pipeline)
pipeline.pause("f")
source.inject({"reading": 1.0})
for item in pipeline._queue.claim(10, 10):
service._run_item(item)
assert seen == []
pipeline.resume("f")
for item in pipeline._queue.unpark("f"):
pipeline._queue.add(item)
for item in pipeline._queue.claim(10, 10):
service._run_item(item)
assert seen == [1.0]
def test_a_step_runs_one_held_item_and_leaves_the_flow_paused():
pipeline, source, _state, seen = _pipeline_with_a_consumer()
service = ExecutionService(pipeline._queue)
service.bind(pipeline)
pipeline.pause("f")
source.inject({"reading": 1.0})
source.inject({"reading": 2.0})
for item in pipeline._queue.claim(10, 10):
service._run_item(item)
assert seen == []
assert service.step("f") == "f.source"
assert seen == [1.0]
# Still paused, and the second value is still waiting for the next step.
assert pipeline.is_paused("f")
assert [i.outputs for i in pipeline._queue.unpark("f")] == [{"f.reading": 2.0}]
def test_stepping_a_flow_with_nothing_held_says_so_rather_than_failing():
pipeline, _source, _state, _seen = _pipeline_with_a_consumer()
service = ExecutionService(pipeline._queue)
service.bind(pipeline)
pipeline.pause("f")
assert service.step("f") is None
def test_work_for_a_stopped_flow_is_dropped():
pipeline, source, state, seen = _pipeline_with_a_consumer()
stopped = Pipeline(
nodes=pipeline.nodes,
state=pipeline.state,
work_queue=pipeline._queue,
disabled_flows={"f"},
)
service = ExecutionService(pipeline._queue)
service.bind(stopped)
# Reaching the queue at all takes a direct add: trigger drops it earlier.
stopped._queue.add(
WorkItem(kind="cascade", node="f.source", flow="f", outputs={"f.reading": 1.0})
)
for item in stopped._queue.claim(10, 10):
service._run_item(item)
assert seen == []
def test_an_item_that_keeps_coming_back_is_dead_lettered():
pipeline, _source, _state, seen = _pipeline_with_a_consumer()
service = ExecutionService(pipeline._queue)
service.bind(pipeline)
item = WorkItem(
kind="cascade",
node="f.source",
flow="f",
outputs={"f.reading": 1.0},
deliveries=4,
)
service._run_item(item)
# Given up on rather than run again, so a poison item cannot loop forever.
assert seen == []
def test_an_item_for_a_node_that_no_longer_exists_is_dropped():
pipeline, _source, _state, seen = _pipeline_with_a_consumer()
service = ExecutionService(pipeline._queue)
service.bind(pipeline)
service._run_item(WorkItem(kind="cascade", node="f.removed", flow="f"))
assert seen == []
def test_a_replayed_item_does_not_repeat_a_side_effect():
"""At-least-once delivery must not mean two of the same outgoing request."""
calls: list[float] = []
def send(reading, params):
calls.append(reading)
return None
source = Node(
f=lambda params: None,
provides=[MessageSpec(name="reading", port="reading", dtype=DType.FLOAT)],
name="source",
)
class SendingNode(Node):
"""Stands in for the built-ins that reach outside."""
idempotent = False
sender = SendingNode(
f=send,
requires=[MessageSpec(name="reading", port="reading", dtype=DType.FLOAT)],
name="sender",
)
source.assign_flow("f", "source")
sender.assign_flow("f", "sender")
queue = MemoryWorkQueue()
pipeline = Pipeline(nodes=[source, sender], state=MemoryState(), work_queue=queue)
service = ExecutionService(queue)
service.bind(pipeline)
source.inject({"reading": 5.0})
(item,) = queue.claim(10, 10)
service._run_item(item)
assert calls == [5.0]
# The same item again, as a reaper would hand it back after a crash.
item.deliveries = 2
service._run_item(item)
assert calls == [5.0]
# -----------------------------------------------------------------------------
# Telling the queue that a long node is working, not lost
#
# What marks an item abandoned is nobody touching it. A node with no timeout
# may run for hours, so the engine holding it says so on a timer — and an
# engine that died says nothing, which is the distinction the reaper needs.
# -----------------------------------------------------------------------------
class RecordingQueue(MemoryWorkQueue):
"""A memory queue that writes down what it was asked to hold on to."""
def __init__(self) -> None:
super().__init__()
self.touched: list[list[str]] = []
def touch(self, entry_ids: list[str]) -> None:
self.touched.append(list(entry_ids))
def test_work_in_flight_is_touched_until_it_finishes(monkeypatch):
monkeypatch.setattr(executor, "TOUCH_INTERVAL_S", 0.0)
monkeypatch.setattr(executor, "DELAYED_INTERVAL_S", 0.05)
running = threading.Event()
release = threading.Event()
def slow(reading, params):
running.set()
release.wait(5)
return {"doubled": reading * 2}
source = Node(
f=lambda params: None,
provides=[MessageSpec(name="reading", dtype=DType.FLOAT)],
name="source",
)
consumer = Node(
f=slow,
requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
provides=[MessageSpec(name="doubled", dtype=DType.FLOAT)],
name="consumer",
)
source.assign_flow("f", "source")
consumer.assign_flow("f", "consumer")
queue = RecordingQueue()
pipeline = Pipeline(nodes=[source, consumer], state=MemoryState(), work_queue=queue)
service = ExecutionService(queue)
service.bind(pipeline)
service.start()
try:
source.inject({"reading": 3.0})
assert running.wait(5)
# Give the timer a couple of passes while the node is still in there.
time.sleep(0.2)
held = [ids for ids in queue.touched if ids]
assert held, "a running item was never touched"
release.set()
time.sleep(0.3)
# Once it is done it is acknowledged, so there is nothing to hold.
assert queue.touched[-1] == []
finally:
release.set()
service.stop()
def test_no_more_is_claimed_than_the_pool_can_run():
"""A backlog belongs in the queue, not inside the process.
Claiming ahead of the pool used to leave every waiting item counted as a
busy cascade and holding its journal entry open, so four cascade threads
reported hundreds in flight on an engine that was merely behind.
"""
release = threading.Event()
def slow(reading, params):
release.wait(5)
return {"doubled": reading * 2}
source = Node(
f=lambda params: None,
provides=[MessageSpec(name="reading", dtype=DType.FLOAT)],
name="source",
)
consumer = Node(
f=slow,
requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
provides=[MessageSpec(name="doubled", dtype=DType.FLOAT)],
name="consumer",
)
source.assign_flow("f", "source")
consumer.assign_flow("f", "consumer")
queue = MemoryWorkQueue()
pipeline = Pipeline(nodes=[source, consumer], state=MemoryState(), work_queue=queue)
service = ExecutionService(queue)
service.bind(pipeline)
service.start()
try:
for i in range(40):
queue.add(
WorkItem(
kind="cascade",
node="f.source",
flow="f",
outputs={"f.reading": float(i)},
)
)
time.sleep(0.5)
stats = service.stats()
assert stats["cascades_busy"] <= service.max_cascades
# And the journal entries of what is only waiting are still free.
assert stats["pending"] <= service.max_cascades
# Waiting is not idle: the rest of the forty is the backlog.
assert stats["backlog"] >= 30
finally:
release.set()
service.stop()
# -----------------------------------------------------------------------------
# Emissions: values a node publishes while it is still running
# -----------------------------------------------------------------------------
def _emitting_pipeline() -> tuple[Pipeline, Node, MemoryState, list]:
"""A source with a streaming port, and a consumer that records every value."""
seen: list[float] = []
def consume(chunk, params):
seen.append(chunk)
return None
source = Node(
f=lambda params: None,
provides=[
MessageSpec(name="chunk", port="chunk", dtype=DType.FLOAT, stream=True)
],
name="source",
)
consumer = Node(
f=consume,
requires=[MessageSpec(name="chunk", port="chunk", dtype=DType.FLOAT)],
name="consumer",
)
source.assign_flow("f", "source")
consumer.assign_flow("f", "consumer")
state = MemoryState()
queue = MemoryWorkQueue()
pipeline = Pipeline(nodes=[source, consumer], state=state, work_queue=queue)
return pipeline, source, state, seen
def test_every_emitted_chunk_reaches_the_consumer():
"""A consumer slower than its producer must not skip what it missed.
Reading state instead would give whichever chunk is newest by the time the
item runs — fine for a temperature, lossy for a second of speech.
"""
pipeline, source, state, seen = _emitting_pipeline()
service = ExecutionService(pipeline._queue)
service.bind(pipeline)
# Both emitted before either item is claimed, so state has moved on.
pipeline.publish_emission(source, {"f.chunk": 1.0})
pipeline.publish_emission(source, {"f.chunk": 2.0})
assert state["f.chunk"] == 2.0
for item in pipeline._queue.claim(10, 10):
service._run_item(item)
assert seen == [1.0, 2.0]
def test_a_delivered_emission_does_not_write_state_a_second_time():
"""The value in state stays the newest one, whenever an item is claimed."""
pipeline, source, state, seen = _emitting_pipeline()
service = ExecutionService(pipeline._queue)
service.bind(pipeline)
pipeline.publish_emission(source, {"f.chunk": 1.0})
# What the node returned at the end, after the emission it made on the way.
pipeline.apply_outputs(source, {"f.chunk": 9.0})
for item in pipeline._queue.claim(10, 10):
service._run_item(item)
assert seen[0] == 1.0
# Re-applying the carried chunk here is what would undo the final value.
assert state["f.chunk"] == 9.0
def test_a_throttled_emission_wakes_nothing():
pipeline, source, _state, seen = _emitting_pipeline()
source.provides["f.chunk"] = MessageSpec(
name="f.chunk", port="chunk", dtype=DType.FLOAT, stream=True, interval=60
)
service = ExecutionService(pipeline._queue)
service.bind(pipeline)
pipeline.publish_emission(source, {"f.chunk": 1.0})
pipeline.publish_emission(source, {"f.chunk": 2.0})
for item in pipeline._queue.claim(10, 10):
service._run_item(item)
# The first is let through; the second is held by the interval, and a value
# nothing published is nothing to wake on.
assert seen == [1.0]
def test_a_due_timer_is_claimable_while_the_backlog_is_not():
"""What a saturated engine uses to get a motor's stop out.
Every cascade slot held by a long node used to mean the due lane was not
merely behind the backlog but unread, because the capacity gate sits in
front of the claim.
"""
queue = MemoryWorkQueue()
queue.add(WorkItem(kind="cascade", node="f.backlog", flow="f"))
queue.add_delayed(
WorkItem(kind="cascade", node="f.timer", flow="f"), time.time() - 1
)
assert queue.move_due(time.time()) == 1
claimed = queue.claim(10, 10, due_only=True)
assert [item.node for item in claimed] == ["f.timer"]
# And the backlog is still there for whoever comes next.
assert [item.node for item in queue.claim(10, 10)] == ["f.backlog"]
def test_a_due_only_claim_of_nothing_does_not_take_the_backlog():
queue = MemoryWorkQueue()
queue.add(WorkItem(kind="cascade", node="f.backlog", flow="f"))
assert queue.claim(10, 10, due_only=True) == []
assert len(queue.claim(10, 10)) == 1
def test_done_markers_do_not_grow_without_bound():
"""The Redis side expires them after an hour; this one is capped by count."""
queue = MemoryWorkQueue()
for i in range(queue_module.DONE_MARKERS + 500):
queue.mark_done(f"entry-{i}", "f.node")
assert len(queue._done) == queue_module.DONE_MARKERS
# The newest are what redelivery would ask about.
assert queue.was_done(f"entry-{queue_module.DONE_MARKERS + 499}", "f.node")
assert not queue.was_done("entry-0", "f.node")
Reference in New Issue View Git Blame Copy Permalink