diff --git a/backend/fluksio/alembic/versions/d4a71e9c2b58_engine_event_run.py b/backend/fluksio/alembic/versions/d4a71e9c2b58_engine_event_run.py new file mode 100644 index 0000000..e697b85 --- /dev/null +++ b/backend/fluksio/alembic/versions/d4a71e9c2b58_engine_event_run.py @@ -0,0 +1,40 @@ +"""engine_event.run + +Which run an event happened in. The payload always carried it and the persist +path dropped it, so reading one run's failures meant filtering the global list. +Rows written before this carry an empty string, the same as anything that +belongs to no run at all. + +Revision ID: d4a71e9c2b58 +Revises: c7e2b9f34a15 +Create Date: 2026-08-27 + +""" + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from alembic import op + +# revision identifiers, used by Alembic. +revision = "d4a71e9c2b58" +down_revision = "c7e2b9f34a15" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column( + "engine_event", + sa.Column( + "run", + sqlmodel.sql.sqltypes.AutoString(length=64), + nullable=False, + server_default="", + ), + ) + op.create_index("ix_engine_event_run", "engine_event", ["run"]) + + +def downgrade(): + op.drop_index("ix_engine_event_run", table_name="engine_event") + op.drop_column("engine_event", "run") diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index 5653153..1333baf 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -108,6 +108,7 @@ class EventRow(BaseModel): node: str detail: str actor: str + run: str class DeadLetter(BaseModel): @@ -389,6 +390,7 @@ def read_events( session: SessionDep, kind: Literal["failure", "audit"] = "failure", flow: str | None = None, + run: str | None = None, since: datetime | None = None, until: datetime | None = None, limit: int = 100, @@ -397,6 +399,9 @@ def read_events( ``since`` is inclusive and ``until`` exclusive, the same window ``/runs`` takes, so a list can cover the span the charts beside it are drawn from. + ``run`` narrows to one run — a batch run's id, or the journaled item a live + cascade came from. Rows recorded before the column existed carry none, so + an old failure answers no run at all rather than the wrong one. """ statement = select(EngineEvent).order_by(col(EngineEvent.ts).desc()) if kind == "audit": @@ -405,6 +410,8 @@ def read_events( statement = statement.where(col(EngineEvent.type) != "audit") if flow: statement = statement.where(col(EngineEvent.flow) == flow) + if run: + statement = statement.where(col(EngineEvent.run) == run) if since: statement = statement.where(col(EngineEvent.ts) >= _aware(since)) if until: diff --git a/backend/fluksio/cli.py b/backend/fluksio/cli.py index eedd9b8..1429945 100644 --- a/backend/fluksio/cli.py +++ b/backend/fluksio/cli.py @@ -220,8 +220,22 @@ def _sign_in(admin_id: Any, url: str, data_dir: Path) -> Path: return write_config(url, token, data_dir) +#: The `serve` flags that are settings under another name, and the setting each +#: one writes. A flag is a real environment variable, which outranks the env +#: file the settings read — so the order is flag, environment, `/env`. +CONCURRENCY_FLAGS = { + "max_workers": "FLOW_MAX_WORKERS", + "max_cascades": "FLOW_MAX_CASCADES", + "max_runs": "FLOW_MAX_RUNS", +} + + def cmd_serve(args: argparse.Namespace) -> int: data_dir = _data_dir(args.data_dir, args.shared) + for flag, name in CONCURRENCY_FLAGS.items(): + value = getattr(args, flag, None) + if value is not None: + os.environ[name] = str(value) _prepare(data_dir) from sqlmodel import Session @@ -370,6 +384,27 @@ def _parser() -> argparse.ArgumentParser: metavar="URL", help=f"the portal --enroll redeems at (default {DEFAULT_PORTAL})", ) + serve.add_argument( + "--max-runs", + type=int, + default=None, + metavar="N", + help="batch runs driven at once (default 4, FLOW_MAX_RUNS)", + ) + serve.add_argument( + "--max-cascades", + type=int, + default=None, + metavar="N", + help="cascades in flight at once (default 4, FLOW_MAX_CASCADES)", + ) + serve.add_argument( + "--max-workers", + type=int, + default=None, + metavar="N", + help="python worker processes (default 4, FLOW_MAX_WORKERS)", + ) serve.set_defaults(func=cmd_serve) enroll = subparsers.add_parser( diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py index f52b64e..1e8729b 100644 --- a/backend/fluksio/core/config.py +++ b/backend/fluksio/core/config.py @@ -102,6 +102,10 @@ class Settings(BaseSettings): # over the mean cascade time, so an installation whose nodes wait on the # network rather than on a CPU wants it higher than the core count. FLOW_MAX_CASCADES: int = 4 + # How many batch runs are driven at once. A different limit from the one + # above: a run drives a whole graph, and its nodes are bounded by the worker + # pool rather than by cascade slots. A sweep is what this governs. + FLOW_MAX_RUNS: int = 4 # How long a python node may be silent before its worker is killed, unless # the node sets its own. 0, the default, disables it: a dead worker still # fails fast, and a slow one is left to finish. Set it where silence means diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 73d897a..06850a0 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -1612,6 +1612,7 @@ class FlowController: observer=observer, emission_observer=emission_observer, run_cache=run_cache, + run_id=run.run_id if run is not None else "", ) pipeline.history_limits = self.history_limits return pipeline diff --git a/backend/fluksio/flow/metrics.py b/backend/fluksio/flow/metrics.py index 4adf66c..9de6085 100644 --- a/backend/fluksio/flow/metrics.py +++ b/backend/fluksio/flow/metrics.py @@ -155,7 +155,11 @@ class MetricsCollector: """Fold one event in. Synchronous: this is arithmetic on dicts.""" kind = str(event.get("type") or "") ts = float(event.get("ts") or time.time()) - run = self._runs.get(str(event.get("run") or "")) + # Two different things: the id as the event carries it, which is what a + # recorded row keeps, and the live cascade it belongs to — which a batch + # run has none of, since nothing here started one for it. + run_id = str(event.get("run") or "")[:64] + run = self._runs.get(run_id) if kind == "node_executed": bucket = self._bucket(event) @@ -199,6 +203,7 @@ class MetricsCollector: detail=(f"{error}\n{traceback}" if traceback else error)[ :DETAIL_CAP ], + run=run_id, ) ) return @@ -223,6 +228,7 @@ class MetricsCollector: flow=str(event.get("flow") or ""), node=str(event.get("node") or ""), detail=_detail(event) or "Reported itself down.", + run=run_id, ) ) return @@ -247,6 +253,7 @@ class MetricsCollector: flow=str(event.get("flow") or ""), node=str(event.get("node") or event.get("task") or ""), detail=_detail(event), + run=run_id, ) ) diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 7b0418b..be80ea0 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -237,6 +237,7 @@ class Pipeline: "observer", "emission_observer", "run_cache", + "run_id", ) def __init__( @@ -252,6 +253,7 @@ class Pipeline: observer: Callable[[NodeOutcome], None] | None = None, emission_observer: Callable[[str, dict[str, Any]], None] | None = None, run_cache: RunCacheLookup | None = None, + run_id: str = "", ) -> None: self._nodes = nodes or [] # Stopped flows are stored and survive a restart; paused ones are a @@ -286,6 +288,11 @@ class Pipeline: # none: a cascade is about what just happened, not about what a node # once returned for the same inputs. self.run_cache = run_cache + # The batch run this pipeline belongs to, if any. A live cascade names + # the journaled item it came from instead, which is what the events + # below carry; a run has no such item, so without this its failures + # would be recorded belonging to nothing. + self.run_id = run_id # How deep to keep each message's series; a chart asking for more # than the default puts its message in here. Swapped, never mutated. self.history_limits: dict[str, int] = {} @@ -853,7 +860,7 @@ class Pipeline: "flow": node.flow, "node": node.id, "error": error, - "run": entry_id, + "run": entry_id or self.run_id, "ts": time.time(), } ) @@ -916,7 +923,7 @@ class Pipeline: "node": node.id, "outputs": len(outputs or {}), "duration_ms": 0.0, - "run": entry_id, + "run": entry_id or self.run_id, "ts": time.time(), } ) @@ -1007,7 +1014,7 @@ class Pipeline: # which is a different thing to show than one that emitted. "outputs": len(result or {}), "duration_ms": duration_ms, - "run": entry_id, + "run": entry_id or self.run_id, "ts": time.time(), } ) diff --git a/backend/fluksio/flow/provision.py b/backend/fluksio/flow/provision.py index 573c7a9..204d4eb 100644 --- a/backend/fluksio/flow/provision.py +++ b/backend/fluksio/flow/provision.py @@ -125,7 +125,12 @@ class SlurmProvisioner: profiles: list[SlurmProfile], ssh_key: str = "", artifact_url: str = "", + #: How long a machine sits idle before it gives itself back; 0 keeps it + #: for as long as the job runs, which is what a queue paid for in hours + #: of wall time wants. max_idle_s: float = 300.0, + #: How long a submitted job may take to attach before it is cancelled; + #: 0 waits for as long as the scheduler makes it wait. provision_timeout_s: float = 900.0, events: EventBus | None = None, ) -> None: @@ -255,7 +260,10 @@ class SlurmProvisioner: if job.worker and job.worker in attached: # It arrived. Asking again is somebody else's decision. del self._outstanding[name] - elif time.monotonic() - job.since > self.provision_timeout_s: + elif ( + self.provision_timeout_s > 0 + and time.monotonic() - job.since > self.provision_timeout_s + ): del self._outstanding[name] if job.job_id: expired.append(job.job_id) diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index d34efd0..72bf8e6 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -68,9 +68,12 @@ LEASE_INTERVAL_S = 20.0 LEASE_STALE_S = 90.0 #: How often stale leases are looked for. SWEEP_INTERVAL_S = 30.0 -#: Runs driven at once. Node bodies are bounded by the worker pool anyway; -#: this only bounds how many graphs are in flight. +#: Runs driven at once, unless `FLOW_MAX_RUNS` says otherwise. Node bodies are +#: bounded by the worker pool anyway; this only bounds how many graphs are in +#: flight. MAX_PARALLEL = 4 +#: Runs claimed per poll, at least. Raising the pool raises this with it, so a +#: queue of ready runs fills the drivers in one pass rather than four a second. CLAIM_COUNT = 4 CLAIM_BLOCK_MS = 1000 ERROR_CAP = 2000 @@ -706,7 +709,10 @@ class RunService: # the isolation it wants, minus surviving the process. self._state_factory = state_factory or (lambda _ns: MemoryState()) self.engine_name = f"{socket.gethostname()}-{os.getpid()}"[:64] - self._pool = ThreadPoolExecutor(max_workers=parallel, thread_name_prefix="run") + self.parallel = max(1, parallel) + self._pool = ThreadPoolExecutor( + max_workers=self.parallel, thread_name_prefix="run" + ) self._stop = threading.Event() self._consumer: threading.Thread | None = None self._keeper: threading.Thread | None = None @@ -882,7 +888,9 @@ class RunService: # Runs put back to wait for a worker come due here. The claim # below blocks for a second, so this is about once a second. self.queue.move_due(time.time()) - items = self.queue.claim(CLAIM_COUNT, CLAIM_BLOCK_MS) + items = self.queue.claim( + max(CLAIM_COUNT, self.parallel), CLAIM_BLOCK_MS + ) failures = 0 except Exception as exc: failures += 1 diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 3c09b76..4ef3024 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -206,8 +206,17 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: queue=_work_queue("run"), state_factory=_run_state, artifacts=artifacts, + parallel=settings.FLOW_MAX_RUNS, ) app.state.run_service = run_service + # Said out loud because it is the only way to tell that a settings file was + # read at all — the numbers are what someone raising them is looking for. + logger.info( + "Engine limits: workers=%s cascades=%s runs=%s", + settings.FLOW_MAX_WORKERS, + execution.max_cascades, + run_service.parallel, + ) watchdog = LoopWatchdog(event_bus) app.state.watchdog = watchdog watchdog_task = asyncio.create_task(watchdog.run(), name="loop-watchdog") diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 82c9aae..06ede43 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -273,6 +273,9 @@ class EngineEvent(SQLModel, table=True): detail: str = "" #: Who did it, on audit rows. actor: str = "" + #: The run this happened in: a batch run's id, or the journaled item a live + #: cascade came from. Empty for everything that belongs to neither. + run: str = Field(default="", index=True, max_length=64) class FlowRun(SQLModel, table=True): diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 5901b71..7b69f1c 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -59,7 +59,12 @@ def _fail(message: str) -> int: return 1 -def _unreachable(exc: Exception, note: str = "") -> int: +#: What to try when nothing answered at all. By far the commonest reason is +#: that no engine is running, and the message said only that it was not. +NO_ENGINE = "Is one running? `fluksio serve` starts one." + + +def _unreachable(exc: Exception, note: str = NO_ENGINE) -> int: """The engine did not answer. Say so as a sentence, not a traceback.""" return _fail( f"engine not answering ({type(exc).__name__}: {exc})" @@ -514,7 +519,7 @@ def cmd_run(args: argparse.Namespace, rest: list[str]) -> int: # The run is the engine's, not this command's: it carries on, and its # id is how to find it again. return _unreachable( - exc, f"Run {handle.id} is still on the engine." if handle else "" + exc, f"Run {handle.id} is still on the engine." if handle else NO_ENGINE ) diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 02b52e2..973cd76 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -299,18 +299,22 @@ class Client: kind: str = "failure", limit: int = 10, flow: str = "", + run: str = "", since: Any = None, until: Any = None, ) -> list[dict[str, Any]]: """What went wrong, or who changed what. Newest first. - Engine-wide, and narrowed by flow or by time. What *one run* did is a - question about that run: :attr:`RunHandle.failures` answers it from the - run's own node rows, which carry the traceback anyway. + Engine-wide, and narrowed by flow, by run or by time. ``run`` takes a + run id and answers what the engine recorded during it. For the failure + that ended a run, :attr:`RunHandle.failures` is still the shorter road: + it reads the run's own node rows, which carry the traceback too. """ query: dict[str, Any] = {"kind": kind, "limit": limit} if flow: query["flow"] = flow + if run: + query["run"] = run for name, value in (("since", since), ("until", until)): if value is not None: query[name] = ( diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py index 29e0b45..95331a5 100644 --- a/backend/tests/api/routes/test_observability.py +++ b/backend/tests/api/routes/test_observability.py @@ -312,3 +312,30 @@ def test_events_narrow_to_one_minute( # The upper bound is exclusive, so the failure a minute later is not in it. assert [event["detail"] for event in events] == ["minute-in"] + + +def test_events_narrow_to_one_run( + client: TestClient, superuser_token_headers: dict[str, str], db: Session +) -> None: + """Reading one run's failures without filtering the engine-wide list.""" + ts = datetime.now(UTC) - timedelta(hours=6) + db.add( + EngineEvent(ts=ts, type="node_error", flow=FLOW, detail="mine", run="run-mine") + ) + db.add( + EngineEvent( + ts=ts, type="node_error", flow=FLOW, detail="theirs", run="run-theirs" + ) + ) + # What a live cascade unrelated to any run leaves, and what every row + # written before the column existed looks like. + db.add(EngineEvent(ts=ts, type="node_error", flow=FLOW, detail="neither")) + db.commit() + + events = client.get( + f"{PREFIX}/events", + headers=superuser_token_headers, + params={"flow": FLOW, "run": "run-mine"}, + ).json() + + assert [event["detail"] for event in events] == ["mine"] diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 95c382c..1bfac14 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -598,3 +598,47 @@ def test_a_step_the_x_metric_never_reached_is_left_out( ).json() assert [point[0] for point in answer["lines"][0]["points"]] == [10.0, 20.0, 30.0] + + +def test_a_run_records_the_code_it_started_with_not_the_code_it_was_queued_with(): + """A sweep queues every run at once and the tree moves while it waits. + + `_restamp` is what runs at claim time, so the digest the record keeps is + the one the run actually executed. + """ + service = RunService(controller=_Unusable(), queue=_Unusable()) + with Session(db_engine) as session: + session.add( + Run( + id="stamp-1", + flow="study", + status="queued", + code_digest="at-submit", + created_at=datetime.now(UTC), + ) + ) + session.commit() + run = session.get(Run, "stamp-1") + + assert service._restamp(run, "at-claim") == "at-claim" + + with Session(db_engine) as session: + assert session.get(Run, "stamp-1").code_digest == "at-claim" + + +def test_a_tree_that_did_not_move_is_not_written_again(): + service = RunService(controller=_Unusable(), queue=_Unusable()) + with Session(db_engine) as session: + session.add( + Run( + id="stamp-2", + flow="study", + status="queued", + code_digest="same", + created_at=datetime.now(UTC), + ) + ) + session.commit() + run = session.get(Run, "stamp-2") + + assert service._restamp(run, "same") == "same" diff --git a/backend/tests/flow/test_provision.py b/backend/tests/flow/test_provision.py index 7ade8b8..816eb39 100644 --- a/backend/tests/flow/test_provision.py +++ b/backend/tests/flow/test_provision.py @@ -8,6 +8,7 @@ it should, and that a job nobody ever attached does not sit in the queue. import subprocess import threading +import time from fluksio.flow.placement import Placer from fluksio.flow.provision import SlurmProfile, SlurmProvisioner, load_provisioners @@ -116,6 +117,22 @@ def test_a_machine_that_arrives_clears_the_way_for_the_next_ask(monkeypatch): def test_a_job_that_never_attaches_is_cancelled(monkeypatch): + ssh = FakeSsh() + monkeypatch.setattr(subprocess, "run", ssh) + hpc = cluster(provision_timeout_s=0.01) + + hpc.provision(cpus=2, gpus=0, ram_mb=0) + assert ssh.ran.wait(5) + # Slept rather than set to zero: zero is what says "no deadline" now. + time.sleep(0.02) + hpc.reconcile(set()) + + assert hpc.status()["outstanding"] == [] + assert ["scancel", "4711"] == ssh.calls[-1][0][-2:] + + +def test_no_deadline_waits_for_as_long_as_the_queue_does(monkeypatch): + """0 is "never give up", not "give up now": a cluster can queue for days.""" ssh = FakeSsh() monkeypatch.setattr(subprocess, "run", ssh) hpc = cluster(provision_timeout_s=0) @@ -124,8 +141,8 @@ def test_a_job_that_never_attaches_is_cancelled(monkeypatch): assert ssh.ran.wait(5) hpc.reconcile(set()) - assert hpc.status()["outstanding"] == [] - assert ["scancel", "4711"] == ssh.calls[-1][0][-2:] + assert hpc.status()["outstanding"][0]["job"] == "4711" + assert [argv for argv, _ in ssh.calls if "scancel" in argv] == [] def test_a_refused_submission_says_what_the_cluster_said(monkeypatch): diff --git a/backend/tests/test_metrics.py b/backend/tests/test_metrics.py index 0b78a3a..a628ba5 100644 --- a/backend/tests/test_metrics.py +++ b/backend/tests/test_metrics.py @@ -224,3 +224,27 @@ def test_a_traceback_no_failure_ever_claims_is_dropped() -> None: asyncio.run(collector.flush()) assert collector._tracebacks == {} + + +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. + """ + collector = MetricsCollector(EventBus()) + ts = datetime.now(UTC).replace(second=0, microsecond=0).timestamp() + collector.handle( + { + "type": "node_error", + "flow": FLOW, + "node": NODE, + "error": "ValueError: in a run", + "run": "run-abc123", + "ts": ts, + } + ) + asyncio.run(collector.flush()) + + failure = db.exec(select(EngineEvent).where(EngineEvent.run == "run-abc123")).one() + assert failure.detail.startswith("ValueError: in a run") diff --git a/docs/code/api.md b/docs/code/api.md index 33baa41..39d51f6 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -176,7 +176,7 @@ read one back. | `/observability/timeseries` | executions and failures over a window | | `/observability/flows` | per-flow rollups with a 60-slice trend | | `/observability/runs` | recent cascades, with `?flow=`, `?since=`, `?until=` | -| `/observability/events?kind=failure\|audit` | what went wrong, or who changed what | +| `/observability/events?kind=failure\|audit` | what went wrong, or who changed what; narrows by `?flow=`, `?run=`, `?since=`, `?until=` | | `/observability/dead-letter` | work the engine gave up on | `GET /utils/health/` is the deep health check the container probe uses: it diff --git a/docs/code/workers.md b/docs/code/workers.md index e3c35bc..c1049fe 100644 --- a/docs/code/workers.md +++ b/docs/code/workers.md @@ -204,6 +204,10 @@ never attaches within `provision_timeout_s` is `scancel`led, as is anything outstanding when the engine stops. `--max-idle` is what ends the job at the other end, so an allocation goes back rather than idling to its walltime. +Both take `0` for "no limit": `provision_timeout_s: 0` waits for as long as the +queue does, which is what a cluster that queues overnight needs, and +`max_idle_s: 0` keeps the machine for the job's whole walltime. + `GET /workers/resources` reports what is outstanding and what last went wrong. !!! note "It needs a route out" diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index c367e79..914efce 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -125,6 +125,7 @@ warning into a refusal to start. |---|---|---| | `FLOW_MAX_WORKERS` | `4` | node-code subprocesses run in parallel | | `FLOW_MAX_CASCADES` | `4` | cascades in flight at once; throughput is this over the mean cascade time, so raise it where nodes wait on a network rather than a CPU | +| `FLOW_MAX_RUNS` | `4` | batch runs driven at once. A different limit from the one above: a run drives a whole graph, and its nodes are bounded by `FLOW_MAX_WORKERS`. This is what a sweep queues behind | | `FLOW_NODE_TIMEOUT` | `0` | seconds a node may be silent, unless it sets its own; 0 is no limit | | `FLOW_CPUS` | `0` | cores nodes that declare `resources` may be given; 0 works it out as every core but two, which are what keeps the engine answering while the machine is busy | | `FLOW_GPUS` | `0` | GPUs on this machine, each held by one node at a time. Not detected — say how many there are | @@ -132,6 +133,10 @@ warning into a refusal to start. | `ARTIFACT_GC_INTERVAL_S` | `3600` | how often artifact bytes nothing refers to are swept away; 0 never sweeps | | `ARTIFACT_GC_GRACE_S` | `3600` | how long a freshly written artifact is spared, whatever refers to it | +The three concurrency limits are also flags on `fluksio serve` — `--max-workers`, +`--max-cascades`, `--max-runs` — which outrank the file, and the engine says +which numbers it started with in its first lines. + An artifact is referred to by a run that recorded it or by a message currently holding it; anything else is what a camera published four hours ago, and the sweep is what keeps a flow streaming media from filling the disk. It stands diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 5e38b81..64ea629 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -715,10 +715,14 @@ export const EventRowSchema = { actor: { type: 'string', title: 'Actor' + }, + run: { + type: 'string', + title: 'Run' } }, type: 'object', - required: ['id', 'ts', 'type', 'flow', 'node', 'detail', 'actor'], + required: ['id', 'ts', 'type', 'flow', 'node', 'detail', 'actor', 'run'], title: 'EventRow' } as const; diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 0cb2aba..105c39a 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -1548,9 +1548,13 @@ export class ObservabilityService { * * ``since`` is inclusive and ``until`` exclusive, the same window ``/runs`` * takes, so a list can cover the span the charts beside it are drawn from. + * ``run`` narrows to one run — a batch run's id, or the journaled item a live + * cascade came from. Rows recorded before the column existed carry none, so + * an old failure answers no run at all rather than the wrong one. * @param data The data for the request. * @param data.kind * @param data.flow + * @param data.run * @param data.since * @param data.until * @param data.limit @@ -1564,6 +1568,7 @@ export class ObservabilityService { query: { kind: data.kind, flow: data.flow, + run: data.run, since: data.since, until: data.until, limit: data.limit diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 888eecd..c11f714 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -222,6 +222,7 @@ export type EventRow = { node: string; detail: string; actor: string; + run: string; }; export type FlavorCreate = { @@ -1662,6 +1663,7 @@ export type ObservabilityReadEventsData = { flow?: (string | null); kind?: 'failure' | 'audit'; limit?: number; + run?: (string | null); since?: (string | null); until?: (string | null); };