diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index 2aebecfe5..2e49bf3e8 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -1492,6 +1492,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"]; @@ -3429,12 +3433,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 aa10e0c86..759cd2417 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; @@ -89,14 +89,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 00f3bf515..ac716e189 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 472327f34..3e178777a 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 726975e85..8a981a57b 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 52ed1644f..3c1039f71 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_reads.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_reads.rs @@ -3,8 +3,11 @@ use crate::authorization::main::assign_builtin_role::assign_builtin_role; use crate::authorization::main::bootstrap_user::bootstrap_user; use crate::credentials::main::identities::create_principal::create_principal; -use crate::scheduler::tests::contracts::ticks::helpers::calendar_text; +use crate::declarations::main::list_person_triggers::list_person_triggers; +use crate::scheduler::tests::contracts::ticks::helpers::{calendar_text, calendar_time}; use crate::snapshots::main::list_person_snapshots::list_person_snapshots; +use crate::store::main::{begin::begin, commit::commit, query::query}; +use crate::store::models::{Parameter, TransactionMode}; use std::collections::BTreeMap; use std::time::Duration; @@ -49,13 +52,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 { @@ -322,6 +325,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(), @@ -329,9 +345,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" + ); + } let mut extra = input.clone(); extra.generation += 1; extra.nodes.remove("products"); diff --git a/ui/src/lib/api/http-client.test.ts b/ui/src/lib/api/http-client.test.ts index 8d05b8dc3..9ae9cd852 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 03baee8ca..523de3dc4 100644 --- a/ui/src/lib/api/http-client.ts +++ b/ui/src/lib/api/http-client.ts @@ -209,6 +209,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); @@ -217,6 +218,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); @@ -235,6 +237,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); @@ -309,7 +312,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(); @@ -317,7 +335,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(); @@ -353,8 +374,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 { @@ -367,12 +387,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 79361e5e0..db4da3e20 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 {