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.