Starting a running flow restarts its nodes, so interval and cron Injects keep firing; a params-dict node is told settings arrive by name

This commit is contained in:
2026-11-01 10:59:14 +01:00
parent e0f5ddc3a4
commit 48d4be6093
3 changed files with 30 additions and 4 deletions
+13 -3
View File
@@ -659,13 +659,19 @@ class FlowController:
await self._acquire(FLOW_REBUILD_WAIT)
try:
if enabled:
running = flow not in self.disabled
# The gate first: a subscription that fires the instant it
# starts must not be turned away by a flag on its way out.
self.disabled = self.disabled - {flow}
self.pipeline.set_disabled(self.disabled)
# Lifts any quarantine, which is what a rebuild used to do by
# throwing the whole supervisor away.
await self.supervisor.cancel_flow(flow)
# throwing the whole supervisor away. A running flow's nodes
# are stopped first: a node already started returns from
# start() at once, which would leave its cancelled loop dead.
if running:
await self._teardown(flow)
else:
await self.supervisor.cancel_flow(flow)
await self._activate(flow)
else:
self.disabled = self.disabled | {flow}
@@ -1989,10 +1995,14 @@ def _signature_gap(
if not missing:
return None
kind = "an input" if missing[0] in ports else "a setting"
return (
gap = (
f"'{missing[0]}' is {kind} of '{node_id}' but "
f"process({', '.join(names)}) takes no such argument"
)
if kind == "a setting" and "params" in names:
# Code written for the `params` dict settings used to arrive in.
gap += f". Settings arrive by name: process(..., {missing[0]})"
return gap
def follow_header(node: NodeDef, code: str, drop: bool = True) -> NodeDef:
+11 -1
View File
@@ -9,7 +9,7 @@ from pathlib import Path
import pytest
from fluksio.flow.controller import FlowController, follow_header
from fluksio.flow.controller import FlowController, _signature_gap, follow_header
from fluksio.flow.messages import DType, MessageSpec
from fluksio.flow.pipeline import Pipeline
from fluksio.flow.schemas import FlowDef, NodeDef
@@ -226,3 +226,13 @@ def test_a_header_it_cannot_read_for_certain_leaves_the_node():
"def run(value): pass\n",
):
assert follow_header(node, code) == node
def test_code_written_for_the_params_dict_is_told_settings_arrive_by_name():
gap = _signature_gap("def process(a, params): pass\n", "n", {"a"}, {"target"})
assert gap is not None
assert "process(..., target)" in gap
assert (
_signature_gap("def process(a, target): pass\n", "n", {"a"}, {"target"}) is None
)
@@ -331,6 +331,12 @@ def test_stopping_one_flow_does_not_touch_the_others(
assert lifecycle == ["start a.io"]
assert {name: engine.loaded[name].node for name in ("a.io", "b.io")} == before
# Started while running is a restart: the cancelled loops come back.
lifecycle.clear()
await engine.set_enabled("a", True)
assert lifecycle == ["stop a.io", "start a.io"]
asyncio.run(asyncio.wait_for(scenario(), timeout=10))