diff --git a/NOTEPAD.md b/NOTEPAD.md index 157b3aa..932de00 100644 --- a/NOTEPAD.md +++ b/NOTEPAD.md @@ -51,7 +51,6 @@ is what M4 still waits on, together with porting the flows. - CHORE/API: the metrics collector is a bus subscriber, so a storm that overflows the bus queue undercounts. The events dropped are the same ones the websocket drops; exact accounting would need the collector to be fed from the engine rather than the bus. - CHORE/API: `/observability/summary` reports the work queue's `depth` as the Redis stream length, which is the journal size (capped at `STREAM_MAXLEN`) rather than a backlog. The health screen shows `pending` instead; the field name still invites the wrong reading. - FEAT/UI: the health screen's window is fixed at 24 hours and the charts fold minute buckets in Python. A range picker (and `date_bin()` behind it) is the next step if anyone wants a week. -- FEAT/API: `/observability/runs` takes no time range and caps at 200 rows, so picking a minute on the throughput chart can only filter the runs the list happens to hold — on a busy engine that is its last minute or so, and any earlier minute reads as empty. A `since`/`until` parameter would let a moment on a chart reach the whole day. Failures are sparse enough that the same pairing works there. - CHORE/FLOW: run records for a deleted flow stay until the retention window passes, so a flow that no longer exists keeps appearing in the history. Deliberate — it is a record of what ran — but `forget_flow` could offer to clear it. - CHORE/API: nothing can ask the collector to flush now, so anything needing the tables to be current has to wait out `FLUSH_INTERVAL_S` — which is what the soak harness does before clearing its own rows. - BUG/UI: `MemoryWorkQueue.stats()` hard-codes `pending: 0`, so the health tile always reads zero on a stack without Redis. diff --git a/backend/app/api/routes/observability.py b/backend/app/api/routes/observability.py index b9ae34a..ff7e55e 100644 --- a/backend/app/api/routes/observability.py +++ b/backend/app/api/routes/observability.py @@ -97,6 +97,11 @@ def _since(hours: int) -> datetime: return datetime.now(timezone.utc) - timedelta(hours=hours) +def _aware(when: datetime) -> datetime: + """A bound as the columns store it. A naive one is read as UTC.""" + return when if when.tzinfo else when.replace(tzinfo=timezone.utc) + + @router.get("/summary", response_model=HealthSummary) async def read_summary( request: Request, controller: FlowControllerDep, session: SessionDep @@ -275,14 +280,24 @@ def read_runs( session: SessionDep, flow: str | None = None, status: str | None = None, + since: datetime | None = None, + until: datetime | None = None, limit: int = 50, ) -> Any: - """Recent cascades, newest first.""" + """Recent cascades, newest first. + + ``since`` is inclusive and ``until`` exclusive, so a window of one minute + holds exactly the runs of the minute bucket the charts are drawn from. + """ statement = select(FlowRun).order_by(col(FlowRun.started_at).desc()) if flow: statement = statement.where(col(FlowRun.flow) == flow) if status: statement = statement.where(col(FlowRun.status) == status) + if since: + statement = statement.where(col(FlowRun.started_at) >= _aware(since)) + if until: + statement = statement.where(col(FlowRun.started_at) < _aware(until)) return list(session.exec(statement.limit(min(limit, 200)))) diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py index f1cdc98..04f5d28 100644 --- a/backend/tests/api/routes/test_observability.py +++ b/backend/tests/api/routes/test_observability.py @@ -102,3 +102,35 @@ def test_the_history_reads_back( dead = client.get(f"{PREFIX}/dead-letter", headers=superuser_token_headers) assert dead.status_code == 200 assert isinstance(dead.json(), list) + + +def test_runs_narrow_to_one_minute( + client: TestClient, superuser_token_headers: dict[str, str], db: Session +) -> None: + """A minute picked off a chart reaches past what the recent list holds.""" + minute = datetime.now(timezone.utc).replace(second=0, microsecond=0) - timedelta( + hours=3 + ) + db.add(FlowRun(id="minute-in", flow=FLOW, started_at=minute, status="ok")) + db.add( + FlowRun( + id="minute-after", + flow=FLOW, + started_at=minute + timedelta(minutes=1), + status="ok", + ) + ) + db.commit() + + runs = client.get( + f"{PREFIX}/runs", + headers=superuser_token_headers, + params={ + "since": minute.isoformat(), + "until": (minute + timedelta(minutes=1)).isoformat(), + }, + ).json() + + # The upper bound is exclusive, so the run starting the next minute is not + # in this one. + assert [run["id"] for run in runs] == ["minute-in"] diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 856ffd5..18c95e9 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -1147,9 +1147,14 @@ export class ObservabilityService { /** * Read Runs * Recent cascades, newest first. + * + * ``since`` is inclusive and ``until`` exclusive, so a window of one minute + * holds exactly the runs of the minute bucket the charts are drawn from. * @param data The data for the request. * @param data.flow * @param data.status + * @param data.since + * @param data.until * @param data.limit * @returns RunRow Successful Response * @throws ApiError @@ -1161,6 +1166,8 @@ export class ObservabilityService { query: { flow: data.flow, status: data.status, + since: data.since, + until: data.until, limit: data.limit }, errors: { diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index aa4db08..c7d6b3b 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -1072,7 +1072,9 @@ export type ObservabilityReadFlowRollupsResponse = (Array); export type ObservabilityReadRunsData = { flow?: (string | null); limit?: number; + since?: (string | null); status?: (string | null); + until?: (string | null); }; export type ObservabilityReadRunsResponse = (Array); diff --git a/frontend/src/components/Health/HealthActivity.tsx b/frontend/src/components/Health/HealthActivity.tsx index e7fd5b9..eb9f33d 100644 --- a/frontend/src/components/Health/HealthActivity.tsx +++ b/frontend/src/components/Health/HealthActivity.tsx @@ -17,6 +17,7 @@ import { failuresQueryOptions, healthKeys, minuteOf, + minuteRunsQueryOptions, runsQueryOptions, timeseriesQueryOptions, } from "./queries" @@ -174,6 +175,7 @@ export function HealthActivity() { const { data: series } = useQuery(timeseriesQueryOptions()) const { data: runs } = useQuery(runsQueryOptions()) + const { data: pinnedRuns } = useQuery(minuteRunsQueryOptions(runsAt.pinned)) const { data: failures } = useQuery(failuresQueryOptions()) const { data: audit } = useQuery(auditQueryOptions()) const { data: dead } = useQuery(deadLetterQueryOptions()) @@ -201,10 +203,16 @@ export function HealthActivity() { ): HistoryPoint[] => points.map((point) => ({ ts: point.ts, value: pick(point) })) + // A pin is read back from the server, so it reaches a minute the recent list + // is nowhere near deep enough to hold. A hover stays the client-side preview + // it is: scrubbing a day's chart would otherwise be a request per minute the + // pointer rests on. const shownRuns = - runsAt.at === null - ? (runs ?? []).slice(0, RUNS_SHOWN) - : (runs ?? []).filter((run) => minuteOf(run.started_at) === runsAt.at) + runsAt.pinned !== null + ? (pinnedRuns ?? []) + : runsAt.at === null + ? (runs ?? []).slice(0, RUNS_SHOWN) + : (runs ?? []).filter((run) => minuteOf(run.started_at) === runsAt.at) const shownFailures = failuresAt.at === null ? (failures ?? []).slice(0, FAILURES_SHOWN) @@ -282,7 +290,9 @@ export function HealthActivity() {

{runsAt.at === null ? "No runs recorded yet." - : "No run from this minute is in the recent list."} + : runsAt.pinned !== null + ? "Nothing ran in this minute." + : "No run from this minute is in the recent list."}

)} diff --git a/frontend/src/components/Health/queries.ts b/frontend/src/components/Health/queries.ts index 97cf5fc..56430e2 100644 --- a/frontend/src/components/Health/queries.ts +++ b/frontend/src/components/Health/queries.ts @@ -9,13 +9,10 @@ const REFRESH = 30_000 /** * How deep the run and failure lists are read. * - * Deeper than they are shown: picking a minute off a chart filters these rows + * Deeper than they are shown: hovering a minute on a chart previews these rows * client-side, and a list holding only the newest handful would have nothing - * to find for any minute but the current one. - * - * ponytail: the runs endpoint caps at 200 and takes no time range, so a busy - * engine still only covers its last minute or so. A `since` parameter is what - * would let a moment on the chart reach the whole day. + * to show for any minute but the current one. Pinning asks the server for the + * minute instead, which is what reaches past this depth. */ const RUN_DEPTH = 200 const EVENT_DEPTH = 100 @@ -53,6 +50,26 @@ export const runsQueryOptions = () => ({ refetchInterval: REFRESH, }) +/** + * The runs of one minute, wherever it sits in the day. + * + * A busy engine writes more runs per minute than the recent list is deep, so a + * pinned moment is asked for rather than filtered out of what is already held. + * `at` is a minute start, and the window is that minute. + */ +export const minuteRunsQueryOptions = (at: number | null) => ({ + queryKey: ["observability", "runs", at] as const, + queryFn: () => + ObservabilityService.readRuns({ + since: new Date((at ?? 0) * 1000).toISOString(), + until: new Date(((at ?? 0) + 60) * 1000).toISOString(), + limit: RUN_DEPTH, + }), + // A minute that has passed does not change, and the current one is refreshed + // by the unpinned list anyway. + enabled: at !== null, +}) + export const failuresQueryOptions = () => ({ queryKey: [...healthKeys.events, "failure"] as const, queryFn: () =>