diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 0817f29..22da5ff 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -1486,6 +1486,17 @@ class Pipeline: complete(n) continue ready_inputs: dict[str, Any] | None = None + if failed is not None: + # A live wave's readiness rule: a node runs once every + # triggering input has a value. A run has nothing to wait + # for, so the node fails instead. + ready_inputs = state.get_present(list(n.requires)) + missing = self._no_value(n, ready_inputs) + if missing: + failed.add(n) + self._observe(NodeOutcome(node=n.id, ok=False, error=missing)) + complete(n) + continue if check_ready: ready, ready_inputs = self._is_node_ready(n, state, entry_id) if not ready: @@ -1530,6 +1541,18 @@ class Pipeline: return state + def _no_value(self, node: Node, values: dict[str, Any]) -> str: + """The first triggering input with nothing in ``values``, as an error, + or "" when there is none.""" + for name, spec in node.requires.items(): + if spec.trigger and name not in values: + producers = ", ".join( + f"'{p.local_id}'" for p in self.produces.get(name, ()) + ) + where = f" from {producers}" if producers else "" + return f"No value on input '{spec.port or name}'{where}" + return "" + def run( self, inputs: dict[str, Any] | None = None, nodes: set[Node] | None = None ) -> StateBackend: diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 894989c..1061161 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -780,6 +780,54 @@ def test_a_failed_stage_skips_what_waits_on_it_and_fails_the_run(tmp_path): session.commit() +def test_a_stage_with_nothing_on_an_input_fails_instead_of_being_called(tmp_path): + """The producer succeeded but published nothing on the port it waits on.""" + model, score, report = ( + MessageSpec(name=n, dtype=DType.FLOAT) for n in ("model", "score", "report") + ) + flow = FlowDef( + name="split", + mode="batch", + nodes=[ + NodeDef(id="fit", provides=[model, score]), + NodeDef(id="readout", requires=[score], provides=[report]), + NodeDef(id="evaluate", requires=[report]), + ], + ) + store = FlowStore(tmp_path / "flows") + store.write_flow(flow) + store.write_node_source( + "split", "fit", "def process():\n return {'model': 1.0}\n" + ) + store.write_node_source( + "split", "readout", "def process(score):\n return {'report': score}\n" + ) + store.write_node_source("split", "evaluate", "def process(report):\n pass\n") + + service = RunService(controller=FlowController(store), queue=_Collect()) + run_id = service.submit("split", {}).id + try: + service._drive(run_id) + with Session(db_engine) as session: + assert session.get(Run, run_id).status == "error" + rows = { + row.node: row + for row in session.exec(select(RunNode).where(RunNode.run_id == run_id)) + } + assert {node: row.status for node, row in rows.items()} == { + "split.fit": "ok", + "split.readout": "error", + "split.evaluate": "skipped", + } + assert rows["split.readout"].error == "No value on input 'score' from 'fit'" + finally: + with Session(db_engine) as session: + for node in session.exec(select(RunNode).where(RunNode.run_id == run_id)): + session.delete(node) + session.delete(session.get(Run, run_id)) + session.commit() + + def test_a_retry_is_a_new_run_that_names_the_one_it_repeats(): """The way back from a run an engine restart interrupted. diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index 2c711e1..333b9a8 100644 --- a/docs/concepts/runs.md +++ b/docs/concepts/runs.md @@ -21,6 +21,9 @@ finish. Missing parameters use their declared initial values. When a node fails, the nodes that wait on it are recorded as `skipped` and the run ends `error`. A node that cannot be loaded fails the run before anything executes, with the node's load error as the reason. +A node with no value on a triggering input fails without being called, as a +live flow would wait for it, so a canvas port with a default needs +`trigger: false` to be optional. Non-streaming ports cannot use `interval` in a batch flow: there is no timer to deliver a delayed value. The same goes for a delay node's `interval`