From 121cb2e8f00ce02e2d22033e29bb198dda1c8a1d Mon Sep 17 00:00:00 2001 From: stroblme Date: Thu, 27 Aug 2026 15:54:15 +0200 Subject: [PATCH] Record batch runs beside cascades so Home lists them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A batch run opens no cascade, and FlowRun was written only from cascade_started — so `fluksio run` showed on /runs and in `fluksio status` and was simply absent from Home. The collector now folds the run_started and run_finished events RunService already published. Such a record is exempt from the staleness sweep in both places: a training step of an hour is a normal one, and only run_finished ends it. Co-Authored-By: Claude Opus 5 (1M context) --- backend/fluksio/flow/metrics.py | 40 ++++++++++++++++++++++----- backend/fluksio/models.py | 13 ++++++--- backend/tests/test_metrics.py | 48 +++++++++++++++++++++++++++++++-- 3 files changed, 89 insertions(+), 12 deletions(-) diff --git a/backend/fluksio/flow/metrics.py b/backend/fluksio/flow/metrics.py index 9de6085..7fc52c5 100644 --- a/backend/fluksio/flow/metrics.py +++ b/backend/fluksio/flow/metrics.py @@ -35,6 +35,10 @@ FLUSH_INTERVAL_S = 15.0 DETAIL_CAP = 8000 #: A cascade still open this long after it started is never finishing. RUN_STALE_S = 600.0 +#: What ``flow_run.source`` says for a batch run, as against the cause a +#: cascade carries. A training step can legitimately run for hours, so this is +#: also what exempts one from the staleness sweep above. +RUN_SOURCE = "run" #: ``metric_minute.flow``/``node`` and ``flow_run.flow`` are this wide, and a #: node id has no length of its own. One row over it aborts the transaction, #: which would take every other row in the flush with it. @@ -214,9 +218,18 @@ class MetricsCollector: if kind == "cascade_finished": if run is not None: - run["finished_at"] = datetime.fromtimestamp(ts, UTC) - run["duration_ms"] = round((ts - run["started_ts"]) * 1000, 2) - run["status"] = "error" if run["errors"] else "ok" + self._finish_run(run, ts) + return + + # A batch run opens no cascade, so without these it ran, recorded its + # nodes and its failures, and appeared in no history at all. + if kind == "run_started": + self._start_run({**event, "cause": RUN_SOURCE}, ts) + return + + if kind == "run_finished": + if run is not None: + self._finish_run(run, ts, str(event.get("status") or "")) return if kind == "node_health": @@ -257,6 +270,14 @@ class MetricsCollector: ) ) + def _finish_run( + self, run: dict[str, Any], ts: float, status: str = "" + ) -> None: + """Close an open record. A run says how it ended; a cascade is told.""" + run["finished_at"] = datetime.fromtimestamp(ts, UTC) + run["duration_ms"] = round((ts - run["started_ts"]) * 1000, 2) + run["status"] = status or ("error" if run["errors"] else "ok") + def _start_run(self, event: dict[str, Any], ts: float) -> None: run_id = str(event.get("run") or "") if not run_id: @@ -305,7 +326,11 @@ class MetricsCollector: self._last_prune = time.monotonic() cutoff = time.time() - RUN_STALE_S for run_id, run in list(self._runs.items()): - if run["status"] != "running" or run["started_ts"] < cutoff: + if run["status"] != "running": + del self._runs[run_id] + # A batch run is held however long it takes: a training step of an + # hour is a normal one, and `run_finished` is what ends it. + elif run["source"] != RUN_SOURCE and run["started_ts"] < cutoff: del self._runs[run_id] # Held tracebacks survive the flush: the log and the failure it belongs # to are two events, and a flush can fall between them. One per node, @@ -405,12 +430,15 @@ class MetricsCollector: session.execute(delete(MetricBucket).where(col(MetricBucket.bucket) < cutoff)) session.execute(delete(EngineEvent).where(col(EngineEvent.ts) < cutoff)) session.execute(delete(FlowRun).where(col(FlowRun.started_at) < cutoff)) - # A run still open long after it started did not finish; saying so is - # more honest than leaving it running forever. + # A cascade still open long after it started did not finish; saying so + # is more honest than leaving it running forever. A batch run is not + # swept: it is allowed to take hours, and `Run` has its own lease sweep + # for the case where the engine died holding one. session.execute( update(FlowRun) .where( col(FlowRun.status) == "running", + col(FlowRun.source) != RUN_SOURCE, col(FlowRun.started_at) < now - timedelta(seconds=RUN_STALE_S), ) .values(status="abandoned", finished_at=now) diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 06ede43..74c8de6 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -279,14 +279,19 @@ class EngineEvent(SQLModel, table=True): class FlowRun(SQLModel, table=True): - """One cascade, from the item that started it to the last node in it.""" + """One cascade, from the item that started it to the last node in it. + + Or one batch run: it opens no cascade, but it is still a thing that ran, + and the history screens are where somebody looks for it. + """ __tablename__ = "flow_run" - #: The queue entry id, or a `manual-` one for a run that never queued. + #: The queue entry id, a `manual-` one for a run that never queued, or the + #: `Run.id` of a batch run. id: str = Field(primary_key=True, max_length=64) flow: str = Field(index=True, max_length=255) - #: What caused it: the work item's cause, or "manual". + #: What caused it: the work item's cause, "manual", or "run" for a batch. source: str = "" started_at: datetime = Field( index=True, @@ -296,7 +301,7 @@ class FlowRun(SQLModel, table=True): default=None, sa_type=UTCDateTime, ) - #: running, ok, error or abandoned. + #: running, ok, error, abandoned — or whatever a batch run ended as. status: str = "running" nodes: int = 0 errors: int = 0 diff --git a/backend/tests/test_metrics.py b/backend/tests/test_metrics.py index a628ba5..688573e 100644 --- a/backend/tests/test_metrics.py +++ b/backend/tests/test_metrics.py @@ -226,11 +226,55 @@ def test_a_traceback_no_failure_ever_claims_is_dropped() -> None: assert collector._tracebacks == {} +def test_a_batch_run_is_recorded_beside_the_cascades(db: Session) -> None: + """`fluksio run` showed on /runs and nowhere on Home. + + It opens no cascade, so the collector saw its nodes and its failures with + no record to fold them into. The start is aged past the staleness cutoff on + purpose: a training step of an hour is a normal one, and only + `run_finished` may end a batch run. + """ + collector = MetricsCollector(EventBus()) + started = time.time() - RUN_STALE_S - 60 + collector.handle( + {"type": "run_started", "run": "run-batch", "flow": FLOW, "ts": started} + ) + collector.handle( + { + "type": "node_executed", + "flow": FLOW, + "node": NODE, + "run": "run-batch", + "duration_ms": 5.0, + "ts": started, + } + ) + asyncio.run(collector.flush()) + + row = db.exec(select(FlowRun).where(FlowRun.id == "run-batch")).one() + assert (row.status, row.source, row.nodes) == ("running", "run", 1) + + collector.handle( + { + "type": "run_finished", + "run": "run-batch", + "flow": FLOW, + "status": "ok", + "ts": started + 30, + } + ) + asyncio.run(collector.flush()) + + db.expire_all() + row = db.exec(select(FlowRun).where(FlowRun.id == "run-batch")).one() + assert (row.status, row.duration_ms) == ("ok", 30_000.0) + + def test_a_failure_keeps_the_run_it_happened_in(db: Session) -> None: """The payload always carried it; the row used to drop it. - A batch run publishes no `cascade_started`, so there is no `FlowRun` beside - this — which is exactly the case that had no way of being asked about. + Nothing told this collector the run had started, so there is no `FlowRun` + beside it — the failure still has to name what it happened in. """ collector = MetricsCollector(EventBus()) ts = datetime.now(UTC).replace(second=0, microsecond=0).timestamp()