diff --git a/.fallowrc.json b/.fallowrc.json index 9d4ff779..bd6f2abc 100644 --- a/.fallowrc.json +++ b/.fallowrc.json @@ -54,6 +54,7 @@ "@fontsource-variable/geist-mono", "recharts", "@base-ui/react", + "@tanstack/react-query", "@tanstack/react-table", "@tanstack/react-virtual", // Recharts' peer dependency: installed for it, never imported by the console itself. diff --git a/e2e/console/actions.spec.ts b/e2e/console/actions.spec.ts index ad4067b9..6fbb7265 100644 --- a/e2e/console/actions.spec.ts +++ b/e2e/console/actions.spec.ts @@ -546,6 +546,8 @@ test.describe("the console acts", () => { try { await page.clock.install(); await openPage(page, fleet.gateway, "/fleet", "Fleet"); + // why: the page tells its views of an answer through a timer, which a frozen clock never fires, so a table still loading when time freezes never shows. + await expect(rowOf(page, "toast-d")).toBeVisible(); // why: with time frozen, only the fast-forwards below age a toast, however slow the machine. await page.clock.pauseAt(new Date(Date.now() + 1_000)); const toasts = page.getByRole("region", { name: "Notifications" }); @@ -555,10 +557,16 @@ test.describe("the console acts", () => { await rowOf(page, worker) .getByRole("button", { name: `Actions for ${worker}` }) .press("Enter"); - await page + const entry = page .getByRole("menu", { name: `Actions for ${worker}` }) - .getByRole("menuitem", { name: item, exact: true }) - .click(); + .getByRole("menuitem", { name: item, exact: true }); + // why: the answer to the read an earlier action set off lands in real time, and the page shows it only once the frozen clock moves again, so until the menu offers the item the clock moves a little at a time. + let tries = 0; + await expect(async () => { + if (tries++ > 0) await page.clock.runFor(500); + await expect(entry).toBeVisible({ timeout: 1_000 }); + }).toPass({ timeout: 20_000 }); + await entry.click(); // why: a pointer left on the clicked item can sit where the toast appears, and a hovered // toast never times out. await page.mouse.move(0, 0); diff --git a/e2e/console/activity.spec.ts b/e2e/console/activity.spec.ts index f0087b79..04ff66bc 100644 --- a/e2e/console/activity.spec.ts +++ b/e2e/console/activity.spec.ts @@ -10,6 +10,7 @@ import { settled, test, } from "./fixtures.js"; +import { fetchFull } from "./full-fetch.js"; const IOS = { availableOsVersions: ["18.4"], knownModels: ["iPhone 16"] }; @@ -555,7 +556,7 @@ test.describe("the Activity page", () => { return; } // why: the page holds only the newest two events, so the rest must come from the daemon's older page. - const body = (await (await route.fetch()).json()) as { + const body = (await (await fetchFull(route)).json()) as { events: Array; }; all = body.events.toSorted((a, b) => b.seq - a.seq); diff --git a/e2e/console/full-fetch.ts b/e2e/console/full-fetch.ts new file mode 100644 index 00000000..ef335594 --- /dev/null +++ b/e2e/console/full-fetch.ts @@ -0,0 +1,8 @@ +import type { Route } from "@playwright/test"; + +// why: the console sends back the ETag of its last answer, and `route.fetch()` forwards the console's +// headers, so a stub that edits the daemon's answer would be handed a 304 with no body to edit. +export function fetchFull(route: Route): ReturnType { + const { "if-none-match": _tag, ...headers } = route.request().headers(); + return route.fetch({ headers }); +} diff --git a/e2e/console/inbox-stub.ts b/e2e/console/inbox-stub.ts index 19f94b29..1f873468 100644 --- a/e2e/console/inbox-stub.ts +++ b/e2e/console/inbox-stub.ts @@ -1,5 +1,7 @@ import type { Page, Route } from "@playwright/test"; +import { fetchFull } from "./full-fetch.js"; + /** why: an Inbox item the daemon cannot make quickly (a request that waited minutes, a lease about to end), in the shape `GET /v1/inbox` sends. */ export type StubbedItem = (now: number) => Record; @@ -71,7 +73,7 @@ export async function addInboxItems( await page.route( (url) => url.pathname === "/v1/inbox", async (route: Route) => { - const response = await route.fetch(); + const response = await fetchFull(route); const now = new Date(response.headers().date ?? "").getTime(); const body = (await response.json()) as { items: unknown[] }; await route.fulfill({ diff --git a/e2e/console/long-lists.spec.ts b/e2e/console/long-lists.spec.ts index 9bf7115e..d35f47bf 100644 --- a/e2e/console/long-lists.spec.ts +++ b/e2e/console/long-lists.spec.ts @@ -11,6 +11,7 @@ import { startDaemon, test as base, } from "./fixtures.js"; +import { fetchFull } from "./full-fetch.js"; import { returnToTab } from "./tab.js"; const IOS = { availableOsVersions: ["18.4"], knownModels: ["iPhone 16"] }; @@ -121,7 +122,7 @@ async function answer(page: Page, pathname: string, body: () => unknown): Promis await page.route( (url) => url.pathname === pathname, async (route: Route) => { - const response = await route.fetch(); + const response = await fetchFull(route); await route.fulfill({ json: body(), response }); }, ); diff --git a/e2e/console/overview.spec.ts b/e2e/console/overview.spec.ts index a91938f8..a524d59e 100644 --- a/e2e/console/overview.spec.ts +++ b/e2e/console/overview.spec.ts @@ -12,6 +12,7 @@ import { startFleet, test, } from "./fixtures.js"; +import { fetchFull } from "./full-fetch.js"; const IOS = { availableOsVersions: ["18.4"], knownModels: ["iPhone 16"] }; const ONE_DEVICE = { limits: { ios: { maxDevices: 1, maxRunning: 1 }, maxRunning: 1 } }; @@ -445,7 +446,7 @@ test.describe("the Overview's five numbers", () => { await page.route( (url) => url.pathname === "/v1/lease-requests", async (route: Route) => { - const response = await route.fetch(); + const response = await fetchFull(route); const now = new Date(response.headers().date ?? "").getTime(); await route.fulfill({ json: { @@ -467,7 +468,7 @@ test.describe("the Overview's five numbers", () => { await page.route( (url) => url.pathname === "/v1/leases", async (route: Route) => { - const response = await route.fetch(); + const response = await fetchFull(route); const now = new Date(response.headers().date ?? "").getTime(); const { leases } = (await response.json()) as { leases: { id: string; ttlDeadline: number }[]; diff --git a/e2e/console/update.spec.ts b/e2e/console/update.spec.ts index daeefeb5..3d743df5 100644 --- a/e2e/console/update.spec.ts +++ b/e2e/console/update.spec.ts @@ -1,6 +1,7 @@ import type { Page } from "@playwright/test"; import { disposeFleet, expect, type RunningDaemon, signIn, startFleet, test } from "./fixtures.js"; +import { fetchFull } from "./full-fetch.js"; const IOS = { availableOsVersions: ["18.4"], knownModels: ["iPhone 16"] }; @@ -12,7 +13,7 @@ async function showWorkerFields(page: Page, fields: () => WorkerFields): Promise await page.route( (url) => url.pathname === "/v1/workers", async (route) => { - const response = await route.fetch(); + const response = await fetchFull(route); const body = (await response.json()) as { workers: Record[] }; await route.fulfill({ json: { ...body, workers: body.workers.map((view) => ({ ...view, ...fields() })) }, @@ -41,7 +42,7 @@ test.describe("the console updates a worker", () => { await page.route( (url) => url.pathname === "/v1/status", async (route) => { - const response = await route.fetch(); + const response = await fetchFull(route); const body = (await response.json()) as { daemon: Record }; await route.fulfill({ json: { ...body, daemon: { ...body.daemon, latestVersion: "0.4.0", version: "0.3.0" } }, diff --git a/e2e/console/workers-answers.ts b/e2e/console/workers-answers.ts index a342fcb2..fb5ca8f0 100644 --- a/e2e/console/workers-answers.ts +++ b/e2e/console/workers-answers.ts @@ -1,4 +1,5 @@ import type { Page, Route } from "@playwright/test"; +import { fetchFull } from "./full-fetch.js"; interface WorkerPayload { connection?: string; @@ -17,7 +18,7 @@ const CONNECT_ERROR = /ECONNREFUSED|ECONNRESET|socket hang up/i; async function fetchOrDrop(route: Route): Promise> | undefined> { try { - return await route.fetch(); + return await fetchFull(route); } catch (error) { if (!(error instanceof Error) || !CONNECT_ERROR.test(error.message)) throw error; await route.abort().catch(() => undefined); diff --git a/e2e/console/workers-gateway-stops-mid-fetch.spec.ts b/e2e/console/workers-gateway-stops-mid-fetch.spec.ts index 6936b231..d4e938dc 100644 --- a/e2e/console/workers-gateway-stops-mid-fetch.spec.ts +++ b/e2e/console/workers-gateway-stops-mid-fetch.spec.ts @@ -133,6 +133,7 @@ test.describe("what the workers page response helper drops and what it lets thro abort: async () => { aborted = true; }, + request: () => ({ headers: () => ({}) }), fetch: async () => { throw new Error("getaddrinfo ENOTFOUND gateway.invalid"); }, @@ -161,6 +162,7 @@ test.describe("what the workers page response helper drops and what it lets thro abort: async () => { aborted = true; }, + request: () => ({ headers: () => ({}) }), fetch: async () => { throw new Error(message); }, diff --git a/ui/AGENTS.md b/ui/AGENTS.md index eff854d7..8b672c03 100644 --- a/ui/AGENTS.md +++ b/ui/AGENTS.md @@ -73,6 +73,13 @@ phase, the health checks and the clock offset, and does one classified GET for a again, any other failure is the connection lost (`ConnectionLost`), which a page never shows as an error. A refused stream is asked for again a second later. +A read sends back the ETag of the route's last answer (`ApiClient.get(path, { tag })`, kept per key in +`state/queries.ts`); a `304` keeps the cached data object, so nothing re-renders, and one that finds +its data dropped is read again without a tag. A `lease.renewed` event moves the lease's `ttlDeadline`, +`lastRenewedAt` and `ttlMs` in `['leases']` and `['workers']` in place (`model/renewal.ts`) with no read, +and the deadline is kept for 60 s against an older answer. Console browser stubs fetch through +`e2e/console/full-fetch.ts`, which drops `If-None-Match`. + ## Actions An action is described once per kind of item in `model/actions.ts` (`leaseActions`, `requestActions`, diff --git a/ui/src/data/api.test.ts b/ui/src/data/api.test.ts index 06770e8b..7cfd5ac4 100644 --- a/ui/src/data/api.test.ts +++ b/ui/src/data/api.test.ts @@ -227,6 +227,36 @@ describe("createApiClient", () => { expect(await api.get("/v1/workers")).toEqual({ body: { workers: [] }, date }); }); + it("a read with a tag sends it as If-None-Match and answers a 304 as not modified, a 200 with its body and tag", async () => { + const date = "Fri, 02 Oct 2026 12:00:00 GMT"; + const sent: (string | undefined)[] = []; + let answer: () => Response = () => new Response(null, { headers: { Date: date }, status: 304 }); + const fetch: Fetch = async (_input, init) => { + sent.push(new Headers(init.headers).get("If-None-Match") ?? undefined); + return answer(); + }; + const api = createApiClient({ fetch, signOut: () => {}, token: () => "slk_op" }); + + expect(await api.get("/v1/workers", { tag: '"a"' })).toEqual({ date, notModified: true }); + answer = () => + new Response(JSON.stringify({ workers: [] }), { + headers: { Date: date, ETag: '"b"' }, + status: 200, + }); + expect(await api.get("/v1/workers", { tag: '"a"' })).toEqual({ + body: { workers: [] }, + date, + tag: '"b"', + }); + answer = () => new Response(JSON.stringify({ workers: [] }), { status: 200 }); + expect(await api.get("/v1/workers", { tag: undefined })).toEqual({ + body: { workers: [] }, + date: null, + tag: null, + }); + expect(sent).toEqual(['"a"', '"a"', undefined]); + }); + it("/v1/healthz is asked with no token, and only a success status counts as up", async () => { const date = "Fri, 02 Oct 2026 12:00:00 GMT"; const up = answering(200, { ok: true }); diff --git a/ui/src/data/api.ts b/ui/src/data/api.ts index dac19009..910dc972 100644 --- a/ui/src/data/api.ts +++ b/ui/src/data/api.ts @@ -31,6 +31,7 @@ interface SendOptions { readonly signal?: AbortSignal; readonly timeoutMs?: number; readonly body?: string; + readonly ifNoneMatch?: string; } export async function send( @@ -57,6 +58,7 @@ export async function send( credentials: "same-origin", headers: { ...(options.token === undefined ? {} : { Authorization: `Bearer ${options.token}` }), + ...(options.ifNoneMatch === undefined ? {} : { "If-None-Match": options.ifNoneMatch }), ...(options.body === undefined ? {} : { "Content-Type": "application/json" }), }, ...(options.body === undefined ? {} : { body: options.body }), @@ -79,9 +81,19 @@ export interface ApiResponse { readonly date: string | null; } +export interface ReadOptions { + // why: the ETag of the last answer to this route; `undefined` asks for the whole answer. + readonly tag: string | undefined; +} + +export type ConditionalResponse = + | { readonly notModified: true; readonly date: string | null } + | (ApiResponse & { readonly notModified?: false; readonly tag: string | null }); + export interface ApiClient { getJson(path: ApiPath): Promise; get(path: ApiPath): Promise>; + get(path: ApiPath, options: ReadOptions): Promise>; // why: a stream's bound, 10 seconds unless `headTimeoutMs` says otherwise, covers its headers // only; the body stays open until it ends or `signal` aborts it. stream( @@ -131,7 +143,10 @@ export function createApiClient(options: ApiClientOptions): ApiClient { ): Promise => { try { return await send(path, sendOptions, async (response) => { - await refuseError(response, refuse); + // why: a 304 answers the tag this read sent, so it is not a refusal. + if (response.status !== 304 || sendOptions.ifNoneMatch === undefined) { + await refuseError(response, refuse); + } return read(response); }); } catch (error: unknown) { @@ -144,11 +159,26 @@ export function createApiClient(options: ApiClientOptions): ApiClient { throw error; } }; - const get = (path: ApiPath) => - watched(path, authorized(), async (response) => ({ - body: (await response.json()) as T, - date: response.headers.get("Date"), - })); + function get(path: ApiPath): Promise>; + function get(path: ApiPath, readOptions: ReadOptions): Promise>; + function get( + path: ApiPath, + readOptions?: ReadOptions, + ): Promise | ConditionalResponse> { + const tag = readOptions?.tag; + return watched( + path, + { ...authorized(), ...(tag === undefined ? {} : { ifNoneMatch: tag }) }, + async (response) => { + const date = response.headers.get("Date"); + if (response.status === 304) return { date, notModified: true as const }; + const body = (await response.json()) as T; + return readOptions === undefined + ? { body, date } + : { body, date, tag: response.headers.get("ETag") }; + }, + ); + } return { get, getJson: async (path: ApiPath) => (await get(path)).body, diff --git a/ui/src/data/live/connection.ts b/ui/src/data/live/connection.ts index e380d6f7..8d8c8f76 100644 --- a/ui/src/data/live/connection.ts +++ b/ui/src/data/live/connection.ts @@ -1,4 +1,11 @@ -import { ApiError, type ApiClient, type ApiPath, RequestTimeoutError } from "../api"; +import { + ApiError, + type ApiClient, + type ApiPath, + type ConditionalResponse, + type ReadOptions, + RequestTimeoutError, +} from "../api"; import { createSseParser, type SseMessage } from "./sse"; import { clockOffset } from "./clock"; @@ -72,6 +79,16 @@ export function isRefusal(error: unknown): boolean { return error instanceof ApiError || error instanceof SyntaxError; } +export type ConditionalRead = + | { readonly notModified: true } + | { readonly notModified?: false; readonly body: T; readonly tag: string | null }; + +function conditional(response: ConditionalResponse): ConditionalRead { + return response.notModified === true + ? { notModified: true } + : { body: response.body, tag: response.tag }; +} + export class LiveConnection { readonly #api: LiveApi; readonly #clock: Clock; @@ -108,11 +125,11 @@ export class LiveConnection { // why: one GET with the classification every read gets: a refusal is the route's own answer and // the daemon is there; a timeout while the daemon starts is asked again; anything else is the // connection lost. - async read(path: ApiPath): Promise { + async read(path: ApiPath, options: ReadOptions): Promise> { for (;;) { let response; try { - response = await this.#api.get(path); + response = await this.#api.get(path, options); } catch (error: unknown) { if (this.#disposed) throw error; switch (this.#classify(error)) { @@ -133,7 +150,7 @@ export class LiveConnection { this.#answeredAt = this.#clock.now(); this.#daemonAnswered(); } - return response.body; + return conditional(response); } } diff --git a/ui/src/data/live/test-support.ts b/ui/src/data/live/test-support.ts index 4d22b37c..52ce171a 100644 --- a/ui/src/data/live/test-support.ts +++ b/ui/src/data/live/test-support.ts @@ -1,6 +1,6 @@ import { vi } from "vitest"; -import { type ApiPath, type ApiResponse } from "../api"; +import { type ApiPath, type ApiResponse, type ConditionalResponse, type ReadOptions } from "../api"; import { type Clock, LiveConnection, @@ -39,28 +39,48 @@ function deferred(): Deferred { } class FakeDaemon implements LiveApi { - readonly gets: { readonly path: ApiPath; readonly at: number }[] = []; + readonly gets: { readonly path: ApiPath; readonly at: number; readonly tag?: string }[] = []; readonly healthChecks: number[] = []; readonly streams: FakeStream[] = []; - reads: "answer" | "hold" | ((path: ApiPath) => Promise>) = "answer"; + reads: + | "answer" + | "hold" + | ((path: ApiPath) => Promise | ConditionalResponse>) = "answer"; holdStreams = false; refuseStreams: (() => Promise) | undefined; health: "up" | "down" | "unreachable" = "up"; date: string | null = null; + // why: the tag every route answers with; a read that sends it back gets a 304. + etag: string | undefined; readonly #held: Deferred>[] = []; readonly #heldStreams: (() => void)[] = []; readonly #encoder = new TextEncoder(); - get(path: ApiPath): Promise> { - this.gets.push({ at: Date.now(), path }); + get(path: ApiPath): Promise>; + get(path: ApiPath, options: ReadOptions): Promise>; + get( + path: ApiPath, + options?: ReadOptions, + ): Promise | ConditionalResponse> { + const tag = options?.tag; + this.gets.push({ at: Date.now(), path, ...(tag === undefined ? {} : { tag }) }); const reads = this.reads; - if (typeof reads === "function") return reads(path) as Promise>; - if (reads === "hold") { - const held = deferred>(); - this.#held.push(held); - return held.promise as Promise>; + if (reads === "answer" && tag !== undefined && tag === this.etag) { + return Promise.resolve({ date: this.date, notModified: true }); } - return Promise.resolve({ body: { read: this.gets.length }, date: this.date } as never); + if (typeof reads === "function") return reads(path); + if (reads === "hold") return this.#hold() as Promise>; + return Promise.resolve({ + body: { read: this.gets.length }, + date: this.date, + tag: this.etag ?? null, + } as never); + } + + #hold(): Promise> { + const held = deferred>(); + this.#held.push(held); + return held.promise; } openHeldStreams(): void { diff --git a/ui/src/model/renewal.ts b/ui/src/model/renewal.ts new file mode 100644 index 00000000..61d4a0f9 --- /dev/null +++ b/ui/src/model/renewal.ts @@ -0,0 +1,100 @@ +import type { ConsoleEvent } from "./events"; +import type { LeaseList, LeaseRecord } from "./leases"; +import type { WorkerList } from "./workers"; + +export interface Renewal { + readonly leaseId: string; + readonly newDeadline: number; + readonly timestamp: number; + // why: set by a gateway, which relays the worker's own lease id. + readonly workerId?: string; +} + +// why: the payload is wire input, so a renewal missing either field is dropped rather than patched +// from a guess. +export function readRenewal(event: ConsoleEvent): Renewal | undefined { + if (event.event !== "lease.renewed") return undefined; + const payload = event.payload; + if (typeof payload !== "object" || payload === null) return undefined; + const { leaseId, newDeadline, workerId } = payload as Record; + if (typeof leaseId !== "string" || typeof newDeadline !== "number") return undefined; + return { + leaseId, + newDeadline, + timestamp: event.timestamp, + ...(typeof workerId === "string" ? { workerId } : {}), + }; +} + +// why: a gateway's lease list names a lease by `.` unless the requester chose +// the id, so its record is found by the worker and the worker's own id; a worker's records carry the id the event names. +function isRenewed(lease: LeaseRecord, renewal: Renewal): boolean { + return renewal.workerId === undefined + ? lease.id === renewal.leaseId + : lease.workerId === renewal.workerId && lease.workerLeaseId === renewal.leaseId; +} + +function renewLease( + lease: LeaseRecord, + renewal: Renewal, + matches: (lease: LeaseRecord) => boolean, +): LeaseRecord { + if (!matches(lease) || renewal.newDeadline <= lease.ttlDeadline) return lease; + return { + ...lease, + lastRenewedAt: renewal.timestamp, + ttlDeadline: renewal.newDeadline, + ttlMs: renewal.newDeadline - renewal.timestamp, + }; +} + +// why: the same list when no lease changed, so an observer of it is not told of an update. +function renewLeases(list: LeaseList, renewal: Renewal): LeaseList { + const leases = list.leases.map((lease) => + renewLease(lease, renewal, (candidate) => isRenewed(candidate, renewal)), + ); + return leases.some((lease, index) => lease !== list.leases[index]) ? { ...list, leases } : list; +} + +// why: a worker's own page lists leases under their worker, by the worker's own lease id. +function renewWorkers(list: WorkerList, renewal: Renewal): WorkerList { + const workers = list.workers.map((worker) => { + if (worker.leases === undefined) return worker; + if (renewal.workerId !== undefined && worker.id !== renewal.workerId) return worker; + const leases = worker.leases.map((lease) => + renewLease(lease, renewal, (candidate) => candidate.id === renewal.leaseId), + ); + return leases.some((lease, index) => lease !== worker.leases?.[index]) + ? { ...worker, leases } + : worker; + }); + return workers.some((worker, index) => worker !== list.workers[index]) + ? { ...list, workers } + : list; +} + +// why: the routes whose answers carry leases, each patched by its own match rule. +export function renewAnswer(route: string, data: T, renewal: Renewal): T { + if (route === "leases") return renewLeases(data as LeaseList, renewal) as T; + if (route === "workers") return renewWorkers(data as WorkerList, renewal) as T; + return data; +} + +// why: the Inbox shows a lease's time left only on its `ending` item, the only item that embeds a lease. +export function endsInInbox(data: unknown, renewal: Renewal): boolean { + const items = (data as { readonly items?: readonly unknown[] } | undefined)?.items; + return (items ?? []).some((item) => { + const { lease } = item as { lease?: LeaseRecord }; + return lease !== undefined && isRenewed(lease, renewal); + }); +} + +// why: a gateway routes a lease under `.`, or under the bare id its requester +// chose; the lease list says which, and an unlisted lease is taken as prefixed. +export function leaseKeyOf(renewal: Renewal, leases: unknown): readonly string[] { + if (renewal.workerId === undefined) return ["lease", renewal.leaseId]; + const listed = (leases as LeaseList | undefined)?.leases?.find((lease) => + isRenewed(lease, renewal), + ); + return ["lease", listed?.id ?? `${renewal.workerId}.${renewal.leaseId}`]; +} diff --git a/ui/src/state/queries.test.ts b/ui/src/state/queries.test.ts index fda91610..239b6010 100644 --- a/ui/src/state/queries.test.ts +++ b/ui/src/state/queries.test.ts @@ -471,3 +471,251 @@ describe("a read's answers and failures", () => { expect(connection.serverNow() - Date.now()).toBe(-30_000); }); }); + +describe("a route's ETag", () => { + it("a re-read of an unchanged route sends the first tag: a 304 keeps the same data object with no update, a 200 brings new data and its tag, a 304 after the data was dropped is read again without a tag", async () => { + const { daemon, queries } = start(); + daemon.etag = '"v1"'; + const observer = new QueryObserver(queries.client, { + ...queries.options("/v1/workers"), + notifyOnChangeProps: ["data"], + }); + const updates = vi.fn(); + disposers.push(observer.subscribe(updates)); + await settle(); + const first = observer.getCurrentResult().data; + expect(first).toBeDefined(); + expect(daemon.gets.map((get) => get.tag)).toEqual([undefined]); + updates.mockClear(); + + await vi.advanceTimersByTimeAsync(15_000); + + expect(daemon.gets.map((get) => get.tag)).toEqual([undefined, '"v1"']); + expect(observer.getCurrentResult().data).toBe(first); + expect(updates).not.toHaveBeenCalled(); + + daemon.etag = '"v2"'; + await vi.advanceTimersByTimeAsync(15_000); + const second = observer.getCurrentResult().data; + expect(daemon.gets.map((get) => get.tag)).toEqual([undefined, '"v1"', '"v1"']); + expect(second).not.toBe(first); + expect(second).toEqual({ read: 3 }); + + await vi.advanceTimersByTimeAsync(15_000); + expect(daemon.gets.map((get) => get.tag)).toEqual([undefined, '"v1"', '"v1"', '"v2"']); + expect(observer.getCurrentResult().data).toBe(second); + + let release: () => void = () => {}; + daemon.reads = () => + new Promise((resolve) => { + release = () => { + daemon.reads = "answer"; + resolve({ date: null, notModified: true }); + }; + }); + await vi.advanceTimersByTimeAsync(15_000); + queries.client.removeQueries({ queryKey: routeKey("/v1/workers") }); + release(); + await settle(); + + expect(daemon.gets.map((get) => get.tag)).toEqual([ + undefined, + '"v1"', + '"v1"', + '"v2"', + '"v2"', + undefined, + ]); + }); +}); + +describe("a lease renewal", () => { + const NOW = Date.parse("2026-10-02T12:00:00Z"); + const lease = (id: string, over: Record = {}) => ({ + deviceId: `dev_${id}`, + grantedAt: NOW, + id, + idChosenByRequester: false, + lastRenewedAt: NOW, + ownerId: "owner", + requesterId: "agent", + ttlDeadline: NOW + 60_000, + ttlMs: 60_000, + ...over, + }); + const renewed = (payload: Record, seq = 2) => + `event: lease.renewed\ndata: ${JSON.stringify({ + event: "lease.renewed", + id: `evt_${String(seq)}`, + payload, + seq, + timestamp: Date.now(), + })}\n\n`; + + it("moves the lease's deadline in place with no read: a worker event matches by id, a gateway event by worker and the worker's lease id in both lists, lastRenewedAt and ttlMs follow, a later answer with the old deadline keeps the new one for 60 seconds and one with a later deadline wins, no match or a malformed payload changes nothing", async () => { + const { daemon, reads, state, watch } = start(); + daemon.reads = () => + Promise.resolve({ body: { leases: [lease("l_1"), lease("l_2")] }, date: null }); + watch("/v1/leases"); + await settle(); + const before = state("/v1/leases").data as { leases: object[] }; + + await vi.advanceTimersByTimeAsync(5_000); + const newDeadline = Date.now() + 120_000; + daemon.openStream()?.send(renewed({ leaseId: "l_1", newDeadline })); + await settle(); + + const after = state("/v1/leases").data as { leases: object[] }; + expect(after.leases[0]).toEqual( + lease("l_1", { lastRenewedAt: Date.now(), ttlDeadline: newDeadline, ttlMs: 120_000 }), + ); + expect(after.leases[1]).toBe(before.leases[1]); + expect(reads("/v1/leases")).toBe(1); + + daemon.openStream()?.send(renewed({ leaseId: "l_9", newDeadline: newDeadline + 1_000 }, 3)); + await settle(); + expect(state("/v1/leases").data).toBe(after); + expect(reads("/v1/leases")).toBe(1); + + daemon.openStream()?.send(renewed({ leaseId: "l_2", newDeadline: "soon" }, 5)); + await settle(); + expect(state("/v1/leases").data).toBe(after); + + // why: a renewal, matched or not, is no read, so it leaves the poll's schedule alone. + const readsBefore = reads("/v1/leases"); + for (let tick = 0; tick < 5; tick += 1) { + await vi.advanceTimersByTimeAsync(5_000); + daemon + .openStream() + ?.send( + renewed( + { leaseId: tick % 2 === 0 ? "l_9" : "l_1", newDeadline: Date.now() + 120_000 }, + 10 + tick, + ), + ); + } + await vi.advanceTimersByTimeAsync(1_000); + expect( + reads("/v1/leases") - readsBefore, + "polls at 10, 20 and 30 s through renewals every 5 s, listed and unlisted", + ).toBe(3); + + const gateway = start(); + const old = { + leases: [ + lease("w1.l_1", { workerId: "w1", workerLeaseId: "l_1" }), + lease("w1.l_2", { workerId: "w1", workerLeaseId: "l_2" }), + lease("w2.l_1", { workerId: "w2", workerLeaseId: "l_1" }), + ], + }; + const fleet = { + workers: [ + { id: "w1", leases: [lease("l_1"), lease("l_2")] }, + { id: "w2", leases: [lease("l_1")] }, + ], + }; + gateway.daemon.reads = (path) => + Promise.resolve({ body: path === "/v1/leases" ? old : fleet, date: null }); + gateway.watch("/v1/leases"); + gateway.watch("/v1/workers"); + await settle(); + const deadline = Date.now() + 120_000; + gateway.daemon + .openStream() + ?.send(renewed({ leaseId: "l_1", newDeadline: deadline, workerId: "w1" }, 4)); + await settle(); + + const moved = { lastRenewedAt: Date.now(), ttlDeadline: deadline, ttlMs: 120_000 }; + expect(gateway.state("/v1/leases").data).toEqual({ + leases: [ + lease("w1.l_1", { ...moved, workerId: "w1", workerLeaseId: "l_1" }), + old.leases[1], + old.leases[2], + ], + }); + expect(gateway.state("/v1/workers").data).toEqual({ + workers: [{ id: "w1", leases: [lease("l_1", moved), lease("l_2")] }, fleet.workers[1]], + }); + expect(gateway.reads("/v1/leases") + gateway.reads("/v1/workers")).toBe(2); + + await vi.advanceTimersByTimeAsync(10_000); + expect(gateway.reads("/v1/leases")).toBe(2); + const kept = gateway.state("/v1/leases").data as { leases: { ttlDeadline: number }[] }; + expect(kept.leases[0]?.ttlDeadline).toBe(deadline); + expect(kept.leases[1]?.ttlDeadline).toBe(NOW + 60_000); + + const later = deadline + 5_000; + gateway.daemon.reads = (path) => + Promise.resolve({ + body: + path === "/v1/leases" + ? { leases: [{ ...old.leases[0], ttlDeadline: later }, ...old.leases.slice(1)] } + : fleet, + date: null, + }); + await vi.advanceTimersByTimeAsync(10_000); + const newer = gateway.state("/v1/leases").data as { leases: { ttlDeadline: number }[] }; + expect(newer.leases[0]?.ttlDeadline).toBe(later); + + gateway.daemon.reads = (path) => + Promise.resolve({ body: path === "/v1/leases" ? old : fleet, date: null }); + await vi.advanceTimersByTimeAsync(30_000); + const still = gateway.state("/v1/leases").data as { leases: { ttlDeadline: number }[] }; + expect(still.leases[0]?.ttlDeadline).toBe(deadline); + + await vi.advanceTimersByTimeAsync(30_000); + const dropped = gateway.state("/v1/leases").data as { leases: { ttlDeadline: number }[] }; + expect(dropped.leases[0]?.ttlDeadline).toBe(NOW + 60_000); + }); + it("marks only what a renewal leaves stale: 20 renewals of ending leases in 1 s give one inbox read, a lease observed gives one read, otherwise no request", async () => { + const ending = { condition: "ending", group: "Heads-up", key: "lease/l_1/ending" }; + const inbox = { items: [{ ...ending, lease: lease("l_1") }] }; + const cases = [ + ["an ending lease in the inbox", "/v1/inbox", "l_1", 20, ["/v1/inbox"]], + ["a lease whose page is open", "/v1/leases/l_1", "l_1", 20, ["/v1/leases/l_1"]], + ["a lease nothing shows", "/v1/inbox", "l_2", 20, []], + ["another lease's page", "/v1/leases/l_1", "l_2", 20, []], + ["a gateway lease's page", "/v1/leases/w1.l_1", "l_1", 20, ["/v1/leases/w1.l_1"]], + [ + "a gateway lease its requester named, whose page has no worker prefix", + "/v1/leases/my-lease", + "my-lease", + 20, + ["/v1/leases/my-lease"], + ], + ] as const; + for (const [label, path, leaseId, count, expected] of cases) { + const { daemon, watch } = start(); + const listed = { leases: [lease("my-lease", { workerId: "w1", workerLeaseId: "my-lease" })] }; + daemon.reads = (read) => + Promise.resolve({ + body: read === "/v1/inbox" ? inbox : read === "/v1/leases" ? listed : {}, + date: null, + }); + watch("/v1/leases"); + watch(path); + await settle(); + const before = daemon.gets.length; + + for (let renewal = 0; renewal < count; renewal += 1) { + daemon.openStream()?.send( + renewed( + { + leaseId, + newDeadline: Date.now() + 120_000 + renewal, + ...(path.includes("w1.") || path.includes("my-lease") ? { workerId: "w1" } : {}), + }, + renewal + 2, + ), + ); + await vi.advanceTimersByTimeAsync(50); + } + await vi.advanceTimersByTimeAsync(300); + + expect( + daemon.gets.slice(before).map((get) => get.path), + label, + ).toEqual(expected); + } + }); +}); diff --git a/ui/src/state/queries.ts b/ui/src/state/queries.ts index d907269c..aaaf5249 100644 --- a/ui/src/state/queries.ts +++ b/ui/src/state/queries.ts @@ -1,5 +1,6 @@ import { focusManager, + hashKey, onlineManager, type Query, QueryClient, @@ -15,6 +16,8 @@ import { type StreamEvent, type Visibility, } from "../data/live/connection"; +import { readEvent } from "../model/events"; +import { endsInInbox, leaseKeyOf, readRenewal, renewAnswer, type Renewal } from "../model/renewal"; import { type StaleKey, staleKeys } from "../model/stale-keys"; // why: data no page uses stays a minute, so going back to a page shows it at once. @@ -26,6 +29,8 @@ const FRESH_MS = 2_000; const QUIET_MS = 300; const CAP_MS = 1_000; const AGAIN_MS = 1_500; +const RENEWAL_KEPT_MS = 60_000; +const RENEWED_ROUTES: readonly string[] = ["leases", "workers"]; interface Times { readonly pollMs?: number; @@ -134,12 +139,28 @@ export function createQueries({ clock, connection, visibility }: QueriesOptions) schedulePoll(event.query); } else if ( event.type === "updated" && - (event.action.type === "success" || event.action.type === "error") + // why: a patch written by setQueryData (manual) is no read, so it leaves the poll's timer running. + (event.action.type === "error" || + (event.action.type === "success" && event.action.manual !== true)) ) { schedulePoll(event.query); } }); + // why: the ETag of each key's last answer, sent back so an unchanged answer costs no download. + const tags = new Map(); + + // why: a gateway relays a worker's event before it refreshes its view of that worker, so a read + // just after a renewal can still carry the old deadline. + const kept: { readonly renewal: Renewal; readonly until: number }[] = []; + const renewedAnswer = (route: string, data: T): T => { + const now = clock.now(); + for (let index = kept.length - 1; index >= 0; index -= 1) { + if ((kept[index]?.until ?? 0) <= now) kept.splice(index, 1); + } + return kept.reduce((answer, { renewal }) => renewAnswer(route, answer, renewal), data); + }; + const marked = new Map(); const again = new Map(); let quietTimer: unknown; @@ -169,13 +190,35 @@ export function createQueries({ clock, connection, visibility }: QueriesOptions) readMarked(later); }, AGAIN_MS); }; + // why: a renewal needs a read only where its lease's deadline is shown from another answer: the + // inbox's ending item and an open lease page. + const renewalKeys = (renewal: Renewal): StaleKey[] => [ + leaseKeyOf(renewal, client.getQueryData(["leases"])), + ...(endsInInbox(client.getQueryData(["inbox"]), renewal) ? [["inbox"]] : []), + ]; + const patchRenewal = (renewal: Renewal): void => { + kept.push({ renewal, until: clock.now() + RENEWAL_KEPT_MS }); + for (const route of RENEWED_ROUTES) { + // why: a renewal that moves no lease leaves the cache, and its observers, alone. + const old = client.getQueryData([route]); + if (old === undefined) continue; + const patched = renewAnswer(route, old, renewal); + if (patched !== old) client.setQueryData([route], patched); + } + }; const stopEvents = connection.onEvent((event: StreamEvent) => { const { data } = event; + const parsed = readEvent(data); + const renewal = parsed === undefined ? undefined : readRenewal(parsed); + if (renewal !== undefined) patchRenewal(renewal); const payload = typeof data === "object" && data !== null ? (data as { payload?: unknown }).payload : undefined; - const keys = staleKeys(event.event, payload); + const keys = [ + ...staleKeys(event.event, payload), + ...(renewal === undefined ? [] : renewalKeys(renewal)), + ]; if (keys.length === 0) return; for (const queryKey of keys) { marked.set(queryKey.join("/"), queryKey); @@ -200,10 +243,31 @@ export function createQueries({ clock, connection, visibility }: QueriesOptions) client.unmount(); client.clear(); }, - options: (path: ApiPath): QueryObserverOptions => ({ - queryFn: () => connection.read(path), - queryKey: routeKey(path), - staleTime: timesOf(routeKey(path)).freshMs, - }), + options: (path: ApiPath): QueryObserverOptions => { + const key = routeKey(path); + const hash = hashKey(key); + return { + queryFn: async () => { + let tag = tags.get(hash); + for (;;) { + const answer = await connection.read(path, { tag }); + if (answer.notModified !== true) { + if (answer.tag === null) tags.delete(hash); + else tags.set(hash, answer.tag); + return RENEWED_ROUTES.includes(key[0] ?? "") + ? renewedAnswer(key[0] ?? "", answer.body) + : answer.body; + } + const kept = client.getQueryData(key); + if (kept !== undefined) return kept; + // why: the data this tag stood for was dropped while the read was out, so the tag + // describes nothing and the whole answer is read again. + tag = undefined; + } + }, + queryKey: key, + staleTime: timesOf(key).freshMs, + }; + }, }; }