diff --git a/CHANGELOG.md b/CHANGELOG.md index 58f5136..6f55dca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,7 @@ the project uses [semantic versioning](https://semver.org/) once it reaches 1.0; ### Changed +- **Live updates no longer poll every 5 seconds.** The console refreshes from the live-update stream and on focus, and polls (with jitter) only while the stream is down. Web processes share one freshness read per organization (at most about one per 400 ms), and stream lifetimes are randomized (40 to 55 seconds) so browsers do not reconnect together. `task-views`, `decisions` and `notifications` reads send `ETag` and answer `304` when unchanged; responses stay `no-store`. - **Console navigation is reorganised by job.** The sidebar groups daily work (**Home**, **Inbox**, **Tasks**, **Chat**), **Agents** (**Agents**, **Find tasks**, **Orchestration**) and **You** (**Notifications**, **Settings**, **Administration** for administrators). - **Administration is its own area** (`/admin`) with **Overview**, **People**, **Teams**, **Agent access** and **Policies & alerts**, plus a link to the audit trail. It replaces the organization cards that sat under **Settings**, which now holds only your browser-stored request defaults and extensions. **Teams** gains **Disable team**. No API changed. - **People management in the console.** With OIDC sign-in, **Administration → People** can add a person, change a role, deactivate and reactivate, all audited (`member.*`) and with the same guards as the API: you cannot change or deactivate yourself. In local development mode the page shows the built-in administrator and explains that people need OIDC. `POST /api/admin/security` gains `provisionMember`, `changeRole`, `deactivateMember` and `reactivateMember`; `GET` adds `identity` and `inactiveMemberships`. diff --git a/docs/concepts/durable-tasks.md b/docs/concepts/durable-tasks.md index f7f0b9d..4703da2 100644 --- a/docs/concepts/durable-tasks.md +++ b/docs/concepts/durable-tasks.md @@ -103,7 +103,7 @@ Raw events live in an append-only ledger. What the screens read is a projection: The projection is derived by a deterministic reducer, so it can be rebuilt from retained events. Rebuild prepares the new generation outside the task lock, verifies archived binary digests, then swaps it in only if the task has not changed in the meantime. A failed rebuild keeps the old generation readable. Rebuild never sends anything to an agent. It cannot restore history that retention has deleted. See [Upgrading and backups](../guides/upgrading-and-backups.md) for the command. -To tell browsers something changed, every accepted projection write queues a `task.freshness` outbox row. A worker publishes it. `GET /api/tasks/events` streams three content-free events: `ready` (sent on every connection), `freshness` and `resync` (after about 15 s of quiet). Connections last about 55 s and then the browser reconnects. Signals carry no task data. On any signal, on reconnect, on focus and on a 5-second fallback poll, the browser re-queries. A lost signal costs at most a few seconds of staleness. +To tell browsers something changed, every accepted projection write queues a `task.freshness` outbox row. A worker publishes it. `GET /api/tasks/events` streams three content-free events: `ready` (sent on every connection), `freshness` and `resync` (after about 15 s of quiet). Connections last 40 to 55 s (randomized) and then the browser reconnects. Signals carry no task data. On any signal, on reconnect and on focus, the browser re-queries. It polls (about every 5–7.5 seconds, with jitter) only while the stream is down: when the browser has no `EventSource`, or no frame has arrived for 20 seconds or a reconnect has failed for 5. The stream stays open for 40 to 55 seconds at a time, so an idle tab makes one re-query per signal, about one every 15 seconds, and no timer reads while the stream is healthy. Re-queries send `If-None-Match` from tab memory and get `304` when nothing changed, so an unchanged read costs no response body and no re-render. ## Identity of remote IDs @@ -118,7 +118,7 @@ Known `referenceTaskIds` on user and agent messages create durable local links w | Command stuck `pending` | No worker running, or the agent is disabled | Check the worker is running. See [Running and workers](../operations/running-and-workers.md) | | Command `uncertain` | Failure after dispatch, or lease expiry | Open the task, check its state, and send again only if the agent did nothing | | Task never leaves `working` with no browser open | Agent lacks streaming and push is off | Reconciliation will catch up within about 15 s. Check agent health | -| UI is a few seconds stale | Freshness signal lost | The 5 s fallback poll covers it | +| UI is a few seconds stale | Freshness signal lost | The next `resync` (within 15 s) covers it, or the fallback poll if the stream is down | | Rebuild refuses | Ledger empty or an archive is missing or corrupt | Restore from backup. It fails closed on purpose | More in [Troubleshooting](../operations/troubleshooting.md). diff --git a/docs/guides/production-deployment.md b/docs/guides/production-deployment.md index 17a4d16..492e6fe 100644 --- a/docs/guides/production-deployment.md +++ b/docs/guides/production-deployment.md @@ -126,7 +126,8 @@ Run both processes under a supervisor that restarts them. A crash loses nothing Terminate TLS at a proxy and forward to the web process. Requirements specific to Taskbay: - Forward everything under `/` to the web process. Agents call `POST /api/webhooks/a2a/` without a browser session; do not put your own authentication in front of that path if you enable push. -- Do not buffer or time out `text/event-stream` responses. The live-update stream (`GET /api/tasks/events`) ends by itself after 55 seconds and the browser reconnects; it already sets `X-Accel-Buffering: no`. +- Do not buffer or time out `text/event-stream` responses. The live-update stream (`GET /api/tasks/events`) ends by itself after 40 to 55 seconds and the browser reconnects; it already sets `X-Accel-Buffering: no`. +- Several web replicas (for example on Kubernetes) need no sticky sessions: a stream is stateless, lives 40 to 55 seconds and reconnects to whichever replica the balancer picks. Freshness travels through the database, so a change committed by any replica or worker wakes browsers connected to every other replica. Each replica reads the freshness token at most about 2.5 times a second per organization that has an open stream, however many browsers are connected. Set the ingress or load-balancer idle timeout above 60 seconds, and let a draining replica close its streams; browsers reconnect on their own. - The request body limit in the application is 16 MB. - The public origin in `A2A_AUTH_ORIGIN` must be exactly what browsers use. The login redirect and the `Origin` check for mutations come from it, not from `Host` headers. diff --git a/docs/guides/use-the-console.md b/docs/guides/use-the-console.md index b43a048..00286ab 100644 --- a/docs/guides/use-the-console.md +++ b/docs/guides/use-the-console.md @@ -370,7 +370,7 @@ Administrators also see a link to **Administration** at the bottom of the page. | A form shows as raw data | The agent does not advertise the extension, the form is outside the supported subset, or it was already answered | See [Rich content](../concepts/rich-content.md#when-it-fails) | | "Conversation not found" / "This conversation is unavailable in the current organization." | The link is from another organization, or you lost read access | Open the chat from **Recent conversations**, or ask for a grant | | "Task not found" | The task does not exist or you cannot read it. The server answers 404 rather than 403. | Ask an administrator for a grant | -| Banner ending "Retrying server sync…" | The browser could not load tasks from the server | Check the server is running. The console retries every 5 seconds. | +| Banner ending "Retrying server sync…" | The browser could not load tasks from the server | Check the server is running. The console retries on the next live-update signal (about every 15 seconds), or about every 5–7 seconds while the live-update stream is down. | | "Administrators only" on **Audit trail** | You are not an administrator | Open a task and use **View the full trail** | | **Administration** missing from the sidebar, or "Administrators only" | You are not an administrator | Ask an administrator | | "Too many requests. Try again shortly." | Per-person rate limit | Wait a minute | diff --git a/docs/operations/running-and-workers.md b/docs/operations/running-and-workers.md index 33f319c..6c4b6f2 100644 --- a/docs/operations/running-and-workers.md +++ b/docs/operations/running-and-workers.md @@ -75,9 +75,9 @@ All loops are idempotent and retry after a database error, logging a single line | Approvals and escalation | Expires overdue approval requests, marks requests superseded when their task finished, refreshes in-flight approved deliveries, and escalates overdue owned tasks. | Every `A2A_DECISION_SWEEP_MS` (default 15000; values below 100, or not a number, fall back to 15000). | | Notifications | Turns events into inbox rows, then sends each to the signed webhook if one is configured. Webhook delivery retries up to 8 times with backoff from 5 s to 10 min. | Polls every 250 ms. | -The browser side: `GET /api/tasks/events` is a server-sent-events stream that sends `ready`, then `freshness` when something committed and `resync` about every 15 seconds. It closes itself after 55 seconds -and the browser reconnects. It carries no task content; the browser re-reads state from the API. With PostgreSQL, web replicas poll a freshness token every 500 ms, so a change made by a worker on another host -shows up without a message broker. +The browser side: `GET /api/tasks/events` is a server-sent-events stream that sends `ready`, then `freshness` when something committed and `resync` about every 15 seconds. It closes itself after 40 to 55 seconds (randomized) +and the browser reconnects. It carries no task content; the browser re-reads state from the API. With PostgreSQL, each web process polls the freshness token every 500 ms while a stream is open, so a change made by a worker on another host +shows up without a message broker. Open streams in a process share those reads: the database sees at most one token read per organization per 400 ms per web process, however many browsers are connected. ## Scaling and leases diff --git a/docs/reference/http-api.md b/docs/reference/http-api.md index 5b15279..ab7e577 100644 --- a/docs/reference/http-api.md +++ b/docs/reference/http-api.md @@ -52,7 +52,7 @@ copied from a browser. That is an awkward workflow, and the reason a token API i | Topic | Behavior | | --- | --- | | Format | JSON in, JSON out, `Content-Type: application/json`. Request bodies larger than 16 MiB are rejected with `400`. A missing body is `400 Request body is required.`; invalid JSON is `400 Invalid request JSON.` | -| Caching | Responses carry `Cache-Control: no-store`. | +| Caching | Responses carry `Cache-Control: no-store`. `GET /api/task-views`, `GET /api/decisions` and `GET /api/notifications` also carry a weak `ETag` of the body and answer `304` (no body, still `no-store`) to a matching `If-None-Match`. The console sends it from tab memory, so the browser stores nothing. A `304` saves the response body and the client re-render; the server still runs the query. | | IDs | Local IDs are UUIDs generated by Agent Taskbay. A task has a local ID (`localId`, used in URLs) and a remote ID (`taskId` from the agent); they are different. Remote IDs are only unique within `(agent, tenant)`. | | Timestamps | ISO 8601 UTC strings. | | Idempotency | Mutations that start durable work take an `Idempotency-Key` header (1 to 255 characters). Reusing a key with identical content returns the original result; with different content it returns `409`. | @@ -458,7 +458,7 @@ A message that reached Taskbay only because the agent's history was loaded has ` `read`. A long-lived SSE signal that "something changed for this organization". Frames carry no task data (`data: {}`); clients re-query. Event names: `ready` (first frame), `freshness` (a change was committed), `resync` (sent after 15 seconds of silence so clients refresh anyway). Each frame is preceded by -`retry: 1000`. The server closes the stream after 55 seconds; reconnect. +`retry: 1000`. The server closes the stream after 40 to 55 seconds (randomized so connections do not expire together); reconnect. ## Task workflow diff --git a/scripts/verify-task-http.mjs b/scripts/verify-task-http.mjs index 90b0c61..adf44c2 100644 --- a/scripts/verify-task-http.mjs +++ b/scripts/verify-task-http.mjs @@ -394,7 +394,13 @@ try { assert.deepEqual((await json("/api/task-views?organizationId=foreign&tenant=foreign")).tasks, browserOne); assert.equal((await fetch(base + "/api/task-views?limit=0")).status, 400); assert.equal((await fetch(base + "/api/task-views?after=" + remoteId)).status, 400); - assert.equal((await fetch(base + "/api/task-views")).headers.get("cache-control"), "no-store"); + const viewsResponse = await fetch(base + "/api/task-views"); + assert.equal(viewsResponse.headers.get("cache-control"), "no-store"); + const viewsTag = viewsResponse.headers.get("etag"); + assert.ok(viewsTag); + const unchangedViews = await fetch(base + "/api/task-views", { headers: { "If-None-Match": viewsTag } }); + assert.equal(unchangedViews.status, 304); + assert.equal(unchangedViews.headers.get("cache-control"), "no-store"); await json(`/api/agents/${one.id}/tasks/${remoteId}/cancel?tenant=tenant-b`, { method: "POST" }); assert.equal((await json(`/api/tasks/${ids[2]}`)).task.state, "TASK_STATE_CANCELED"); assert.equal((await json("/api/task-views")).tasks.find((task) => task.localId === ids[2]).state, "TASK_STATE_CANCELED"); diff --git a/src/app/api/decisions/route.ts b/src/app/api/decisions/route.ts index ea2ecc6..bc2f810 100644 --- a/src/app/api/decisions/route.ts +++ b/src/app/api/decisions/route.ts @@ -1,4 +1,5 @@ import { authenticatedRoute } from "@/server/runtime/identity"; +import { conditionalJson } from "@/lib/conditional-json"; import { apiError } from "@/lib/api-response"; import { readJsonRequest } from "@/lib/request-guard"; import { createDecisionService, listDecisions, openDecisionSchema, parseBody, toAction, requestView, requirePrincipal } from "@/server/runtime/decisions"; @@ -14,7 +15,7 @@ async function handleGET(request: Request) { if (status && !(STATUSES as readonly string[]).includes(status)) return Response.json({ error: { message: "Unknown status." } }, { status: 400 }); const requests = await listDecisions({ status: (status ?? undefined) as (typeof STATUSES)[number] | undefined, taskId: params.get("taskId") ?? undefined, mine: params.get("assigned") === "me" }); - return Response.json({ decisions: requests }, { headers: { "Cache-Control": "no-store" } }); + return conditionalJson(request, { decisions: requests }); } catch (error) { return apiError(error, 400); } } diff --git a/src/app/api/notifications/route.ts b/src/app/api/notifications/route.ts index e934a71..7a36e53 100644 --- a/src/app/api/notifications/route.ts +++ b/src/app/api/notifications/route.ts @@ -1,4 +1,5 @@ import { authenticatedRoute } from "@/server/runtime/identity"; +import { conditionalJson } from "@/lib/conditional-json"; import { apiError } from "@/lib/api-response"; import { inboxQuerySchema, listInbox } from "@/server/runtime/notifications"; export const runtime = "nodejs"; @@ -8,7 +9,7 @@ async function handleGET(request: Request) { try { const parsed = inboxQuerySchema.safeParse(Object.fromEntries(new URL(request.url).searchParams)); if (!parsed.success) return Response.json({ error: { message: parsed.error.issues.map((issue) => issue.message).join("; ") } }, { status: 400 }); - return Response.json(await listInbox({ unreadOnly: parsed.data.unread === "true", limit: parsed.data.limit, cursor: parsed.data.cursor }), { headers: { "Cache-Control": "no-store" } }); + return conditionalJson(request, await listInbox({ unreadOnly: parsed.data.unread === "true", limit: parsed.data.limit, cursor: parsed.data.cursor })); } catch (error) { return apiError(error, 400); } } diff --git a/src/app/api/task-views/route.ts b/src/app/api/task-views/route.ts index 5c00e47..88472bd 100644 --- a/src/app/api/task-views/route.ts +++ b/src/app/api/task-views/route.ts @@ -1,4 +1,5 @@ import { authenticatedRoute } from "@/server/runtime/identity"; +import { conditionalJson } from "@/lib/conditional-json"; import { apiError } from "@/lib/api-response"; import { withTaskQueries } from "@/server/runtime/task-persistence"; @@ -16,7 +17,7 @@ async function handleGET(request: Request) { return Response.json({ error: { message: "limit must be 1–100 and after must be a local UUID." } }, { status: 400 }); } const page = await withTaskQueries((queries, organizationId) => queries.contentPage(organizationId, limit, after)); - return Response.json(page, { headers: { "Cache-Control": "no-store" } }); + return conditionalJson(request, page); } catch (error) { return apiError(error, 500); } } diff --git a/src/app/api/tasks/events/route.ts b/src/app/api/tasks/events/route.ts index d192c09..29b1d88 100644 --- a/src/app/api/tasks/events/route.ts +++ b/src/app/api/tasks/events/route.ts @@ -1,7 +1,7 @@ import { authenticatedRoute } from "@/server/runtime/identity"; import { apiError } from "@/lib/api-response"; import { withTaskQueries } from "@/server/runtime/task-persistence"; -import { runtimeFreshness } from "@/server/runtime/freshness"; +import { liveFreshnessReader } from "@/server/runtime/freshness"; import { freshnessStream } from "@/server/adapters/live/freshness-stream"; export const runtime = "nodejs"; @@ -11,7 +11,7 @@ async function handleGET(request: Request) { try { // Like task reads, scope comes from the server, never a query/header selector. const organizationId = await withTaskQueries(async (_queries, organization) => organization); - const reader = runtimeFreshness(); + const reader = liveFreshnessReader(); await reader.readToken(organizationId); // Fail before returning streaming headers. return new Response(freshnessStream(reader, organizationId, request.signal), { headers: { "Content-Type": "text/event-stream", "Cache-Control": "no-store, no-transform", diff --git a/src/components/chat/durable-task-sync.tsx b/src/components/chat/durable-task-sync.tsx index ac013b3..6b5eb0c 100644 --- a/src/components/chat/durable-task-sync.tsx +++ b/src/components/chat/durable-task-sync.tsx @@ -10,7 +10,6 @@ import { useTaskStore } from "@/store/task-store"; export function DurableTaskSync() { useEffect(() => { const controller = new AbortController(); - let timer: ReturnType; const load = queuedRead(async () => { const revision = useTaskStore.getState().revision; try { @@ -20,15 +19,11 @@ export function DurableTaskSync() { if (!controller.signal.aborted) useTaskStore.getState().fail(error instanceof Error ? error.message : "Could not load server tasks."); } }, controller.signal); - const poll = async () => { - await load(); - if (!controller.signal.aborted) timer = setTimeout(poll, 5000); - }; - void poll(); + void load(); const onFocus = () => { void load(); }; const unsubscribe = subscribeTaskFreshness(onFocus); window.addEventListener("focus", onFocus); - return () => { controller.abort(); clearTimeout(timer); unsubscribe(); window.removeEventListener("focus", onFocus); }; + return () => { controller.abort(); unsubscribe(); window.removeEventListener("focus", onFocus); }; }, []); return null; } diff --git a/src/lib/conditional-fetch.ts b/src/lib/conditional-fetch.ts new file mode 100644 index 0000000..bc9ba36 --- /dev/null +++ b/src/lib/conditional-fetch.ts @@ -0,0 +1,21 @@ +const MAX_ENTRIES = 50; +const held = new Map(); + +/** + * GET JSON with `If-None-Match` from a tab-local memory cache. Responses are `no-store`, so the browser + * keeps nothing; a `304` reuses the body held here. `unchanged` lets callers skip re-rendering. + */ +export async function fetchJson(url: string, signal?: AbortSignal) { + const previous = held.get(url); + const response = await fetch(url, { cache: "no-store", signal, + ...(previous ? { headers: { "If-None-Match": previous.etag } } : {}) }); + if (response.status === 304 && previous) return { ok: true as const, body: previous.body as T, unchanged: true }; + const body = await response.json(); + const etag = response.headers.get("etag"); + if (response.ok && etag) { + held.delete(url); + held.set(url, { etag, body }); + if (held.size > MAX_ENTRIES) held.delete(held.keys().next().value as string); + } else held.delete(url); + return { ok: response.ok, body: body as T, unchanged: false }; +} diff --git a/src/lib/conditional-json.test.ts b/src/lib/conditional-json.test.ts new file mode 100644 index 0000000..e638620 --- /dev/null +++ b/src/lib/conditional-json.test.ts @@ -0,0 +1,36 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { conditionalJson } from "./conditional-json"; +import { fetchJson } from "./conditional-fetch"; + +afterEach(() => vi.unstubAllGlobals()); + +describe("conditional JSON reads", () => { + it("answers 304 only for the body the caller holds, and stays no-store", async () => { + const first = conditionalJson(new Request("http://x/a"), { n: 1 }); + const etag = first.headers.get("etag")!; + expect(first.headers.get("cache-control")).toBe("no-store"); + expect(await first.json()).toEqual({ n: 1 }); + const same = conditionalJson(new Request("http://x/a", { headers: { "If-None-Match": `"other", ${etag}` } }), { n: 1 }); + expect(same.status).toBe(304); expect(same.headers.get("cache-control")).toBe("no-store"); + expect(await same.text()).toBe(""); + const changed = conditionalJson(new Request("http://x/a", { headers: { "If-None-Match": etag } }), { n: 2 }); + expect(changed.status).toBe(200); expect(changed.headers.get("etag")).not.toBe(etag); + }); + + it("revalidates from memory, reuses the body on 304 and drops entries after errors", async () => { + const calls: Array = []; + const responses = [ + () => new Response(JSON.stringify({ n: 1 }), { headers: { ETag: 'W/"v1"' } }), + () => new Response(null, { status: 304, headers: { ETag: 'W/"v1"' } }), + () => new Response(JSON.stringify({ error: { message: "down" } }), { status: 503 }), + () => new Response(JSON.stringify({ n: 1 }), { headers: { ETag: 'W/"v1"' } }), + ]; + vi.stubGlobal("fetch", vi.fn(async (_url: string, init?: RequestInit) => { calls.push(init); return responses.shift()!(); })); + expect(await fetchJson("/api/held")).toEqual({ ok: true, body: { n: 1 }, unchanged: false }); + expect(await fetchJson("/api/held")).toEqual({ ok: true, body: { n: 1 }, unchanged: true }); + expect(calls[1]).toMatchObject({ cache: "no-store", headers: { "If-None-Match": 'W/"v1"' } }); + expect((await fetchJson("/api/held")).ok).toBe(false); + await fetchJson("/api/held"); + expect(calls[3]).not.toHaveProperty("headers"); // Nothing is held after a failure. + }); +}); diff --git a/src/lib/conditional-json.ts b/src/lib/conditional-json.ts new file mode 100644 index 0000000..51f13e5 --- /dev/null +++ b/src/lib/conditional-json.ts @@ -0,0 +1,18 @@ +import { createHash } from "node:crypto"; + +const opaque = (tag: string) => tag.trim().replace(/^W\//, ""); + +/** + * JSON response with a weak ETag of its body, answering `304` when the caller already holds it. + * Responses stay `no-store`; clients revalidate by hand from memory (see conditional-fetch.ts). + */ +export function conditionalJson(request: Request, body: unknown) { + const text = JSON.stringify(body); + const etag = `W/"${createHash("sha256").update(text).digest("base64url").slice(0, 27)}"`; + const headers = { "Cache-Control": "no-store", ETag: etag }; + const held = request.headers.get("if-none-match"); + if (held && (held.trim() === "*" || held.split(",").some((tag) => opaque(tag) === opaque(etag)))) { + return new Response(null, { status: 304, headers }); + } + return new Response(text, { headers: { ...headers, "Content-Type": "application/json" } }); +} diff --git a/src/lib/durable-task-cache.ts b/src/lib/durable-task-cache.ts index ec81eaa..ee8e168 100644 --- a/src/lib/durable-task-cache.ts +++ b/src/lib/durable-task-cache.ts @@ -1,14 +1,18 @@ +import { fetchJson } from "./conditional-fetch"; import type { DurableTaskView } from "../shared/task-types"; +type TaskViewPage = { tasks: DurableTaskView[]; next: string | null; error?: { message?: string } }; + /** Complete a refresh before activating it; a failed page preserves the prior cache. */ export async function readDurableTasks(signal: AbortSignal): Promise { const tasks = new Map(); const cursors = new Set(); let after: string | null = null; do { - const response = await fetch(`/api/task-views?limit=100${after ? `&after=${encodeURIComponent(after)}` : ""}`, { cache: "no-store", signal }); - const page: { tasks: DurableTaskView[]; next: string | null; error?: { message?: string } } = await response.json(); - if (!response.ok) throw new Error(page.error?.message ?? "Could not load server tasks."); + const result: { ok: boolean; body: TaskViewPage } = await fetchJson( + `/api/task-views?limit=100${after ? `&after=${encodeURIComponent(after)}` : ""}`, signal); + const { ok, body: page } = result; + if (!ok) throw new Error(page.error?.message ?? "Could not load server tasks."); for (const task of page.tasks as DurableTaskView[]) tasks.set(task.localId, task); after = page.next; if (after) { diff --git a/src/lib/task-freshness.test.ts b/src/lib/task-freshness.test.ts index a0b9e53..3fb5f4c 100644 --- a/src/lib/task-freshness.test.ts +++ b/src/lib/task-freshness.test.ts @@ -29,9 +29,48 @@ describe("REL-002 browser freshness", () => { expect(two).toHaveBeenCalledTimes(3); }); - it("allows periodic/focus reads when EventSource is unavailable", () => { + it("polls (with jitter) only while EventSource is unavailable", async () => { + vi.useFakeTimers(); vi.stubGlobal("EventSource", undefined); - const leave = subscribeTaskFreshness(vi.fn()); + const listener = vi.fn(); + const leave = subscribeTaskFreshness(listener); + await vi.advanceTimersByTimeAsync(7_500); + expect(listener).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(7_500); + expect(listener).toHaveBeenCalledTimes(2); + leave(); + await vi.advanceTimersByTimeAsync(60_000); + expect(listener).toHaveBeenCalledTimes(2); + }); + + it("does not poll while the stream is healthy, and polls only after it stalls or fails", async () => { + vi.useFakeTimers(); + const events = new EventTarget(); + vi.stubGlobal("EventSource", vi.fn(function () { return Object.assign(events, { close: vi.fn() }); })); + const listener = vi.fn(); + const leave = subscribeTaskFreshness(listener); + events.dispatchEvent(new Event("ready")); + for (let i = 0; i < 4; i++) { // Resync frames every 15 s keep the stream healthy. + await vi.advanceTimersByTimeAsync(15_000); + events.dispatchEvent(new Event("resync")); + } + await vi.advanceTimersByTimeAsync(25); + expect(listener).toHaveBeenCalledTimes(5); // Signals only, no timer reads. + events.dispatchEvent(new Event("error")); // Brief reconnect gap: still no polling. + await vi.advanceTimersByTimeAsync(1_000); + events.dispatchEvent(new Event("ready")); + await vi.advanceTimersByTimeAsync(25); + expect(listener).toHaveBeenCalledTimes(6); + events.dispatchEvent(new Event("error")); // Stream stays down: fall back to polling. + await vi.advanceTimersByTimeAsync(5_000 + 7_500); + expect(listener.mock.calls.length).toBeGreaterThanOrEqual(7); + const during = listener.mock.calls.length; + events.dispatchEvent(new Event("ready")); // Recovered: polling stops. + await vi.advanceTimersByTimeAsync(25); + const after = listener.mock.calls.length; + expect(after).toBe(during + 1); + await vi.advanceTimersByTimeAsync(14_000); + expect(listener).toHaveBeenCalledTimes(after); leave(); }); }); diff --git a/src/lib/task-freshness.ts b/src/lib/task-freshness.ts index 3d6815d..803bc1c 100644 --- a/src/lib/task-freshness.ts +++ b/src/lib/task-freshness.ts @@ -1,27 +1,51 @@ /** One SSE connection per tab; consumers always fetch scoped durable projections. */ +const STALE_MS = 20_000; // The server sends a frame at least every 15 s. +const RECONNECT_GRACE_MS = 5_000; // EventSource retries after 1 s; a healthy stream sends `ready` at once. +const FALLBACK_MS = 5_000; + const listeners = new Set<() => void>(); let source: EventSource | undefined; let pending: ReturnType | undefined; +let watchdog: ReturnType | undefined; +let fallback: ReturnType | undefined; + +const notify = () => { for (const callback of listeners) callback(); }; + +/** Polling exists only while the stream is down; jitter keeps tabs from polling in lockstep. */ +function startFallback() { + if (fallback !== undefined) return; + const tick = () => { + fallback = setTimeout(() => { notify(); if (fallback !== undefined) tick(); }, FALLBACK_MS * (1 + Math.random() * 0.5)); + }; + tick(); +} +function stopFallback() { clearTimeout(fallback); fallback = undefined; } +function armWatchdog(ms: number) { clearTimeout(watchdog); watchdog = setTimeout(startFallback, ms); } export function subscribeTaskFreshness(listener: () => void) { listeners.add(listener); - if (!source && typeof EventSource !== "undefined") { - source = new EventSource("/api/tasks/events"); - const invalidate = () => { - if (pending !== undefined) return; - pending = setTimeout(() => { - pending = undefined; - for (const callback of listeners) callback(); - }, 25); - }; - // Every reconnect receives ready, including when all prior signals were lost. - for (const event of ["ready", "freshness", "resync"]) source.addEventListener(event, invalidate); + if (!source) { + if (typeof EventSource === "undefined") startFallback(); + else { + source = new EventSource("/api/tasks/events"); + const invalidate = () => { + stopFallback(); armWatchdog(STALE_MS); + if (pending !== undefined) return; + pending = setTimeout(() => { pending = undefined; notify(); }, 25); + }; + // Every reconnect receives ready, including when all prior signals were lost. + for (const event of ["ready", "freshness", "resync"]) source.addEventListener(event, invalidate); + source.addEventListener("error", () => armWatchdog(RECONNECT_GRACE_MS)); + armWatchdog(RECONNECT_GRACE_MS); + } } return () => { listeners.delete(listener); if (!listeners.size) { source?.close(); source = undefined; clearTimeout(pending); pending = undefined; + clearTimeout(watchdog); watchdog = undefined; + stopFallback(); } }; } diff --git a/src/lib/use-server-resource.ts b/src/lib/use-server-resource.ts index ab09403..324547e 100644 --- a/src/lib/use-server-resource.ts +++ b/src/lib/use-server-resource.ts @@ -2,22 +2,23 @@ import { useEffect, useReducer, useState } from "react"; import { subscribeTaskFreshness } from "./task-freshness"; +import { fetchJson } from "./conditional-fetch"; import { queuedRead } from "./queued-read"; -/** Read after live invalidations, with focus and periodic recovery for missed signals. */ +/** Read after live invalidations and on focus; task-freshness polls only while the live stream is down. */ export function useServerResource(url: string | null) { const [state, setState] = useState<{ url: string | null; data?: T; error?: string; loading: boolean }>({ url, loading: true }); const [revision, refresh] = useReducer((value: number) => value + 1, 0); useEffect(() => { if (!url) return; const controller = new AbortController(); - let timer: ReturnType; const load = queuedRead(async () => { try { - const response = await fetch(url, { cache: "no-store", signal: controller.signal }); - const body = await response.json(); - if (!response.ok) throw new Error(body.error?.message ?? "Could not load server state."); - if (!controller.signal.aborted) setState({ url, data: body as T, loading: false }); + const { ok, body, unchanged } = await fetchJson(url, controller.signal); + if (!ok) throw new Error(body.error?.message ?? "Could not load server state."); + // A 304 leaves a settled, error-free state as is, so nothing re-renders. + if (!controller.signal.aborted) setState((previous) => unchanged && previous.url === url && !previous.error && !previous.loading ? previous + : { url, data: body as T, loading: false }); } catch (error) { if (!controller.signal.aborted) setState((previous) => ({ url, data: previous.url === url ? previous.data : undefined, @@ -25,15 +26,11 @@ export function useServerResource(url: string | null) { })); } }, controller.signal); - const poll = async () => { - await load(); - if (!controller.signal.aborted) timer = setTimeout(poll, 5000); - }; - void poll(); + void load(); const onFocus = () => { void load(); }; const unsubscribe = subscribeTaskFreshness(onFocus); window.addEventListener("focus", onFocus); - return () => { controller.abort(); clearTimeout(timer); unsubscribe(); window.removeEventListener("focus", onFocus); }; + return () => { controller.abort(); unsubscribe(); window.removeEventListener("focus", onFocus); }; }, [url, revision]); return { data: state.url === url ? state.data : undefined, diff --git a/src/server/adapters/live/coalesced-freshness.test.ts b/src/server/adapters/live/coalesced-freshness.test.ts new file mode 100644 index 0000000..08ce347 --- /dev/null +++ b/src/server/adapters/live/coalesced-freshness.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it, vi } from "vitest"; +import { CoalescedFreshnessReader } from "./coalesced-freshness"; + +describe("coalesced freshness reads", () => { + it("shares in-flight and recent reads per organization and refreshes after the ttl", async () => { + let clock = 0; + let token = "a"; + const source = { readToken: vi.fn(async () => token) }; + const reader = new CoalescedFreshnessReader(source, 400, () => clock); + const first = await Promise.all(Array.from({ length: 100 }, () => reader.readToken("one"))); + expect(new Set(first)).toEqual(new Set(["a"])); + expect(source.readToken).toHaveBeenCalledTimes(1); + token = "b"; clock = 399; + expect(await reader.readToken("one")).toBe("a"); + await reader.readToken("two"); + expect(source.readToken).toHaveBeenCalledTimes(2); + clock = 400; + expect(await reader.readToken("one")).toBe("b"); + expect(source.readToken).toHaveBeenCalledTimes(3); + }); + + it("does not cache failures", async () => { + const source = { readToken: vi.fn().mockRejectedValueOnce(new Error("down")).mockResolvedValue("ok") }; + const reader = new CoalescedFreshnessReader(source, 400, () => 0); + await expect(Promise.all([reader.readToken("one"), reader.readToken("one")])).rejects.toThrow("down"); + expect(await reader.readToken("one")).toBe("ok"); + expect(source.readToken).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/server/adapters/live/coalesced-freshness.ts b/src/server/adapters/live/coalesced-freshness.ts new file mode 100644 index 0000000..9adc4e2 --- /dev/null +++ b/src/server/adapters/live/coalesced-freshness.ts @@ -0,0 +1,35 @@ +import type { FreshnessReader } from "../../application/ports/realtime"; + +type Entry = { at: number; token?: string; reading?: Promise }; + +/** + * Shares token reads between streams: every open stream polls, but the database sees at most one + * read per organization per `ttlMs` in this process. Idle organizations cost nothing. + */ +export class CoalescedFreshnessReader implements FreshnessReader { + private readonly entries = new Map(); + constructor(private readonly source: FreshnessReader, private readonly ttlMs = 400, private readonly now = Date.now) {} + + readToken(organizationId: string): Promise { + const entry = this.entries.get(organizationId); + if (entry?.reading) return entry.reading; + if (entry?.token !== undefined && this.now() - entry.at < this.ttlMs) return Promise.resolve(entry.token); + this.prune(); + const next: Entry = { at: entry?.at ?? 0 }; + const reading = this.source.readToken(organizationId).then((token) => { + next.token = token; next.at = this.now(); next.reading = undefined; + return token; + }, (error) => { + this.entries.delete(organizationId); // A failure is never cached; the next caller retries. + throw error; + }); + next.reading = reading; + this.entries.set(organizationId, next); + return reading; + } + + private prune() { + if (this.entries.size < 256) return; + for (const [key, entry] of this.entries) if (!entry.reading && this.now() - entry.at >= this.ttlMs) this.entries.delete(key); + } +} diff --git a/src/server/adapters/live/freshness-stream.ts b/src/server/adapters/live/freshness-stream.ts index 15fd16a..9dfe9df 100644 --- a/src/server/adapters/live/freshness-stream.ts +++ b/src/server/adapters/live/freshness-stream.ts @@ -1,6 +1,9 @@ import type { FreshnessReader } from "../../application/ports/realtime"; -/** Backpressure keeps at most one frame queued; every frame is an invalidation. */ +/** + * Backpressure keeps at most one frame queued; every frame is an invalidation. The default lifetime is + * jittered (40–55 s) so streams opened together, such as after a rollout, do not reconnect together. + */ export function freshnessStream(reader: FreshnessReader, organizationId: string, signal: AbortSignal, options: { pollMs?: number; resyncMs?: number; lifetimeMs?: number } = {}) { const encoder = new TextEncoder(); @@ -25,7 +28,7 @@ export function freshnessStream(reader: FreshnessReader, organizationId: string, start(controller) { close = () => { if (!closed) { cleanup(); controller.close(); } }; signal.addEventListener("abort", abort, { once: true }); - deadline = setTimeout(close, options.lifetimeMs ?? 55_000); + deadline = setTimeout(close, options.lifetimeMs ?? 40_000 + Math.random() * 15_000); if (signal.aborted) close(); }, async pull(controller) { diff --git a/src/server/runtime/freshness.ts b/src/server/runtime/freshness.ts index 8f17336..60d1276 100644 --- a/src/server/runtime/freshness.ts +++ b/src/server/runtime/freshness.ts @@ -2,6 +2,7 @@ import type { MikroORM } from "@mikro-orm/core"; import type { RealtimePublisher } from "../application/ports/realtime"; import { FreshnessDispatcher } from "../application/services/task-freshness"; import { DatabaseFreshness } from "../adapters/live/database-freshness"; +import { CoalescedFreshnessReader } from "../adapters/live/coalesced-freshness"; import { localFreshness } from "../adapters/live/in-process-freshness"; import { loadDatabaseConfig } from "../adapters/db/config"; import { withJobEntityManager } from "../adapters/db/orm"; @@ -11,6 +12,14 @@ export function runtimeFreshness() { return loadDatabaseConfig().profile === "pglite" ? localFreshness() : new DatabaseFreshness(); } +// One shared reader per process (surviving HMR) so open streams share database reads. +type SharedGlobal = typeof globalThis & { __a2aSharedFreshness?: CoalescedFreshnessReader }; +export function liveFreshnessReader() { + if (loadDatabaseConfig().profile === "pglite") return localFreshness(); + const state = globalThis as SharedGlobal; + return state.__a2aSharedFreshness ??= new CoalescedFreshnessReader(new DatabaseFreshness()); +} + export function createFreshnessDispatcher(options: { orm?: MikroORM; publisher?: RealtimePublisher } = {}) { return new FreshnessDispatcher({ run: (work) => withJobEntityManager((em) => em.transactional((tx) => work(createPersistenceRepositories(tx).outbox)), options.orm) }, options.publisher ?? runtimeFreshness());