From afcbb63a56ef658eacbbc66ff9fc052218ce3c6e Mon Sep 17 00:00:00 2001 From: stroblme Date: Thu, 15 Oct 2026 09:32:35 +0200 Subject: [PATCH] Delay node: hold the latest value inside the interval and send it at the window's end 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 Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx --- backend/fluksio/flow/nodes/delay.py | 25 ++++++++++--------- backend/tests/flow/test_discretization.py | 29 ++++++++++++++++++++++- docs/reference/node-types.md | 9 +++---- 3 files changed, 45 insertions(+), 18 deletions(-) diff --git a/backend/fluksio/flow/nodes/delay.py b/backend/fluksio/flow/nodes/delay.py index a9d448e..d8d23f7 100644 --- a/backend/fluksio/flow/nodes/delay.py +++ b/backend/fluksio/flow/nodes/delay.py @@ -23,7 +23,8 @@ class DelayNode(Node): This node can: - Add a fixed delay before forwarding messages - - Rate-limit messages to a minimum interval between forwards + - Rate-limit messages to a minimum interval between forwards, sending the + latest one held inside the window when the window ends - Schedule emissions using crontab syntax (like a cron job) The node passes through all input values to outputs with matching names. @@ -34,7 +35,7 @@ class DelayNode(Node): - Without ``requires``: Emits current timestamp at each cron tick - With ``requires``: Emits last received input at each cron tick - **Execution order:** rate check → delay → forward + **Execution order:** delay → rate check → forward :param params: Parameters dict containing: - ``delay`` (int): Fixed delay in seconds before forwarding (default: 0) @@ -89,7 +90,6 @@ class DelayNode(Node): __slots__ = ( "delay", "interval", - "ts", "mapping", "cron_expr", "last_input", @@ -113,11 +113,18 @@ class DelayNode(Node): self.interval = cfg.interval self.mapping = cfg.mapping self.cron_expr = cfg.cron - self.ts = 0.0 self.last_input: dict[str, Any] = {} self._cron_task: asyncio.Task[None] | None = None self._stop_cron: asyncio.Event | None = None + if cfg.interval > 0: + # The engine's own rate limit, on every output: it keeps the latest + # value inside the window and publishes it when the window ends. + kwargs["provides"] = [ + spec.model_copy(update={"interval": max(spec.interval, cfg.interval)}) + for spec in self._normalize_ports(kwargs.get("provides", ())) + ] + super().__init__(f=self._f, params=params, **kwargs) def _port_pairs(self) -> list[tuple[str, str]]: @@ -130,7 +137,7 @@ class DelayNode(Node): ] def _f(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: - """Forward messages with optional delay, rate-limiting, and alarm.""" + """Forward messages, after a delay if one is set.""" logger.debug("[%s] Received %s", self.name, kwargs) @@ -138,14 +145,6 @@ class DelayNode(Node): if kwargs: self.last_input = dict(kwargs) - # Apply rate limiting first - if self.interval > 0: - ts = time.time() - if ts <= self.ts + self.interval: - logger.debug("[%s] Stashing %s", self.name, kwargs) - return None - self.ts = ts - if not kwargs: return None diff --git a/backend/tests/flow/test_discretization.py b/backend/tests/flow/test_discretization.py index cae205c..6fe3f87 100644 --- a/backend/tests/flow/test_discretization.py +++ b/backend/tests/flow/test_discretization.py @@ -4,7 +4,7 @@ import time from fluksio.flow.executor import ExecutionService from fluksio.flow.messages import DType, MessageSpec -from fluksio.flow.nodes import Node +from fluksio.flow.nodes import DelayNode, Node from fluksio.flow.pipeline import Pipeline from fluksio.flow.queue import MemoryWorkQueue @@ -213,6 +213,33 @@ def test_a_node_whose_inputs_did_not_change_is_not_run(): 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]] = [] diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md index 7366d04..6f2e861 100644 --- a/docs/reference/node-types.md +++ b/docs/reference/node-types.md @@ -171,13 +171,14 @@ Use `cron` for scheduled emissions, or trigger the node by hand for testing. | Setting | Default | Notes | |---|---|---| | `delay` | `0` | seconds to hold each message, fractional | -| `interval` | `0` | minimum seconds between forwards | +| `interval` | `0` | minimum seconds between forwards, per output | | `mapping` | `{}` | input port → output port; paired in order when empty | | `cron` | — | five-field expression | -Order of operations: rate check → delay → forward. With a `cron` and no inputs -it emits the current time on each tick; with inputs it emits the last value it -received. +Order of operations: delay → rate check → forward. A message arriving inside +the `interval` window is held, and when the window ends the latest one held is +sent. With a `cron` and no inputs it emits the current time on each tick; with +inputs it emits the last value it received. Timers wait until their deadline; busy workers can delay execution.