stroblmeandClaude Opus 5.5
afcbb63a56
The interval now rides the output ports, so the pipeline's own hold and flush keep what arrives inside the window instead of the node dropping it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx
294 lines
9.3 KiB
Python
294 lines
9.3 KiB
Python
"""Per-port intervals: deliver at most every x seconds."""
|
|
|
|
import time
|
|
|
|
from fluksio.flow.executor import ExecutionService
|
|
from fluksio.flow.messages import DType, MessageSpec
|
|
from fluksio.flow.nodes import DelayNode, Node
|
|
from fluksio.flow.pipeline import Pipeline
|
|
from fluksio.flow.queue import MemoryWorkQueue
|
|
|
|
# Short enough to wait out in a test, long enough not to race the engine.
|
|
WINDOW = 0.05
|
|
|
|
|
|
def spec(name: str, interval: float = 0) -> MessageSpec:
|
|
return MessageSpec(name=name, dtype=DType.FLOAT, interval=interval)
|
|
|
|
|
|
def make_node(node_id: str, f, requires=(), provides=()) -> Node:
|
|
node = Node(f=f, requires=list(requires), provides=list(provides), name=node_id)
|
|
node.assign_flow("demo", node_id)
|
|
return node
|
|
|
|
|
|
def test_a_limited_output_publishes_once_inside_its_window():
|
|
readings = iter([1.0, 2.0, 3.0])
|
|
source = make_node(
|
|
"source",
|
|
lambda params: {"temp": next(readings)},
|
|
provides=[spec("temp", interval=60)],
|
|
)
|
|
pipeline = Pipeline(nodes=[source])
|
|
|
|
pipeline.run({})
|
|
assert pipeline.state["demo.temp"] == 1.0
|
|
|
|
# Same window: the reading is taken but not published.
|
|
pipeline.run({})
|
|
assert pipeline.state["demo.temp"] == 1.0
|
|
|
|
|
|
def test_an_unlimited_output_publishes_every_time():
|
|
readings = iter([1.0, 2.0])
|
|
source = make_node(
|
|
"source",
|
|
lambda params: {"temp": next(readings)},
|
|
provides=[spec("temp")],
|
|
)
|
|
pipeline = Pipeline(nodes=[source])
|
|
|
|
pipeline.run({})
|
|
pipeline.run({})
|
|
assert pipeline.state["demo.temp"] == 2.0
|
|
|
|
|
|
def test_a_limited_input_wakes_its_node_once_inside_the_window():
|
|
seen: list[float] = []
|
|
source = make_node("source", lambda params: {"temp": 20.0}, provides=[spec("temp")])
|
|
consumer = make_node(
|
|
"consumer",
|
|
lambda temp, params: seen.append(temp),
|
|
requires=[spec("temp", interval=60)],
|
|
)
|
|
# Binding the nodes is what the pipeline is for here.
|
|
Pipeline(nodes=[source, consumer])
|
|
|
|
source.inject({"temp": 20.0})
|
|
source.inject({"temp": 21.0})
|
|
|
|
assert seen == [20.0]
|
|
|
|
|
|
def test_an_unthrottled_input_still_wakes_a_node_beside_a_throttled_one():
|
|
seen: list[tuple[float, float]] = []
|
|
fast = make_node("fast", lambda params: None, provides=[spec("quick")])
|
|
slow = make_node("slow", lambda params: None, provides=[spec("rare")])
|
|
consumer = make_node(
|
|
"consumer",
|
|
lambda quick, rare, params: seen.append((quick, rare)),
|
|
requires=[spec("quick"), spec("rare", interval=60)],
|
|
)
|
|
pipeline = Pipeline(nodes=[fast, slow, consumer])
|
|
pipeline.state["demo.rare"] = 1.0
|
|
|
|
fast.inject({"quick": 1.0})
|
|
fast.inject({"quick": 2.0})
|
|
|
|
# The throttled port holds back only itself.
|
|
assert [quick for quick, _ in seen] == [1.0, 2.0]
|
|
|
|
|
|
def _running(nodes: list[Node]) -> tuple[Pipeline, MemoryWorkQueue, ExecutionService]:
|
|
"""A pipeline with the timer the engine uses for held-back values."""
|
|
queue = MemoryWorkQueue()
|
|
pipeline = Pipeline(nodes=nodes, work_queue=queue)
|
|
service = ExecutionService(queue)
|
|
service.bind(pipeline)
|
|
return pipeline, queue, service
|
|
|
|
|
|
def _run_due(queue: MemoryWorkQueue, service: ExecutionService) -> int:
|
|
"""What the engine's timer thread does once the window has passed."""
|
|
moved = queue.move_due(time.time())
|
|
for item in queue.claim(10, 10):
|
|
service._run_item(item)
|
|
return moved
|
|
|
|
|
|
def test_a_limited_output_publishes_its_last_value_when_the_window_ends():
|
|
"""A producer going quiet must not strand the reading it held back."""
|
|
readings = iter([1.0, 2.0])
|
|
source = make_node(
|
|
"source",
|
|
lambda params: {"temp": next(readings)},
|
|
provides=[spec("temp", interval=WINDOW)],
|
|
)
|
|
pipeline, queue, service = _running([source])
|
|
|
|
pipeline.run({})
|
|
pipeline.run({})
|
|
# Inside the window, so the second reading is held rather than published.
|
|
assert pipeline.state["demo.temp"] == 1.0
|
|
|
|
time.sleep(WINDOW * 2)
|
|
assert _run_due(queue, service) == 1
|
|
|
|
assert pipeline.state["demo.temp"] == 2.0
|
|
|
|
|
|
def test_a_limited_input_wakes_its_node_when_the_window_ends():
|
|
seen: list[float] = []
|
|
source = make_node("source", lambda params: None, provides=[spec("temp")])
|
|
consumer = make_node(
|
|
"consumer",
|
|
lambda temp, params: seen.append(temp),
|
|
requires=[spec("temp", interval=WINDOW)],
|
|
)
|
|
_pipeline, queue, service = _running([source, consumer])
|
|
|
|
source.inject({"temp": 20.0}, durable=False)
|
|
source.inject({"temp": 21.0}, durable=False)
|
|
assert seen == [20.0]
|
|
|
|
time.sleep(WINDOW * 2)
|
|
assert _run_due(queue, service) == 1
|
|
|
|
# The value that arrived inside the window is delivered at the end of it.
|
|
assert seen == [20.0, 21.0]
|
|
|
|
|
|
def test_an_inject_inside_the_window_costs_no_cascade():
|
|
"""The limit used to thin the messages and not the work.
|
|
|
|
It was applied after the item came off the queue, so a subscriber told to
|
|
publish every 15s still cost a queue entry, a run record and a walk of
|
|
everything downstream for every message the broker sent.
|
|
"""
|
|
source = make_node(
|
|
"source",
|
|
lambda params: None,
|
|
provides=[spec("temp", interval=WINDOW)],
|
|
)
|
|
pipeline, queue, service = _running([source])
|
|
|
|
source.inject({"temp": 20.0})
|
|
for item in queue.claim(10, 10):
|
|
service._run_item(item)
|
|
assert pipeline.state["demo.temp"] == 20.0
|
|
|
|
# Inside the window: held where it is, with nothing journaled for it.
|
|
source.inject({"temp": 21.0})
|
|
assert queue.claim(10, 10) == []
|
|
|
|
# And the timer still lets it out at the end of the window.
|
|
time.sleep(WINDOW * 2)
|
|
assert _run_due(queue, service) == 1
|
|
assert pipeline.state["demo.temp"] == 21.0
|
|
|
|
|
|
def test_a_node_whose_inputs_did_not_change_is_not_run():
|
|
"""A cascade used to walk everything reachable, changed or not."""
|
|
seen: list[float] = []
|
|
source = make_node("source", lambda params: None, provides=[spec("raw")])
|
|
relay = make_node(
|
|
"relay",
|
|
lambda raw, params: {"level": raw},
|
|
requires=[spec("raw")],
|
|
provides=[spec("level", interval=WINDOW)],
|
|
)
|
|
watcher = make_node(
|
|
"watcher",
|
|
lambda level, params: seen.append(level),
|
|
requires=[spec("level")],
|
|
)
|
|
_pipeline, queue, service = _running([source, relay, watcher])
|
|
|
|
def deliver(value: float) -> None:
|
|
source.inject({"raw": value})
|
|
for item in queue.claim(10, 10):
|
|
service._run_item(item)
|
|
|
|
deliver(1.0)
|
|
assert seen == [1.0]
|
|
|
|
# The relay runs — its own input did change — but publishes nothing, so
|
|
# the watcher is left on the value it already has.
|
|
deliver(2.0)
|
|
assert seen == [1.0]
|
|
|
|
# The held value reaches it when the window ends.
|
|
time.sleep(WINDOW * 2)
|
|
_run_due(queue, service)
|
|
assert seen == [1.0, 2.0]
|
|
|
|
|
|
def test_a_delay_interval_sends_the_latest_value_when_the_window_ends():
|
|
"""It used to drop a value inside the window rather than hold it."""
|
|
seen: list[float] = []
|
|
source = make_node("source", lambda params: None, provides=[spec("raw")])
|
|
delay = DelayNode(
|
|
params={"interval": WINDOW},
|
|
requires=[spec("raw")],
|
|
provides=[spec("out")],
|
|
name="delay",
|
|
)
|
|
delay.assign_flow("demo", "delay")
|
|
watcher = make_node(
|
|
"watcher", lambda out, params: seen.append(out), requires=[spec("out")]
|
|
)
|
|
_pipeline, queue, service = _running([source, delay, watcher])
|
|
|
|
for value in (1.0, 2.0, 3.0):
|
|
source.inject({"raw": value})
|
|
for item in queue.claim(10, 10):
|
|
service._run_item(item)
|
|
assert seen == [1.0]
|
|
|
|
time.sleep(WINDOW * 2)
|
|
assert _run_due(queue, service) == 1
|
|
assert seen == [1.0, 3.0]
|
|
|
|
|
|
def test_a_join_behind_a_skipped_branch_still_runs():
|
|
"""Skipping a node frees its consumers, which may already have been passed."""
|
|
seen: list[tuple[float, float]] = []
|
|
source = make_node(
|
|
"source",
|
|
lambda params: None,
|
|
provides=[spec("fast"), spec("slow", interval=WINDOW)],
|
|
)
|
|
middle = make_node(
|
|
"middle",
|
|
lambda slow, params: {"derived": slow},
|
|
requires=[spec("slow")],
|
|
provides=[spec("derived")],
|
|
)
|
|
join = make_node(
|
|
"join",
|
|
lambda fast, derived, params: seen.append((fast, derived)),
|
|
requires=[spec("fast"), spec("derived")],
|
|
)
|
|
_pipeline, queue, service = _running([source, middle, join])
|
|
|
|
def deliver(value: float) -> None:
|
|
source.inject({"fast": value, "slow": value})
|
|
for item in queue.claim(10, 10):
|
|
service._run_item(item)
|
|
|
|
deliver(1.0)
|
|
assert seen == [(1.0, 1.0)]
|
|
|
|
# `slow` is held this time, so `middle` is skipped — and `join` still has
|
|
# to run, because `fast` did change.
|
|
deliver(2.0)
|
|
assert seen == [(1.0, 1.0), (2.0, 1.0)]
|
|
|
|
|
|
def test_a_manual_run_is_never_throttled_on_its_inputs():
|
|
seen: list[float] = []
|
|
source = make_node("source", lambda params: {"temp": 20.0}, provides=[spec("temp")])
|
|
consumer = make_node(
|
|
"consumer",
|
|
lambda temp, params: seen.append(temp),
|
|
requires=[spec("temp", interval=3600)],
|
|
)
|
|
pipeline = Pipeline(nodes=[source, consumer])
|
|
|
|
# Pressing Run is an explicit ask; the interval governs the flow's own
|
|
# traffic, not what the person in front of it asked for.
|
|
pipeline.run({})
|
|
pipeline.run({})
|
|
|
|
assert len(seen) == 2
|