stroblmeandClaude Opus 5
d4a9406c51
Docs / docs (push) Successful in 49s
Playwright Tests / test-playwright (1, 2) (push) Failing after 1m11s
Playwright Tests / test-playwright (2, 2) (push) Failing after 23s
pre-commit / pre-commit (push) Successful in 3m2s
Test Backend / test-backend (push) Successful in 2m22s
Compose Smoke Test / test-compose (push) Failing after 22s
Playwright Tests / merge-reports (push) Canceled after 1s
The gates have never gone green on the new runners. Three separate reasons: - backend/Dockerfile shipped Python 3.10 while the code imports typing.Self and datetime.UTC, so the container exited on import and the suite could not even load its conftest. The image moves to 3.13 and the packages declare >=3.12, which is the floor the tests actually pass on; ruff's target follows and rewrites timezone.utc and asyncio.TimeoutError accordingly. Relocking drops the 3.10 branch, which bumps FastAPI and so regenerates the SDK. - frontend/README.md had no trailing newline and two dashboard widgets used arbitrary text-[…] sizes. Both are em-relative on purpose, so they move to the inline style the neighbouring ramp already uses. - Every commit left its own run queued: without a concurrency group a runner that was offline for a while works through a backlog nobody reads. A stack that fails to come up now prints its logs before the teardown removes it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
334 lines
11 KiB
Python
334 lines
11 KiB
Python
"""Timing nodes: fixed delay, rate limit, and the cron schedule."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from pydantic import BaseModel, ConfigDict
|
|
|
|
from fluksio.flow.nodes.base import Node
|
|
|
|
if TYPE_CHECKING:
|
|
from fastapi import FastAPI
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DelayNode(Node):
|
|
"""
|
|
A node that adds delay, rate-limiting, and/or cron-scheduled emissions.
|
|
|
|
This node can:
|
|
- Add a fixed delay before forwarding messages
|
|
- Rate-limit messages to a minimum interval between forwards
|
|
- Schedule emissions using crontab syntax (like a cron job)
|
|
|
|
The node passes through all input values to outputs with matching names.
|
|
If ``requires`` and ``provides`` have the same message names, the values are
|
|
forwarded directly. Otherwise, you can specify a ``mapping`` in params.
|
|
|
|
**Cron functionality:**
|
|
- 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
|
|
|
|
:param params: Parameters dict containing:
|
|
- ``delay`` (int): Fixed delay in seconds before forwarding (default: 0)
|
|
- ``interval`` (int): Minimum interval in seconds between forwards (default: 0)
|
|
- ``mapping`` (dict): Optional mapping from input names to output names
|
|
- ``cron`` (str): Crontab expression for scheduled emissions (optional).
|
|
Standard 5-field format: ``minute hour day-of-month month day-of-week``
|
|
:type params: dict
|
|
:param kwargs: Additional arguments passed to Node (requires, provides, name)
|
|
|
|
:example:
|
|
Simple passthrough with delay:
|
|
|
|
>>> delay_node = DelayNode(
|
|
... requires=[MessageSpec(name="temperature", dtype=DType.FLOAT)],
|
|
... provides=[MessageSpec(name="temperature", dtype=DType.FLOAT)],
|
|
... params={"delay": 1},
|
|
... )
|
|
|
|
Rate-limited forwarding:
|
|
|
|
>>> throttle_node = DelayNode(
|
|
... requires=[MessageSpec(name="sensor_data", dtype=DType.FLOAT)],
|
|
... provides=[MessageSpec(name="sensor_data", dtype=DType.FLOAT)],
|
|
... params={"interval": 5},
|
|
... )
|
|
|
|
Cron: emit timestamp every day at 08:30:
|
|
|
|
>>> alarm_node = DelayNode(
|
|
... provides=[MessageSpec(name="timestamp", dtype=DType.FLOAT)],
|
|
... params={"cron": "30 8 * * *"},
|
|
... )
|
|
|
|
Cron: emit stored input every 5 minutes:
|
|
|
|
>>> periodic_node = DelayNode(
|
|
... requires=[MessageSpec(name="value", dtype=DType.FLOAT)],
|
|
... provides=[MessageSpec(name="value", dtype=DType.FLOAT)],
|
|
... params={"cron": "*/5 * * * *"},
|
|
... )
|
|
|
|
Cron: emit on weekdays at 23:59:
|
|
|
|
>>> weekday_node = DelayNode(
|
|
... requires=[MessageSpec(name="daily_summary", dtype=DType.FLOAT)],
|
|
... provides=[MessageSpec(name="daily_summary", dtype=DType.FLOAT)],
|
|
... params={"cron": "59 23 * * 1-5"},
|
|
... )
|
|
"""
|
|
|
|
__slots__ = (
|
|
"delay",
|
|
"interval",
|
|
"ts",
|
|
"mapping",
|
|
"cron_expr",
|
|
"last_input",
|
|
"_cron_task",
|
|
"_stop_cron",
|
|
)
|
|
|
|
class Params(BaseModel):
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
# Seconds, fractional: a shutter's run-time is not a whole number.
|
|
delay: float = 0
|
|
interval: float = 0
|
|
# Input port → output port; ports are paired in order when empty.
|
|
mapping: dict[str, str] = {}
|
|
cron: str | None = None
|
|
|
|
def __init__(self, params: dict[str, Any] | None = None, **kwargs: Any) -> None:
|
|
cfg = self.Params.model_validate(params or {})
|
|
self.delay = cfg.delay
|
|
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
|
|
|
|
super().__init__(f=self._f, params=params, **kwargs)
|
|
|
|
def _port_pairs(self) -> list[tuple[str, str]]:
|
|
"""Input port → output port pairs this node forwards along."""
|
|
if self.mapping:
|
|
return list(self.mapping.items())
|
|
return [
|
|
(i.port, o.port)
|
|
for i, o in zip(self.input_ports, self.output_ports, strict=False)
|
|
]
|
|
|
|
def _f(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
|
|
"""Forward messages with optional delay, rate-limiting, and alarm."""
|
|
|
|
logger.info("[%s] Received %s", self.name, kwargs)
|
|
|
|
# Store last input for cron use
|
|
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.info("[%s] Stashing %s", self.name, kwargs)
|
|
return None
|
|
self.ts = ts
|
|
|
|
if not kwargs:
|
|
return None
|
|
|
|
output = {
|
|
out_port: kwargs[in_port]
|
|
for in_port, out_port in self._port_pairs()
|
|
if in_port in kwargs
|
|
}
|
|
|
|
if self.delay > 0:
|
|
# Hand the wait to the queue rather than sit on a worker thread:
|
|
# a handful of delay nodes would otherwise occupy the whole pool
|
|
# and the engine would stop dead until they woke up.
|
|
if self._pipeline is not None and self._pipeline.defer(
|
|
self, self._to_messages(output) or {}, self.delay
|
|
):
|
|
logger.info("[%s] Sending %s in %ss", self.name, output, self.delay)
|
|
return None
|
|
time.sleep(self.delay)
|
|
|
|
logger.info("[%s] Sending %s", self.name, output)
|
|
return output
|
|
|
|
# -----------------------------------------------------------------
|
|
# Cron scheduler
|
|
# -----------------------------------------------------------------
|
|
|
|
async def start(self, app: FastAPI | None = None) -> None:
|
|
"""Only a node with a schedule has anything to run on its own."""
|
|
if self.cron_expr:
|
|
await self.start_cron()
|
|
|
|
async def stop(self, app: FastAPI | None = None) -> None:
|
|
if self.cron_expr:
|
|
await self.stop_cron()
|
|
|
|
async def start_cron(self) -> None:
|
|
"""
|
|
Start the cron scheduler.
|
|
|
|
Requires the ``croniter`` package. The scheduler runs in the background
|
|
and triggers emissions according to the ``cron`` expression in params.
|
|
|
|
:raises ValueError: If no ``cron`` expression is configured.
|
|
"""
|
|
if not self.cron_expr:
|
|
logger.warning("No cron expression configured for node '%s'", self.name)
|
|
return
|
|
|
|
if self._cron_task is not None:
|
|
logger.info("Cron already running for node '%s'", self.name)
|
|
return
|
|
|
|
self._stop_cron = asyncio.Event()
|
|
self._cron_task = self._run_supervised("cron", self._cron_loop)
|
|
logger.info("Started cron for node '%s': %s", self.name, self.cron_expr)
|
|
|
|
async def stop_cron(self) -> None:
|
|
"""Stop the cron scheduler."""
|
|
if self._stop_cron is None:
|
|
return
|
|
|
|
self._stop_cron.set()
|
|
|
|
if self._cron_task is not None:
|
|
self._cron_task.cancel()
|
|
try:
|
|
await self._cron_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
self._cron_task = None
|
|
self._stop_cron = None
|
|
logger.info("Stopped cron for node '%s'", self.name)
|
|
|
|
async def _cron_loop(self) -> None:
|
|
"""Background loop that sleeps until the next cron tick and triggers."""
|
|
from datetime import datetime
|
|
|
|
try:
|
|
from croniter import croniter # type: ignore[import-untyped]
|
|
except ImportError:
|
|
logger.error(
|
|
"croniter package is required for cron scheduling. "
|
|
"Install it with: pip install croniter"
|
|
)
|
|
return
|
|
|
|
if not croniter.is_valid(self.cron_expr):
|
|
logger.error(
|
|
"Invalid cron expression '%s' for node '%s'",
|
|
self.cron_expr,
|
|
self.name,
|
|
)
|
|
return
|
|
|
|
cron = croniter(self.cron_expr, datetime.now())
|
|
|
|
# Walrus so the body can use the event without re-reading a field
|
|
# stop_cron() may have cleared in the meantime.
|
|
while (stop := self._stop_cron) is not None and not stop.is_set():
|
|
try:
|
|
# Compute seconds until next tick
|
|
next_dt = cron.get_next(datetime)
|
|
now = datetime.now()
|
|
wait_seconds = max(0, (next_dt - now).total_seconds())
|
|
|
|
logger.info(
|
|
"Node '%s' cron: next tick at %s (in %.1fs)",
|
|
self.name,
|
|
next_dt.isoformat(),
|
|
wait_seconds,
|
|
)
|
|
|
|
# Sleep until next tick (wake up on stop signal)
|
|
try:
|
|
await asyncio.wait_for(stop.wait(), timeout=wait_seconds)
|
|
# If we get here, stop was requested
|
|
break
|
|
except TimeoutError:
|
|
# Timeout means it's time to fire
|
|
pass
|
|
|
|
logger.info(
|
|
"Cron triggered for node '%s' (%s)",
|
|
self.name,
|
|
self.cron_expr,
|
|
)
|
|
await self._trigger_cron()
|
|
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
# One bad tick should not cost the schedule; the supervisor
|
|
# picks it up from here, backoff included.
|
|
logger.error(
|
|
"Error in cron loop for node '%s'", self.name, exc_info=True
|
|
)
|
|
raise
|
|
|
|
async def _trigger_cron(self) -> None:
|
|
"""Emit data into the pipeline on a cron tick."""
|
|
if self._pipeline is None:
|
|
logger.warning(
|
|
"Node '%s' not bound to pipeline, cannot trigger cron",
|
|
self.name,
|
|
)
|
|
return
|
|
|
|
try:
|
|
output = {}
|
|
|
|
if not self.input_ports:
|
|
# Nothing to forward — emit the current time on every output.
|
|
output = {spec.port: time.time() for spec in self.output_ports}
|
|
logger.info("Cron emission for node '%s': timestamp", self.name)
|
|
elif self.last_input:
|
|
output = {
|
|
out_port: self.last_input[in_port]
|
|
for in_port, out_port in self._port_pairs()
|
|
if in_port in self.last_input
|
|
}
|
|
logger.info(
|
|
"Cron emission for node '%s': stored input %s",
|
|
self.name,
|
|
output,
|
|
)
|
|
else:
|
|
logger.warning(
|
|
"Cron for node '%s' triggered but no input stored yet",
|
|
self.name,
|
|
)
|
|
return
|
|
|
|
if output:
|
|
await asyncio.to_thread(
|
|
self._pipeline.trigger, self, self._to_messages(output)
|
|
)
|
|
except Exception as e:
|
|
logger.error(
|
|
"Error triggering cron for node '%s': %s",
|
|
self.name,
|
|
e,
|
|
exc_info=True,
|
|
)
|