From d2e1fb8a48566feecc100f2e90c4ba7797930a17 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Sun, 8 Nov 2026 08:05:59 +0000 Subject: [PATCH] fix(ui): stop a burst of control API requests on every live refresh The person trigger listing now carries each trigger's newest tick, so the jobs and triggers pages no longer read ticks once per trigger. Pages coalesce change notices into one reload, and the HTTP client shares an in-flight listing between pages that refresh together. --- clients/typescript/src/schema.d.ts | 7 ++- .../api/_helpers/resources/person_reads.rs | 6 +-- .../src/declarations/_helpers/reads.rs | 14 ++++++ .../declarations/main/list_person_triggers.rs | 21 ++++++-- .../src/declarations/models.rs | 8 ++++ .../tests/api/test_reads.rs | 48 +++++++++++++++++-- ui/src/lib/api/http-client.test.ts | 42 +++++++++++++++- ui/src/lib/api/http-client.ts | 33 +++++++++++-- ui/src/lib/state/resource.svelte.ts | 37 +++++++++++++- 9 files changed, 196 insertions(+), 20 deletions(-) diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index 117fb8c855..fa63012262 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -1476,6 +1476,10 @@ export interface components { /** Format: uuid */ principal_id: string; }; + /** @description A trigger as people list it, with its newest tick so a list needs no read per trigger. */ + PersonTriggerRecord: components["schemas"]["TriggerRecord"] & { + latest_tick?: null | components["schemas"]["TickRecord"]; + }; /** @description Observed physical evidence is not a live warehouse-existence guarantee. */ PhysicalEvidence: { observation: components["schemas"]["CatalogObservation"]; @@ -3350,12 +3354,13 @@ export interface operations { }; requestBody?: never; responses: { + /** @description Published triggers with their newest tick */ 200: { headers: { [name: string]: unknown; }; content: { - "application/json": components["schemas"]["TriggerRecord"][]; + "application/json": components["schemas"]["PersonTriggerRecord"][]; }; }; /** @description Session rejected */ diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs index 79fbb670ad..e4c055796c 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs @@ -10,7 +10,7 @@ use crate::api::models::ReadQuery; use crate::credentials::models::{AuditContext, AuthenticatedPerson, DefaultRole}; use crate::declarations::main::list_person_jobs::list_person_jobs; use crate::declarations::main::list_person_triggers::list_person_triggers; -use crate::declarations::models::{JobRecord, TriggerRecord}; +use crate::declarations::models::{JobRecord, PersonTriggerRecord}; use crate::runs::main::list_person_run_nodes::list_person_run_nodes; use crate::runs::models::RunNodeRecord; use crate::scheduler::main::list_person_ticks::list_person_ticks; @@ -87,14 +87,14 @@ pub(crate) async fn jobs( #[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/triggers", operation_id = "listPersonTriggers", params(("location_id" = Uuid, Path), ("target" = String, Path)), - responses((status = 200, body = Vec), (status = 401, description = "Session rejected"), + responses((status = 200, body = Vec, description = "Published triggers with their newest tick"), (status = 401, description = "Session rejected"), (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))] pub(crate) async fn triggers( (State(store), Extension(context)): (State, Extension), Extension(default_role): Extension, Path((location_id, target)): Path<(Uuid, String)>, Extension(AuthenticatedPerson(principal, _)): Extension, -) -> Result>, ApiError> { +) -> Result>, ApiError> { Ok(Json( list_person_triggers( &store, diff --git a/crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs b/crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs index 00f3bf5150..ac716e189b 100644 --- a/crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs +++ b/crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs @@ -3,6 +3,7 @@ use uuid::Uuid; use crate::declarations::models::{JobRecord, StoredJob, TriggerRecord}; +use crate::scheduler::models::TickRecord; use crate::store::errors::ControlError; use crate::store::main::query_records::query_records; use crate::store::models::{ControlTransaction, Parameter}; @@ -42,3 +43,16 @@ pub(crate) async fn jobs_in( .collect(); Ok((transaction, jobs)) } + +/// The newest tick of every published trigger in the target, each found through the trigger-time index. +pub(crate) async fn latest_ticks_in( + transaction: ControlTransaction, + (location_id, target): (Uuid, &str), +) -> Result<(ControlTransaction, Vec), ControlError> { + query_records(transaction, + "SELECT k.tick_id, k.trigger_id, k.scheduled_for, k.evaluated_at, k.epoch, k.outcome, k.run_id, k.covered_by_run_id \ + FROM sqb_triggers t JOIN sqb_ticks k ON k.tick_id=(SELECT x.tick_id FROM sqb_ticks x WHERE x.trigger_id=t.trigger_id \ + ORDER BY x.scheduled_for DESC LIMIT 1) \ + WHERE t.location_id=$1 AND t.target=$2 AND t.removed_in_snapshot_id IS NULL", + vec![Parameter::Id(location_id), Parameter::Text(target.into())]).await +} diff --git a/crates/sqlbuild-orchestration/src/declarations/main/list_person_triggers.rs b/crates/sqlbuild-orchestration/src/declarations/main/list_person_triggers.rs index 472327f34c..3e178777ac 100644 --- a/crates/sqlbuild-orchestration/src/declarations/main/list_person_triggers.rs +++ b/crates/sqlbuild-orchestration/src/declarations/main/list_person_triggers.rs @@ -1,11 +1,13 @@ //! A person lists triggers only with view on every current node of the target. +use std::collections::BTreeMap; + use uuid::Uuid; use crate::authorization::main::authorize_listing::authorize_listing; use crate::credentials::models::{AuditContext, DefaultRole}; -use crate::declarations::_helpers::reads::triggers_in; -use crate::declarations::models::TriggerRecord; +use crate::declarations::_helpers::reads::{latest_ticks_in, triggers_in}; +use crate::declarations::models::PersonTriggerRecord; use crate::store::errors::ControlError; use crate::store::main::{begin::begin, commit::commit}; use crate::store::models::{ControlStore, TransactionMode}; @@ -15,7 +17,7 @@ pub async fn list_person_triggers( (context, credentials): (&AuditContext, Uuid), (location_id, target): (Uuid, &str), default_role: DefaultRole, -) -> Result, ControlError> { +) -> Result, ControlError> { let (transaction, _) = authorize_listing( begin(store, TransactionMode::Write).await?, (context, credentials), @@ -24,6 +26,17 @@ pub async fn list_person_triggers( ) .await?; let (transaction, triggers) = triggers_in(transaction, (location_id, target)).await?; + let (transaction, ticks) = latest_ticks_in(transaction, (location_id, target)).await?; commit(transaction).await?; - Ok(triggers) + let mut latest: BTreeMap = ticks + .into_iter() + .map(|tick| (tick.trigger_id, tick)) + .collect(); + Ok(triggers + .into_iter() + .map(|trigger| PersonTriggerRecord { + latest_tick: latest.remove(&trigger.trigger_id), + trigger, + }) + .collect()) } diff --git a/crates/sqlbuild-orchestration/src/declarations/models.rs b/crates/sqlbuild-orchestration/src/declarations/models.rs index 726975e856..8a981a57b4 100644 --- a/crates/sqlbuild-orchestration/src/declarations/models.rs +++ b/crates/sqlbuild-orchestration/src/declarations/models.rs @@ -119,6 +119,14 @@ pub struct TriggerRecord { pub next_due_at: Option>, } +/// A trigger as people list it, with its newest tick so a list needs no read per trigger. +#[derive(Clone, Debug, PartialEq, Serialize, utoipa::ToSchema)] +pub struct PersonTriggerRecord { + #[serde(flatten)] + pub trigger: TriggerRecord, + pub latest_tick: Option, +} + /// Operator changes are immutable facts; clearing an override restores the published code default. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] diff --git a/crates/sqlbuild-orchestration/tests/api/test_reads.rs b/crates/sqlbuild-orchestration/tests/api/test_reads.rs index e3adc23031..2dabd755c2 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_reads.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_reads.rs @@ -1,12 +1,16 @@ //! Real HTTP reads keep both location and target scope, including guessed nested resource IDs. -use crate::scheduler::tests::contracts::ticks::helpers::calendar_text; +use crate::authorization::main::bootstrap_user::bootstrap_user; +use crate::declarations::main::list_person_triggers::list_person_triggers; +use crate::scheduler::tests::contracts::ticks::helpers::{calendar_text, calendar_time}; +use crate::store::main::{begin::begin, commit::commit, query::query}; +use crate::store::models::{Parameter, TransactionMode}; use std::collections::BTreeMap; use std::time::Duration; use crate::credentials::main::tokens::issue_token::issue_token; use crate::credentials::main::tokens::revoke_token::revoke_token; -use crate::credentials::models::{AuditContext, CredentialScope}; +use crate::credentials::models::{AuditContext, CredentialScope, DefaultRole}; use crate::declarations::main::change_trigger::change_trigger; use crate::declarations::main::list_triggers::list_triggers; use crate::declarations::models::{TriggerChange, TriggerChangeRequest}; @@ -43,13 +47,13 @@ fn given_published_locations_and_scheduled_work_when_read_through_http_then_reso description: "SQLite scoped HTTP reads", backend: Backend::Sqlite, expected_runs: 3, - expected_ticks: 1, + expected_ticks: 2, }, ApiReadTestCase { description: "PostgreSQL scoped HTTP reads", backend: Backend::Postgres, expected_runs: 3, - expected_ticks: 1, + expected_ticks: 2, }, ]; for test_case in test_cases { @@ -316,6 +320,19 @@ fn given_published_locations_and_scheduled_work_when_read_through_http_then_reso let events = read_json(&client, (&event_path, &token.bearer)).await; assert_eq!(events.as_array().expect("events").len(), 1); assert_eq!(events[0]["request_json"], json!(change)); + let (transaction, _) = query( + begin(store, TransactionMode::Write).await.expect("seed older tick"), + "INSERT INTO sqb_ticks (tick_id,trigger_id,scheduled_for,evaluated_at,epoch,outcome) \ + VALUES ($1,$2,$3,$3,1,'covered')", + vec![ + Parameter::Id(Uuid::now_v7()), + Parameter::Id(trigger_id), + Parameter::Time(calendar_time(1)), + ], + ) + .await + .expect("older tick"); + commit(transaction).await.expect("commit older tick"); let ticks = read_json(&client, (&tick_path, &token.bearer)).await; assert_eq!( ticks.as_array().expect("ticks").len(), @@ -323,6 +340,29 @@ fn given_published_locations_and_scheduled_work_when_read_through_http_then_reso ); assert_eq!(ticks[0]["outcome"], "requested"); assert_eq!(ticks[0]["scheduled_for"], calendar_text(2)); + assert_eq!(ticks[1]["outcome"], "covered"); + let administrator = bootstrap_user(store, &audit, "orders-admin") + .await + .expect("administrator"); + for (target, expected) in [ + ("development", ticks[0].clone()), + ("validation", json!(null)), + ] { + let listed = list_person_triggers( + store, + (&audit, administrator), + (location_id, target), + DefaultRole::None, + ) + .await + .expect("person triggers"); + assert_eq!(listed.len(), 1, "{target}"); + assert_eq!( + json!(listed[0].latest_tick), + expected, + "newest tick only, never another target's" + ); + } for path in [&run_path, &event_path, &tick_path, &invocation_path] { assert_eq!( read_json(&client, (&format!("{path}?limit=1"), &token.bearer)) diff --git a/ui/src/lib/api/http-client.test.ts b/ui/src/lib/api/http-client.test.ts index 8d05b8dc33..9ae9cd852e 100644 --- a/ui/src/lib/api/http-client.test.ts +++ b/ui/src/lib/api/http-client.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { HttpChangeFeed, feedAllowed } from './http-client'; +import { HttpChangeFeed, createHttpClient, feedAllowed } from './http-client'; import type { ChangedSignal, ConnectionState } from './signals'; /** A stand-in `EventSource` the test opens, fails and feeds by hand. */ @@ -237,3 +237,43 @@ describe('feedAllowed', () => { await expect(feedAllowed('', ORDERS, () => {})).rejects.toThrow('network down'); }); }); + +describe('listing requests', () => { + const tick = { tick_id: 'k-2', trigger_id: 't-orders', scheduled_for: '2026-11-07T06:00:00Z', evaluated_at: '2026-11-07T06:00:02Z', epoch: 1, outcome: 'requested', run_id: null, covered_by_run_id: null }; + const declaration = { description: 'Daily orders', job_key: 'orders', schedule: { kind: 'cron', expression: '0 6 * * *', timezone: 'UTC' } }; + const triggers = [ + { trigger_id: 't-orders', name: 'orders_daily', job_key: 'orders', definition_json: declaration, status: 'running', last_tick_for: null, next_due_at: null, latest_tick: tick }, + { trigger_id: 't-stock', name: 'stock_daily', job_key: 'stock', definition_json: declaration, status: 'running', last_tick_for: null, next_due_at: null, latest_tick: null } + ]; + + function server() { + const calls: string[] = []; + vi.stubGlobal('document', { cookie: '' }); + vi.stubGlobal( + 'fetch', + vi.fn(async (request: Request) => { + const path = new URL(request.url).pathname.replace('/api/control/locations/0191f0aa-orders/targets/development', ''); + calls.push(`${request.method} ${path}`); + const body = request.method === 'POST' ? { applied: true } : path === '/triggers' ? triggers : []; + return new Response(JSON.stringify(body), { status: 200, headers: { 'content-type': 'application/json' } }); + }) + ); + return { calls, client: createHttpClient({ baseUrl: 'http://ui.test', onUnauthorized: () => {} }) }; + } + + it('lists triggers with their newest tick in one request, never one per trigger', async () => { + const { calls, client } = server(); + const listed = await client.listTriggers(ORDERS); + expect(listed.map((t) => t.lastTick?.id ?? null)).toEqual(['k-2', null]); + expect(calls.sort()).toEqual(['GET /runs', 'GET /triggers']); + }); + + it('shares listings between pages that refresh together, and reads again after a write', async () => { + const { calls, client } = server(); + await Promise.all([client.listJobs(ORDERS), client.listTriggers(ORDERS)]); + expect(calls.sort()).toEqual(['GET /jobs', 'GET /runs', 'GET /triggers']); + calls.length = 0; + await client.setTriggerStates(ORDERS, ['t-orders'], 'paused'); + expect([calls[0], ...calls.slice(1).sort()]).toEqual(['POST /api/control/triggers', 'GET /runs', 'GET /triggers']); + }); +}); diff --git a/ui/src/lib/api/http-client.ts b/ui/src/lib/api/http-client.ts index 70a1df67b9..8698a32f8a 100644 --- a/ui/src/lib/api/http-client.ts +++ b/ui/src/lib/api/http-client.ts @@ -207,6 +207,7 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt async cancelRun(scope: Scope, runId: string): Promise { const body = { run_id: runId, event_id: crypto.randomUUID(), reason: 'Cancelled from the web UI' }; const { error, response } = await http.POST('/api/control/runs/cancel', { body, params: { header: { 'x-sqb-csrf': csrfToken(document.cookie) } } }); + inFlight.clear(); if (response.status === 401) onUnauthorized(); if (!response.ok) throw new ApiError(errorText(error, response), response.status); return requireRun(scope, runId); @@ -215,6 +216,7 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt const source = await requireRun(scope, runId); const body = { run_id: crypto.randomUUID(), source_run_id: runId, from_failure: mode === 'from_failure', priority: source.priority, reason: 'Rerun from the web UI' }; const { error, response } = await http.POST('/api/control/runs/rerun', { body, params: { header: { 'x-sqb-csrf': csrfToken(document.cookie) } } }); + inFlight.clear(); if (response.status === 401) onUnauthorized(); if (!response.ok) throw new ApiError(errorText(error, response), response.status); return requireRun(scope, body.run_id); @@ -233,6 +235,7 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt const { locationId, target } = parseLocationKey(scope.locationId); const body = { run_id: crypto.randomUUID(), location_id: locationId, target, job_key: key, priority: 0, reason: 'Started from the web UI' }; const { error, response } = await http.POST('/api/control/runs', { body, params: { header: { 'x-sqb-csrf': csrfToken(document.cookie) } } }); + inFlight.clear(); if (response.status === 401) onUnauthorized(); if (!response.ok) throw new ApiError(errorText(error, response), response.status); return requireRun(scope, body.run_id); @@ -297,7 +300,22 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt } }; - async function listedJobs(scope: Scope) { + /** + * One request per listing at a time: pages that refresh together on a change notice share it + * instead of each asking the server again. A write clears it, so reads after a write see it. + */ + const inFlight = new Map>(); + function shared(key: string, load: () => Promise): Promise { + const pending = inFlight.get(key) as Promise | undefined; + if (pending) return pending; + const started: Promise = load().finally(() => inFlight.get(key) === started && inFlight.delete(key)); + inFlight.set(key, started); + return started; + } + + const listedJobs = (scope: Scope) => shared(`jobs ${scope.locationId}`, () => fetchJobs(scope)); + + async function fetchJobs(scope: Scope) { const { locationId, target } = parseLocationKey(scope.locationId); const { data, error, response } = await http.GET('/api/control/locations/{location_id}/targets/{target}/jobs', { params: { path: { location_id: locationId, target } } }); if (response.status === 401) onUnauthorized(); @@ -305,7 +323,10 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt return data; } - async function listedTriggers(scope: Scope) { + const listedTriggers = (scope: Scope) => shared(`triggers ${scope.locationId}`, () => fetchTriggers(scope)); + + /** Published triggers, each with its newest tick. */ + async function fetchTriggers(scope: Scope) { const { locationId, target } = parseLocationKey(scope.locationId); const { data, error, response } = await http.GET('/api/control/locations/{location_id}/targets/{target}/triggers', { params: { path: { location_id: locationId, target } } }); if (response.status === 401) onUnauthorized(); @@ -341,8 +362,7 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt /** Triggers with their latest tick and the newest listed run each started. */ async function triggersOf(scope: Scope): Promise { const [wires, runs] = await Promise.all([listedTriggers(scope), recentRuns(scope)]); - const latest = await Promise.all(wires.map(async (t) => (await ticksOf(scope, t.trigger_id, 1)).map(toTick)[0] ?? null)); - return wires.map((t, i) => toTrigger(t, runs, latest[i])); + return wires.map((t) => toTrigger(t, runs, t.latest_tick ? toTick(t.latest_tick) : null)); } async function requireTrigger(scope: Scope, id: string): Promise { @@ -355,12 +375,15 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt async function changeTrigger(triggerId: string, change: Wire['TriggerChange'], reason?: string | null) { const body = { trigger_id: triggerId, event_id: crypto.randomUUID(), actor: '', reason: reason?.trim() || 'Changed from the web UI', change }; const { error, response } = await http.POST('/api/control/triggers', { body, params: { header: { 'x-sqb-csrf': csrfToken(document.cookie) } } }); + inFlight.clear(); if (response.status === 401) onUnauthorized(); if (!response.ok) throw new ApiError(errorText(error, response), response.status); } + const listedRuns = (scope: Scope) => shared(`runs ${scope.locationId}`, () => fetchRuns(scope)); + /** The newest runs of the location, up to the server's cap. */ - async function listedRuns(scope: Scope) { + async function fetchRuns(scope: Scope) { const { locationId, target } = parseLocationKey(scope.locationId); const { data, error, response } = await http.GET('/api/control/locations/{location_id}/targets/{target}/runs', { params: { path: { location_id: locationId, target }, query: { limit: RUN_WINDOW } } diff --git a/ui/src/lib/state/resource.svelte.ts b/ui/src/lib/state/resource.svelte.ts index 79361e5e0f..db4da3e202 100644 --- a/ui/src/lib/state/resource.svelte.ts +++ b/ui/src/lib/state/resource.svelte.ts @@ -2,6 +2,9 @@ import type { ApiClient, Scope } from '$lib/api'; import { channelMatches } from '$lib/api'; import { app } from './app.svelte'; +/** How long a burst of change signals may last before one reload starts. */ +const SIGNAL_PAUSE_MS = 300; + export type Resource = { readonly data: T | undefined; readonly error: string | null; @@ -13,6 +16,10 @@ export type Resource = { * Loads data for the selected location and target, reloading when the scope * changes and when a "changed" signal arrives on one of `topics`. * + * Change signals arrive in bursts while a run progresses, so they are coalesced: + * one reload starts after a short pause, and signals during a reload queue at + * most one more. + * * Must be called during component initialisation. Reactive values read * synchronously inside `load` (such as route params) are also dependencies. */ @@ -24,25 +31,46 @@ export function createResource( let error = $state(null); let loading = $state(true); let latest = 0; + let settled = 0; + let reloadAgain = false; + let pause: ReturnType | null = null; function run(scope: Scope) { const request = ++latest; loading = true; + const done = () => { + settled = request; + if (reloadAgain && request === latest) { + reloadAgain = false; + refresh(); + } + }; load(app.client, scope).then( (value) => { if (request !== latest) return; data = value; error = null; loading = false; + done(); }, (reason: unknown) => { if (request !== latest) return; error = reason instanceof Error ? reason.message : String(reason); loading = false; + done(); } ); } + function signalled() { + if (pause) return; + pause = setTimeout(() => { + pause = null; + if (settled === latest) refresh(); + else reloadAgain = true; + }, SIGNAL_PAUSE_MS); + } + function refresh() { if (app.scope) run(app.scope); } @@ -54,9 +82,14 @@ export function createResource( $effect(() => { if (!topics.length) return; - return app.client.changes.subscribe((signal) => { - if (channelMatches(signal.channel, topics)) refresh(); + const unsubscribe = app.client.changes.subscribe((signal) => { + if (channelMatches(signal.channel, topics)) signalled(); }); + return () => { + unsubscribe(); + if (pause) clearTimeout(pause); + pause = null; + }; }); return {