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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx
This commit is contained in:
2026-10-15 09:32:35 +02:00
co-authored by Claude Opus 5.5
parent 7d28b44451
commit afcbb63a56
3 changed files with 45 additions and 18 deletions
+12 -13
View File
@@ -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