diff --git a/AGENTS.md b/AGENTS.md index 275f6c3ff..fd354dde9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -424,6 +424,16 @@ so merge ranges from all mounted editors before replacing a registry entry. after `send()` loses edits on disconnect. - Rebuilds and reconnects must replay unacknowledged updates only when the epoch is still compatible. +- Shutdown closes every socket with 1012 (`OpenSockets.stopForRestart()`) before + `server.stop(true)`, whose own 1001 reads as a loss. The browser takes 1012 as a + quiet restart and a probe's 4xx as a refusal only with `x-chopin-refusal: 1` + (or a 401); answer any new refusal of `/ws` or `/ws/sidebar` through + `refused()` so a proxy's 4xx cannot be mistaken for it. A request waiting when + the 1012 arrives rejects as `WireLost` with `restarting` set; only a caller whose + request carries its own identity, as `chat:send` does, may send it again quietly. + For `RESTART_QUIET_MS` after a 1012 `Wire` raises no `onOutage`, so the document's one + offline notice stays away, and every other surface that speaks of a lost connection + waits `useConnectionNotice(lost, wire.restarting)`; a new one must honour both. - Room opening and closing share the document lifecycle lock. Opening waits for an active close; eviction forgets the room only after runtime shutdown and persistence complete. Preserve that ordering. diff --git a/apps/server/src/client.test.ts b/apps/server/src/client.test.ts index 498040166..0259ca098 100644 --- a/apps/server/src/client.test.ts +++ b/apps/server/src/client.test.ts @@ -61,7 +61,7 @@ test("unfingerprinted public files are not cached as immutable", async () => { test("HTML and SPA fallbacks are revalidated", async () => { let client = await createClient({ "index.html": "
app
" }); try { - for (let path of ["/index.html", "/rooms/room-123", "/assets/missing-AjK9nqi6.js"]) { + for (let path of ["/index.html", "/rooms/room-123", "/assets", "/assetsx/app.js"]) { let response = await client.get(path); expect(response.status).toBe(200); expect(response.headers.get("cache-control")).toBe("no-cache"); @@ -73,6 +73,35 @@ test("HTML and SPA fallbacks are revalidated", async () => { } }); +test("a missing file under /assets answers 404, where the app's HTML would fail as a script", async () => { + let client = await createClient({ + "index.html": "
app
", + "assets/index-AjK9nqi6.js": "console.log('build');", + }); + try { + for ( + let path of [ + "/assets/old-chunk-Zy8wXq1p.js", + "/assets/old-style-Zy8wXq1p.css", + "/assets/unhashed.js", + "/assets/nested/dir/chunk-Zy8wXq1p.js", + "/assets/", + ] + ) { + let response = await client.get(path); + expect([path, response.status]).toEqual([path, 404]); + expect(response.headers.get("content-type")).not.toContain("text/html"); + expect(response.headers.get("cache-control")).toBe("no-store"); + expect(await response.text()).not.toContain("
app
"); + } + let present = await client.get("/assets/index-AjK9nqi6.js"); + expect(present.status).toBe(200); + expect(present.headers.get("cache-control")).toBe("public, max-age=31536000, immutable"); + } finally { + await client.close(); + } +}); + test("a missing built client keeps its existing 404 response", async () => { let directory = await mkdtemp(join(tmpdir(), "chopin-client-")); try { diff --git a/apps/server/src/client.ts b/apps/server/src/client.ts index cd71e7550..bccd71429 100644 --- a/apps/server/src/client.ts +++ b/apps/server/src/client.ts @@ -32,6 +32,7 @@ const PRIVATE = new Set(["authorization", "cookie"]); /** Vite's default build names use an eight-character hash before the extension. */ const FINGERPRINTED_ASSET_PATH = /^\/assets\/.+-[A-Za-z0-9_-]{8}\.[^/]+$/; +const ASSETS_PATH = "/assets/"; const IMMUTABLE_CACHE_CONTROL = "public, max-age=31536000, immutable"; function forwardable(headers: Headers): Headers { @@ -76,7 +77,10 @@ export async function proxy(req: Request, url: URL, origin: string): Promise { if (!existsSync(dir)) { @@ -96,6 +100,12 @@ export async function serve(url: URL, dir: string): Promise { } return new Response(file, { headers }); } + if (url.pathname.startsWith(ASSETS_PATH)) { + return new Response("not found", { + status: 404, + headers: { "cache-control": "no-store", "content-type": "text/plain; charset=utf-8" }, + }); + } return new Response(Bun.file(join(dir, "index.html")), { headers: { "cache-control": "no-cache" }, }); diff --git a/apps/server/src/http/health.test.ts b/apps/server/src/http/health.test.ts new file mode 100644 index 000000000..9f50be3b9 --- /dev/null +++ b/apps/server/src/http/health.test.ts @@ -0,0 +1,80 @@ +import { expect, test } from "bun:test"; + +import { Readiness, registerHealthRoutes } from "./health"; +import { Router } from "./router"; + +function service(holdsLease = () => true) { + let readiness = new Readiness(holdsLease); + let router = new Router(); + registerHealthRoutes(router, readiness); + let get = async (path: string, method = "GET") => { + let response = (await router.handle(new Request(`http://chopin.test${path}`, { method })))!; + return { status: response.status, body: await response.text(), response }; + }; + return { + readiness, + live: () => get("/api/health/live"), + ready: () => get("/api/health/ready"), + get, + }; +} + +test("readiness is false until startup recovery has finished", async () => { + let { readiness, ready } = service(); + expect(await ready()).toMatchObject({ status: 503, body: "starting" }); + readiness.started(); + expect(await ready()).toMatchObject({ status: 200, body: "ready" }); +}); + +test("readiness needs the writer lease as well as finished recovery", async () => { + let held = false; + let { readiness, ready } = service(() => held); + readiness.started(); + expect(await ready()).toMatchObject({ status: 503, body: "starting" }); + held = true; + expect(await ready()).toMatchObject({ status: 200, body: "ready" }); +}); + +test("readiness is false from the moment a drain starts, and stays false", async () => { + let { readiness, ready } = service(); + readiness.started(); + expect((await ready()).status).toBe(200); + readiness.draining(); + expect(await ready()).toMatchObject({ status: 503, body: "draining" }); + readiness.started(); + expect(await ready()).toMatchObject({ status: 503, body: "draining" }); +}); + +test("a drain that begins during startup keeps the process from ever being ready", async () => { + let { readiness, ready } = service(); + readiness.draining(); + readiness.started(); + expect(await ready()).toMatchObject({ status: 503, body: "draining" }); +}); + +test("liveness is true while the process runs, whatever readiness says", async () => { + let { readiness, live } = service(() => false); + expect(await live()).toMatchObject({ status: 200, body: "live" }); + readiness.started(); + readiness.draining(); + expect(await live()).toMatchObject({ status: 200, body: "live" }); +}); + +test("both answer one word to a caller with no credentials, and are never cached", async () => { + let { readiness, live, ready } = service(); + readiness.started(); + for (let { status, body, response } of [await live(), await ready()]) { + expect(status).toBe(200); + expect(body).toMatch(/^[a-z]+$/); + expect(response.headers.get("cache-control")).toBe("no-store"); + expect(response.headers.get("set-cookie")).toBeNull(); + } +}); + +test("a probe may ask with HEAD and gets the status without a body", async () => { + let { readiness, get } = service(); + expect(await get("/api/health/ready", "HEAD")).toMatchObject({ status: 503, body: "" }); + readiness.started(); + expect(await get("/api/health/ready", "HEAD")).toMatchObject({ status: 200, body: "" }); + expect(await get("/api/health/live", "HEAD")).toMatchObject({ status: 200, body: "" }); +}); diff --git a/apps/server/src/http/health.ts b/apps/server/src/http/health.ts new file mode 100644 index 000000000..bf913766d --- /dev/null +++ b/apps/server/src/http/health.ts @@ -0,0 +1,57 @@ +import type { Router } from "./router"; + +export type ReadinessStatus = "starting" | "ready" | "draining"; + +/** + * Whether this process may be sent traffic. + * + * Ready only once startup recovery has finished and the process holds the writer lease, and + * never again once a drain has begun: a replacement waits for exactly this process to let go. + */ +export class Readiness { + readonly #holdsLease: () => boolean; + #started = false; + #draining = false; + + constructor(holdsLease: () => boolean) { + this.#holdsLease = holdsLease; + } + + /** Startup recovery has finished. */ + started(): void { + this.#started = true; + } + + /** A drain has begun. Irreversible, because the process is on its way out. */ + draining(): void { + this.#draining = true; + } + + status(): ReadinessStatus { + if (this.#draining) return "draining"; + return this.#started && this.#holdsLease() ? "ready" : "starting"; + } +} + +function answer(request: Request, status: number, word: string): Response { + return new Response(request.method === "HEAD" ? null : word, { + status, + headers: { "cache-control": "no-store", "content-type": "text/plain; charset=utf-8" }, + }); +} + +/** + * `GET /api/health/live` answers while the process runs, and `GET /api/health/ready` says + * whether it may be sent traffic. Both are unauthenticated and say one word, nothing else. + */ +export function registerHealthRoutes(router: Router, readiness: Readiness): void { + let live = (request: Request) => answer(request, 200, "live"); + let ready = (request: Request) => { + let status = readiness.status(); + return answer(request, status === "ready" ? 200 : 503, status); + }; + for (let method of ["GET", "HEAD"]) { + router.on(method, "/api/health/live", live); + router.on(method, "/api/health/ready", ready); + } +} diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts index cf8033695..86831c0b7 100644 --- a/apps/server/src/main.ts +++ b/apps/server/src/main.ts @@ -65,6 +65,7 @@ import { describe, load } from "./config"; import { GitHubClient, GitHubError } from "./github/client"; import { GitHubReferences } from "./github/references"; import { registerConnectorScript } from "./http/connector-script"; +import { Readiness, registerHealthRoutes } from "./http/health"; import { registerMachinesPath } from "./http/machines-path"; import { Router } from "./http/router"; import { JobExecutionError, JobRegistry } from "./jobs/registry"; @@ -111,7 +112,7 @@ import { } from "./research/service"; import { placeResearchReference as placeResearch } from "./research/placement"; import * as Rooms from "./rooms"; -import { admit, admitSidebar } from "./socket/admission"; +import { admit, admitSidebar, refused } from "./socket/admission"; import { refreshAuthorization } from "./socket/authorization"; import { recheckDecisionWatch, @@ -122,6 +123,7 @@ import { watchDecisions, watchRequest, } from "./socket/decision-watch"; +import { OpenSockets } from "./socket/open-sockets"; import { SessionSockets } from "./socket/session-sockets"; import { StorageError } from "./storage/errors"; import { createStorage } from "./storage/registry"; @@ -138,6 +140,7 @@ import { } from "./wire"; import { AGENT_PATH, LINK_CLOSE } from "@chopin/agent-link"; +import { SERVICE_RESTART } from "@chopin/protocol/restart"; import type { Server, ServerWebSocket } from "bun"; import type { AgentSocketData } from "./harness/remote/endpoint"; @@ -192,6 +195,7 @@ const writerLease = new WriterLease(storage.leases, { safetyMs: LEASE_SAFETY_MS, retryMs: LEASE_RETRY_MS, }, () => signal()); +const readiness = new Readiness(() => writerLease.held !== undefined); let experiments: ExperimentRuntime | undefined; let implementations: ReturnType | undefined; let experimentTimer: ReturnType | undefined; @@ -203,6 +207,7 @@ let plannerAttachments: PlannerAttachments | undefined; let systemTurns = new SystemTurns(); /** Browser sockets by login, so a login that ends closes them before another broadcast. */ let sessionSockets = new SessionSockets(sessionId => hostedAuth.sessions.isLive(sessionId)); +const openSockets = new OpenSockets(); let jobRunner: JobRunner | undefined; let researchService: ResearchWorkspaceService | undefined; let researchRecoveryTimer: ReturnType | undefined; @@ -1783,9 +1788,7 @@ function listen(): Server { deleting: channelId => deletingChannels.has(channelId), readOnly: channelId => archivingChannels.has(channelId), }); - if ("status" in outcome) { - return new Response(outcome.reason, { status: outcome.status }); - } + if ("status" in outcome) return refused(outcome); return upgraded(req, self, outcome.data); } if (url.pathname === AGENT_PATH) { @@ -1800,9 +1803,7 @@ function listen(): Server { } if (url.pathname === "/ws/sidebar") { let outcome = await admitSidebar(req, hostedAuth); - if ("status" in outcome) { - return new Response(outcome.reason, { status: outcome.status }); - } + if ("status" in outcome) return refused(outcome); return upgraded(req, self, outcome.data); } let routed = await router.handle(req, url); @@ -1817,6 +1818,7 @@ function listen(): Server { websocket: { open(ws: AnySocket) { + openSockets.opened(ws); if (isAgentSocket(ws)) { if (!plannerAttachments) return ws.close(LINK_CLOSE.failed, "unavailable"); openAgentSocket(ws, hostedAuth, { @@ -1852,6 +1854,7 @@ function listen(): Server { }, close(ws: AnySocket) { + openSockets.closed(ws); if (isAgentSocket(ws)) { ws.data.alive?.stop(); return ws.data.link?.disconnected(); @@ -1882,9 +1885,10 @@ function drain(): Promise { record(err); } }; + readiness.draining(); // Before the sockets close, so no link withdraws its questions on the way out. plannerAttachments?.shutdown(); - if (server) await attempt(() => server.stop(true)); + if (server) await attempt(() => openSockets.stopForRestart(server)); if (researchRecoveryTimer) clearTimeout(researchRecoveryTimer); if (experimentTimer) clearInterval(experimentTimer); if (recoveringResearch) await attempt(() => recoveringResearch!); @@ -2142,7 +2146,7 @@ async function deleteChannelLocked(channelId: string): Promise { channel = await storage.channels.get(channelId); } catch { for (let ws of room?.members.values() ?? []) { - ws.close(1012, "document state is temporarily unavailable"); + ws.close(SERVICE_RESTART, "document state is temporarily unavailable"); } throw err; } @@ -2153,7 +2157,7 @@ async function deleteChannelLocked(channelId: string): Promise { await plan(survivingRoom, server); } catch { for (let ws of survivingRoom.members.values()) { - ws.close(1012, "document reload required"); + ws.close(SERVICE_RESTART, "document reload required"); } } } @@ -2892,6 +2896,7 @@ registerConnectorScript(router, { source: join(import.meta.dir, "../../connector/src/main.ts"), }); registerMachinesPath(router); +registerHealthRoutes(router, readiness); try { await storage.health(); @@ -2957,6 +2962,7 @@ try { scheduleResearchRecovery(recovery.deferred); await researchService.recoverTerminalPlannerInline(); jobRunner.start(); + readiness.started(); } catch (err) { if (listening) await listening.stop(true).catch(() => {}); if (researchRecoveryTimer) clearTimeout(researchRecoveryTimer); diff --git a/apps/server/src/socket/admission.test.ts b/apps/server/src/socket/admission.test.ts index 4ee957b46..51cf26628 100644 --- a/apps/server/src/socket/admission.test.ts +++ b/apps/server/src/socket/admission.test.ts @@ -1,3 +1,4 @@ +import { REFUSAL_HEADER } from "@chopin/protocol/restart"; import { Machines } from "../auth/machines"; import { describe, expect, it } from "bun:test"; @@ -5,7 +6,7 @@ import { Sessions } from "../auth/session"; import { Admission } from "../auth/admission"; import { GitHubError } from "../github/client"; import { MemoryStorage } from "../storage/memory/adapter"; -import { admit } from "./admission"; +import { admit, refused } from "./admission"; import type { HostedAuth } from "../auth/routes"; import type { @@ -129,6 +130,13 @@ describe("socket admission", () => { auth, ); expect(wrongOrigin).toEqual({ status: 403, reason: "origin is not allowed" }); + if (!("status" in wrongOrigin)) throw new Error("expected a refusal"); + let answer = refused(wrongOrigin); + expect([answer.status, await answer.text(), answer.headers.get(REFUSAL_HEADER)]).toEqual([ + 403, + "origin is not allowed", + "1", + ]); // A page still open on the name the deployment moved from keeps its socket. let previousOrigin = await admit( new Request(url, { diff --git a/apps/server/src/socket/admission.ts b/apps/server/src/socket/admission.ts index 3b4c30d2a..262087dba 100644 --- a/apps/server/src/socket/admission.ts +++ b/apps/server/src/socket/admission.ts @@ -1,3 +1,5 @@ +import { REFUSAL_HEADER } from "@chopin/protocol/restart"; + import { ownOrigin } from "../auth/config"; import { isChannelId } from "../channels/id"; import { GitHubError } from "../github/client"; @@ -13,6 +15,17 @@ export type Admission = { data: SocketData } | { status: number; reason: string export type SidebarAdmission = { data: SidebarSocketData } | { status: number; reason: string }; +/** + * The HTTP answer to a socket handshake that admission refused, marked as Chopin's own so the + * client can tell it from a proxy's 4xx while Chopin is away. + */ +export function refused(outcome: { status: number; reason: string }): Response { + return new Response(outcome.reason, { + status: outcome.status, + headers: { [REFUSAL_HEADER]: "1" }, + }); +} + function probing(request: Request): boolean { return request.headers.get("x-chopin-socket-probe") === "1" && request.headers.get("upgrade")?.toLowerCase() !== "websocket"; diff --git a/apps/server/src/socket/open-sockets.test.ts b/apps/server/src/socket/open-sockets.test.ts new file mode 100644 index 000000000..783d449a4 --- /dev/null +++ b/apps/server/src/socket/open-sockets.test.ts @@ -0,0 +1,74 @@ +import { SERVICE_RESTART } from "@chopin/protocol/restart"; +import { afterEach, expect, test } from "bun:test"; + +import { OpenSockets } from "./open-sockets"; + +import type { Server, ServerWebSocket } from "bun"; + +let clients: WebSocket[] = []; + +afterEach(() => { + for (let client of clients) client.close(); + clients = []; +}); + +function serve(sockets: OpenSockets) { + let server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + fetch(request, self) { + if (self.upgrade(request)) return undefined; + return new Response("plain"); + }, + websocket: { + open(ws: ServerWebSocket) { + sockets.opened(ws); + }, + close(ws: ServerWebSocket) { + sockets.closed(ws); + }, + message() {}, + }, + }); + return server; +} + +async function connect(server: Server): Promise<{ closed: Promise }> { + let client = new WebSocket(`ws://127.0.0.1:${server.port}/`); + clients.push(client); + let closed = new Promise(resolve => client.addEventListener("close", resolve)); + await new Promise(resolve => client.addEventListener("open", resolve, { once: true })); + return { closed }; +} + +test("every open socket is closed with 1012, where server.stop(true) alone says 1001", async () => { + let plain = serve(new OpenSockets()); + let before = await connect(plain); + await plain.stop(true); + expect((await before.closed).code).toBe(1001); + + let sockets = new OpenSockets(); + let server = serve(sockets); + let open = await Promise.all([connect(server), connect(server), connect(server)]); + await sockets.stopForRestart(server); + for (let socket of open) { + let event = await socket.closed; + expect([event.code, event.reason, event.wasClean]).toEqual([ + SERVICE_RESTART, + "server restarting", + true, + ]); + } +}); + +test("a socket that closed on its own is forgotten, and the listener stops taking new ones", async () => { + let sockets = new OpenSockets(); + let server = serve(sockets); + let leaving = await connect(server); + clients.at(-1)!.close(1000, "done"); + expect((await leaving.closed).code).toBe(1000); + let staying = await connect(server); + await sockets.stopForRestart(server); + expect((await staying.closed).code).toBe(SERVICE_RESTART); + await expect(fetch(`http://127.0.0.1:${server.port}/`)).rejects.toThrow(); +}); diff --git a/apps/server/src/socket/open-sockets.ts b/apps/server/src/socket/open-sockets.ts new file mode 100644 index 000000000..b59536251 --- /dev/null +++ b/apps/server/src/socket/open-sockets.ts @@ -0,0 +1,31 @@ +import { SERVICE_RESTART } from "@chopin/protocol/restart"; + +import type { ServerWebSocket } from "bun"; + +type Listener = { stop(closeActiveConnections?: boolean): Promise }; + +/** + * Every socket the server has open, so that stopping can end each with the code that says why. + * + * `server.stop(true)` closes every socket with 1001, "going away", which a browser cannot tell + * from a server it lost for good. A restart is closed with 1012 first. + */ +export class OpenSockets { + readonly #open = new Set>(); + + opened(ws: ServerWebSocket): void { + this.#open.add(ws); + } + + closed(ws: ServerWebSocket): void { + this.#open.delete(ws); + } + + /** Stop accepting connections, close every open socket with 1012, then drop what is left. */ + async stopForRestart(server: Listener): Promise { + let stopping = server.stop(); + for (let ws of this.#open) ws.close(SERVICE_RESTART, "server restarting"); + await server.stop(true); + await stopping; + } +} diff --git a/apps/server/src/storage/postgres/startup.test.ts b/apps/server/src/storage/postgres/startup.test.ts index 0ed8efabc..f08df5e76 100644 --- a/apps/server/src/storage/postgres/startup.test.ts +++ b/apps/server/src/storage/postgres/startup.test.ts @@ -88,6 +88,11 @@ if (database) { await ready(9071); let session = await fetch("http://127.0.0.1:9071/api/session"); expect(await session.json()).toEqual({ user: null, agent: false }); + // Started means the lease is held and recovery has finished, for a caller with no login. + for (let [path, answer] of [["live", "live"], ["ready", "ready"]]) { + let health = await fetch(`http://127.0.0.1:9071/api/health/${path}`); + expect([health.status, await health.text()]).toEqual([200, answer]); + } let storage = new PostgresStorage(database); let now = new Date(); diff --git a/apps/web/bundle-budget.ts b/apps/web/bundle-budget.ts index abd7901d6..ca927aff7 100644 --- a/apps/web/bundle-budget.ts +++ b/apps/web/bundle-budget.ts @@ -24,9 +24,12 @@ export type JavaScriptBudget = { gzip: number; raw: number }; // while retaining the raw limit and keeping the editor out of project-only navigation. // A document's Machine, in the title menu every sidebar row shares, adds 500 B raw. // The sidebar's Machines menu, on Base UI Menu, adds about 50 B gzip on Linux; allow 200 B. +// Reloading for a chunk a deploy removed, with the registry of unsaved work it asks, and the +// reconnect that reads a server restart, measure 258,947 B raw and 82,789 B gzip in the +// container's Linux build; allow about 500 B raw and 200 B gzip above that. export const INITIAL_JAVASCRIPT_BUDGET: JavaScriptBudget = { - gzip: 82_328, - raw: 258_000, + gzip: 83_000, + raw: 259_500, }; export function enforceInitialJavaScript( diff --git a/apps/web/src/build-view.tsx b/apps/web/src/build-view.tsx index 88513ba8c..d57507c94 100644 --- a/apps/web/src/build-view.tsx +++ b/apps/web/src/build-view.tsx @@ -1,3 +1,4 @@ +import { usePendingWork } from "@chopin/editor"; import { Face } from "@chopin/editor/face"; import { CheckIcon, @@ -132,6 +133,7 @@ export function BuildView( let [stopping, setStopping] = useState(false); let [returning, setReturning] = useState(false); let [reason, setReason] = useState(""); + usePendingWork(returning && reason.trim() !== ""); let [open, setOpen] = useState>({}); let [linked, setLinked] = useState(); let [now, setNow] = useState(() => Date.now()); diff --git a/apps/web/src/chat/chat.tsx b/apps/web/src/chat/chat.tsx index feb868b28..6a4c7efa2 100644 --- a/apps/web/src/chat/chat.tsx +++ b/apps/web/src/chat/chat.tsx @@ -18,13 +18,15 @@ import { } from "react"; import { - CONNECTION_GRACE, + connectionGrace, isDemoHandle, MAX_RESEARCH_BRIEF, useConnectionNotice, + usePendingWork, usePopoverDismissal, } from "@chopin/editor"; import { chatDraftKey } from "../demo-actor"; +import { WireLost } from "../wire"; import { canWithdraw } from "./withdraw"; import type { DraftInputHandle } from "./draft-input"; import type { PlannerNotice } from "./planner-status"; @@ -77,7 +79,12 @@ import { import { runCounts, RunStack } from "./run-card"; import { runStages } from "../api"; import { Transcript } from "./transcript"; -import { initialTranscript, transcriptReducer } from "./transcript-state"; +import { + initialTranscript, + progressOf, + silentRestart, + transcriptReducer, +} from "./transcript-state"; import type { TranscriptDecisions } from "./transcript"; import type { CardLink } from "../conversation-plan/links"; import type { ExcerptCorrectionAction } from "../conversation-plan/analysis-overview"; @@ -270,6 +277,7 @@ export function Chat( // A reload must not take an unsent message with it. Kept as it changes // rather than on unload, which a browser does not promise to announce. useEffect(() => keepDraft(room, handle, draft), [room, handle, draft]); + usePendingWork(!!draft.text.trim()); let pickerId = useId(); let mentionPickerId = useId(); let commandPickerId = useId(); @@ -290,18 +298,21 @@ export function Chat( // A draft never waits on the connection; only sending does. let draftable = !readonly && !archived; // The document says when the connection is lost; Chat only holds back what it cannot send. - let connectionNotice = useConnectionNotice(!connected); + let connectionNotice = useConnectionNotice(!connected, wire?.restarting); let effectiveMode = !!agent && (mode || draft.submission?.to === "planner" || addressedOutsideReferences(draft.text, draft.references)); - let workingTurn = transcriptReady ? turn : undefined; - let suspendedWork = !transcriptReady && turn && transcript.activeAnchorId - ? { - turnId: turn.id, - entryOffset: turn.entryOffset, - endOffset: entries.length, - anchorId: transcript.activeAnchorId, - } - : undefined; + let quietRestart = useRef(false); + quietRestart.current = silentRestart( + quietRestart.current, + !!wire?.restarting, + transcriptReady, + connectionNotice !== "none", + ); + let { working: workingTurn, suspended: suspendedWork } = progressOf( + transcript, + transcriptReady, + quietRestart.current, + ); let detected = referencesEnabled && composerReady && !submitting ? referenceTrigger(draft.text, selection.start, selection.end) : undefined; @@ -545,7 +556,10 @@ export function Chat( void wire.ask("chat:send", payload).then(() => { clearSubmittedDraft(submitted); }, error => { - setSendError(boundedChatError(error)); + // A restart cut the send off. The draft keeps its request id, so it is made again once + // the server is back, and the server answers one it already holds with that message. + if (error instanceof WireLost && error.restarting) setHeld(true); + else setSendError(boundedChatError(error)); }).finally(() => { if (submission.current !== token) return; submission.current = undefined; @@ -571,7 +585,7 @@ export function Chat( // Never held longer than the grace period, however the wait ends. useEffect(() => { if (!held) return; - let timer = setTimeout(() => setHeld(false), CONNECTION_GRACE); + let timer = setTimeout(() => setHeld(false), connectionGrace(wire?.restarting)); return () => clearTimeout(timer); }, [held]); useEffect(() => { diff --git a/apps/web/src/chat/research-draft-controller.test.ts b/apps/web/src/chat/research-draft-controller.test.ts index 7d9520594..d08cc5401 100644 --- a/apps/web/src/chat/research-draft-controller.test.ts +++ b/apps/web/src/chat/research-draft-controller.test.ts @@ -1,4 +1,5 @@ import { expect, test } from "bun:test"; +import { hasPendingWork } from "@chopin/protocol/pending-work"; import type { ConversationPlan } from "@chopin/protocol"; import { researchDraftHarness } from "../../../server/src/conversation-plan/research-draft.test-fixtures"; import { ResearchDraftController } from "./research-draft-controller"; @@ -36,3 +37,36 @@ test("lost edit acknowledgements retain patches for idempotent reconnect replay" expect(controller.get().text).toBe(h.offer().brief); expect(controller.get().syncing).toBe(false); }); + +test("a brief edit holds the page from reloading until the server has acknowledged it", async () => { + let h = researchDraftHarness(); + let lose = false; + let wire = { + connected: true, + send() {}, + async ask( + _kind: string, + payload: Omit, + ): Promise { + let result = await h.drafts.edit(payload.offerId, payload.operation, h.actor); + if (lose && payload.operation.kind === "patch") throw new Error("lost acknowledgement"); + return result as T; + }, + } as unknown as Pick; + let controller = new ResearchDraftController(h.offer()); + controller.configure(wire, true, h.offer()); + await controller.begin(); + expect(controller.unsaved).toBe(false); + + lose = true; + controller.change("Investigate Jev alternatives and costs."); + await expect(controller.flush()).rejects.toThrow("not synchronized"); + expect(controller.unsaved).toBe(true); + expect(hasPendingWork()).toBe(true); + + lose = false; + controller.configure(wire, false, h.offer()); + controller.configure(wire, true, h.offer()); + await controller.flush(); + expect(controller.unsaved).toBe(false); +}); diff --git a/apps/web/src/chat/research-draft-controller.ts b/apps/web/src/chat/research-draft-controller.ts index d0cda6371..57f8208eb 100644 --- a/apps/web/src/chat/research-draft-controller.ts +++ b/apps/web/src/chat/research-draft-controller.ts @@ -1,4 +1,5 @@ import * as Draft from "@chopin/draft"; +import { trackPendingWork } from "@chopin/protocol/pending-work"; import type { ConversationPlan } from "@chopin/protocol"; import type { Wire } from "../wire"; @@ -33,6 +34,12 @@ export class ResearchDraftController { syncing: false, closed: offer.status !== "offered", }; + trackPendingWork(this); + } + + /** Whether an edit of the brief has not reached the server, which a reload would lose. */ + get unsaved(): boolean { + return this.#outbox.length > 0; } get = () => this.#snapshot; subscribe = (listener: () => void) => { diff --git a/apps/web/src/chat/transcript-state.test.ts b/apps/web/src/chat/transcript-state.test.ts index 05323cd7f..877195f0a 100644 --- a/apps/web/src/chat/transcript-state.test.ts +++ b/apps/web/src/chat/transcript-state.test.ts @@ -1,7 +1,12 @@ import { describe, expect, it } from "bun:test"; import { group } from "./model"; -import { initialTranscript, transcriptReducer } from "./transcript-state"; +import { + initialTranscript, + progressOf, + silentRestart, + transcriptReducer, +} from "./transcript-state"; import type { Chat } from "@chopin/protocol"; @@ -119,3 +124,70 @@ describe("transcript work retention", () => { }]); }); }); + +describe("how a turn in progress is shown while its connection is lost", () => { + function working() { + let state = transcriptReducer(initialTranscript, { + kind: "history", + entries: [entry("prompt", { kind: "member", handle: "ana" })], + turn, + }); + return transcriptReducer(state, { + kind: "message", + entry: entry("prose", { kind: "agent" }, "I will inspect this."), + }); + } + + it("goes on showing a turn as working while its connection is current", () => { + let shown = progressOf(working(), true, false); + expect(shown.working?.id).toBe("turn-1"); + expect(shown.suspended).toBeUndefined(); + }); + + it("goes on showing it as working while a restart is waited out in silence, so a restart leaves no trace", () => { + let shown = progressOf(working(), false, true); + expect(shown.working?.id).toBe("turn-1"); + expect(shown.suspended).toBeUndefined(); + }); + + it("marks it disconnected as soon as a connection that was not a restart is gone", () => { + let shown = progressOf(working(), false, false); + expect(shown.working).toBeUndefined(); + expect(shown.suspended).toEqual({ + turnId: "turn-1", + entryOffset: 1, + endOffset: 2, + anchorId: "prose", + }); + }); + + it("has nothing to mark before the turn has written anything", () => { + let state = transcriptReducer(initialTranscript, { + kind: "history", + entries: [entry("prompt", { kind: "member", handle: "ana" })], + turn, + }); + expect(progressOf(state, false, false)).toEqual({}); + expect(progressOf(initialTranscript, false, false)).toEqual({}); + }); +}); + +describe("when a restart is waited out in silence", () => { + it("begins with a loss the server said was a restart, and with no other", () => { + expect(silentRestart(false, true, false, false)).toBe(true); + expect(silentRestart(false, false, false, false)).toBe(false); + }); + + it("lasts through the moment after the connection opens, until the transcript is current", () => { + // The connection is back, so it no longer says restarting, but the history has not come. + expect(silentRestart(true, false, false, false)).toBe(true); + expect(silentRestart(true, false, true, false)).toBe(false); + }); + + it("ends once the connection is spoken of, and does not begin again in that outage", () => { + expect(silentRestart(true, true, false, true)).toBe(false); + expect(silentRestart(false, true, false, true)).toBe(false); + // Back after a long outage: the next loss's marker is cleared, and nothing is silent. + expect(silentRestart(false, false, false, false)).toBe(false); + }); +}); diff --git a/apps/web/src/chat/transcript-state.ts b/apps/web/src/chat/transcript-state.ts index 64f6b5461..ec29c0a40 100644 --- a/apps/web/src/chat/transcript-state.ts +++ b/apps/web/src/chat/transcript-state.ts @@ -16,6 +16,47 @@ export type TranscriptAction = | { kind: "turn"; turn?: Chat.Turn } | { kind: "responded" }; +/** + * How the turn in progress is shown. `current` says the transcript is this connection's own, and + * `quiet` that a restart is being waited out in silence (`silentRestart`). + * + * A turn whose transcript is not current is marked disconnected, as soon as the connection that + * carried it is gone. A restart is the exception: the page says nothing of it, so the turn goes on + * showing as working and a restart that ends in moments leaves no trace. + */ +export function progressOf( + state: TranscriptState, + current: boolean, + quiet: boolean, +): { working?: Chat.Turn; suspended?: CompletedWork } { + if (current || quiet) return { working: state.turn }; + if (!state.turn || !state.activeAnchorId) return {}; + return { + suspended: { + turnId: state.turn.id, + entryOffset: state.turn.entryOffset, + endOffset: state.entries.length, + anchorId: state.activeAnchorId, + }, + }; +} + +/** + * Whether a restart is being waited out in silence, given whether it was a moment ago. It begins + * with a loss the server said was a restart (`restarting`) and lasts through the moment after the + * connection opens again, until the transcript is current (`current`). It ends once the page speaks + * of the connection (`spoken`) and does not begin again for the rest of that outage. A loss that + * was not a restart never begins it. + */ +export function silentRestart( + was: boolean, + restarting: boolean, + current: boolean, + spoken: boolean, +): boolean { + return !current && !spoken && (was || restarting); +} + export let initialTranscript: TranscriptState = { entries: [], completedWork: [] }; function anchor(entries: Chat.Entry[], turn: Chat.Turn): string | undefined { diff --git a/apps/web/src/document-rename.tsx b/apps/web/src/document-rename.tsx index 6ae328457..aa772fcb9 100644 --- a/apps/web/src/document-rename.tsx +++ b/apps/web/src/document-rename.tsx @@ -1,5 +1,6 @@ import { useEffect, useRef, useState } from "react"; import { boundDocumentTitle, normalizeDocumentTitle } from "@chopin/dialect"; +import { usePendingWork } from "@chopin/editor"; import * as Api from "./api"; import { heldTyping, holdTyping } from "./title-edit"; @@ -44,6 +45,7 @@ export function DocumentRename( let [title, setTitle] = useState(channel.title); let [error, setError] = useState(); let [saving, setSaving] = useState(false); + usePendingWork(saving || normalizeDocumentTitle(title) !== channel.title); // Enter and the blur it can cause arrive before React re-renders `saving`. let busy = useRef(false); let cancelled = useRef(false); diff --git a/apps/web/src/experiments/evidence.tsx b/apps/web/src/experiments/evidence.tsx index c46b8fcd1..83737ba41 100644 --- a/apps/web/src/experiments/evidence.tsx +++ b/apps/web/src/experiments/evidence.tsx @@ -1,3 +1,4 @@ +import { usePendingWork } from "@chopin/editor"; import { useState, useSyncExternalStore } from "react"; import { ExperimentView } from "@chopin/experiment/react"; import type { PublishedInvestigation } from "@chopin/experiment/records"; @@ -19,6 +20,7 @@ export function EvidenceActions({ store, item, view, disabled }: { let [busy, setBusy] = useState(false); let [error, setError] = useState(""); let [operation] = useState(() => new RequestIdentity()); + usePendingWork(open && (conclusion.trim() !== "" || rationale.trim() !== "")); async function act(action: () => Promise) { setBusy(true); setError(""); diff --git a/apps/web/src/main.tsx b/apps/web/src/main.tsx index 1226d867a..aa74334e0 100644 --- a/apps/web/src/main.tsx +++ b/apps/web/src/main.tsx @@ -8,6 +8,7 @@ import { isDesignAuditRoute } from "./design-audit/route"; import { isDiagramGalleryRoute } from "./diagram-gallery/route"; import { useFocusInput } from "./focus-input"; import { useMotionInput } from "./motion-input"; +import { watchPreloadErrors } from "./preload-recovery"; import { watchScrollbarStyle } from "./scrollbars"; import { useVisualViewport } from "./viewport"; @@ -29,6 +30,7 @@ import "./favorite-star.css"; let root = document.getElementById("root"); if (!root) throw new Error("missing #root"); watchScrollbarStyle(); +watchPreloadErrors(); function Root() { useFocusInput(); diff --git a/apps/web/src/preload-recovery.test.ts b/apps/web/src/preload-recovery.test.ts new file mode 100644 index 000000000..89c06c8ae --- /dev/null +++ b/apps/web/src/preload-recovery.test.ts @@ -0,0 +1,154 @@ +import { hasPendingWork } from "@chopin/protocol/pending-work"; +import { describe, expect, it } from "bun:test"; + +import { ResearchDraftStore } from "../../../packages/editor/src/research-draft"; +import { entryOf, isNewerBuild, recoverFromPreloadError, RELOAD_GAP_MS } from "./preload-recovery"; + +import type { Reloading } from "./preload-recovery"; + +function page( + options: { + pending?: boolean | (() => boolean); + newer?: boolean; + storage?: "none" | "refusing"; + } = {}, +) { + let values = new Map(); + let reloads = 0; + let clock = 1_000_000; + let state = { pending: options.pending ?? false, newer: options.newer ?? true }; + let pending = () => typeof state.pending === "function" ? state.pending() : state.pending; + let storage: Reloading["storage"] = options.storage === "none" ? undefined : { + getItem: key => { + if (options.storage === "refusing") throw new Error("storage is blocked"); + return values.get(key) ?? null; + }, + setItem: (key, value) => { + if (options.storage === "refusing") throw new Error("storage is blocked"); + values.set(key, value); + }, + }; + let reloading: Reloading = { + pending, + newer: async () => { + await Promise.resolve(); + return state.newer; + }, + reload: () => { + reloads++; + }, + now: () => clock, + storage, + }; + return { + fail: () => recoverFromPreloadError(reloading), + state, + wait: (ms: number) => { + clock += ms; + }, + reloads: () => reloads, + }; +} + +describe("recoverFromPreloadError", () => { + it("reloads for a chunk that is gone when the server answers with a new build and the page holds nothing", async () => { + let tab = page(); + expect(await tab.fail()).toBe(true); + expect(tab.reloads()).toBe(1); + }); + + it("does not reload while the page holds unsaved work", async () => { + let tab = page({ pending: true }); + expect(await tab.fail()).toBe(false); + expect(tab.reloads()).toBe(0); + }); + + it("does not reload while a question typed into the document has not been started", async () => { + let drafts = new ResearchDraftStore("investigation"); + drafts.open({ top: 1, right: 2, bottom: 3, left: 4, width: 5, height: 6 }); + let tab = page({ pending: hasPendingWork }); + drafts.change("Which index is faster?"); + expect(await tab.fail()).toBe(false); + expect(tab.reloads()).toBe(0); + + // Nothing typed, or typed and put away, is nothing to lose. + drafts.change(""); + expect(drafts.unsaved).toBe(false); + }); + + it("does not reload unless the server answers with a new build, which a chunk failing with the server away, or for no deploy at all, does not give", async () => { + let tab = page({ newer: false }); + expect(await tab.fail()).toBe(false); + expect(tab.reloads()).toBe(0); + }); + + it("does not reload for work begun while the server was being asked", async () => { + let tab = page(); + let failing = tab.fail(); + tab.state.pending = true; + expect(await failing).toBe(false); + expect(tab.reloads()).toBe(0); + }); + + it("does not reload again for the next failure straight after one", async () => { + let tab = page(); + expect(await tab.fail()).toBe(true); + tab.wait(RELOAD_GAP_MS - 1); + expect(await tab.fail()).toBe(false); + expect(tab.reloads()).toBe(1); + }); + + it("reloads again once the gap has passed, as for a second deploy in one session", async () => { + let tab = page(); + await tab.fail(); + tab.wait(RELOAD_GAP_MS); + expect(await tab.fail()).toBe(true); + expect(tab.reloads()).toBe(2); + }); + + it("reloads once for several chunks failing together", async () => { + let tab = page(); + let results = await Promise.all([tab.fail(), tab.fail(), tab.fail()]); + expect(results.filter(Boolean)).toHaveLength(1); + expect(tab.reloads()).toBe(1); + }); + + it("makes no reload it could not remember, so that it cannot loop", async () => { + for (let storage of ["none", "refusing"] as const) { + let tab = page({ storage }); + expect(await tab.fail()).toBe(false); + expect(tab.reloads()).toBe(0); + } + }); +}); + +describe("entryOf", () => { + it("names the build by the entry script its page loads", () => { + let html = ` + + `; + expect(entryOf(html)).toBe("/assets/index-CS81MqW4.js"); + expect(entryOf('')).toBe( + "/assets/index-CS81MqW4.js", + ); + }); + + it("is undefined for a page that loads none, such as an error page", () => { + expect(entryOf("")).toBeUndefined(); + expect(entryOf("Bad gateway")).toBeUndefined(); + expect(entryOf('')) + .toBeUndefined(); + }); +}); + +describe("isNewerBuild", () => { + it("is the server's page being of another build than the one running", () => { + expect(isNewerBuild("/assets/index-AAAA.js", "/assets/index-BBBB.js")).toBe(true); + expect(isNewerBuild("/assets/index-AAAA.js", "/assets/index-AAAA.js")).toBe(false); + }); + + it("is never so while either build is unknown", () => { + expect(isNewerBuild(undefined, "/assets/index-BBBB.js")).toBe(false); + expect(isNewerBuild("/assets/index-AAAA.js", undefined)).toBe(false); + }); +}); diff --git a/apps/web/src/preload-recovery.ts b/apps/web/src/preload-recovery.ts new file mode 100644 index 000000000..2b0716268 --- /dev/null +++ b/apps/web/src/preload-recovery.ts @@ -0,0 +1,97 @@ +/** + * What a page does when a chunk it needs cannot be loaded. + * + * A tab opened before a deploy asks for the chunks of its own build, and the server no longer has + * them. Reloading fetches the new build, but only if the server answers with one, and only if the + * page holds nothing a reload would throw away. A chunk also fails to load while the server is + * restarting, and a reload then lands on the browser's error page; and it fails for reasons that + * have nothing to do with a deploy, which the page's own recovery says what to do about, and a + * reload of the same build would not cure. + */ + +import { hasPendingWork } from "@chopin/protocol/pending-work"; + +const RELOADED_AT = "chopin:preload-reloaded-at"; + +/** + * A reload that did not cure the failure is not repeated within this long, so a build that is + * really broken cannot reload the page for ever. + */ +export const RELOAD_GAP_MS = 10_000; + +/** + * The entry script a page's HTML loads, which names its build: any file of the build that changes + * changes the entry's hash. Undefined when the HTML loads none. + */ +export function entryOf(html: string): string | undefined { + for (let [tag] of html.matchAll(/]*>/g)) { + if (/\btype="module"/.test(tag)) { + let source = /\bsrc="([^"]+)"/.exec(tag); + if (source) return source[1]; + } + } + return undefined; +} + +/** Whether the server's page is of another build than the one running, both entries being known. */ +export function isNewerBuild(running: string | undefined, served: string | undefined): boolean { + return !!running && !!served && running !== served; +} + +export type Reloading = { + pending: () => boolean; + /** Whether the server answers with a page of another build than this one, which a reload would show. */ + newer: () => Promise; + reload: () => void; + now: () => number; + /** Remembers when the page last reloaded; without it no reload is made. */ + storage: Pick | undefined; +}; + +/** + * Reload for a chunk that failed to load if nothing would be lost, the server answers with another + * build and the last reload was not just now. Returns whether it did. The failure itself is left to + * be thrown, as it was, to whatever was loading the chunk. + */ +export async function recoverFromPreloadError(reloading: Reloading): Promise { + if (reloading.pending() || !await reloading.newer()) return false; + // Work may have begun while the server was asked. + if (reloading.pending()) return false; + try { + let now = reloading.now(); + if ( + !reloading.storage || now - Number(reloading.storage.getItem(RELOADED_AT)) < RELOAD_GAP_MS + ) { + return false; + } + reloading.storage.setItem(RELOADED_AT, String(now)); + } catch { + return false; + } + reloading.reload(); + return true; +} + +/** Listen for Vite's `vite:preloadError`, which it raises when a dynamic import fails to load. */ +export function watchPreloadErrors(): void { + let storage: Storage | undefined; + try { + storage = sessionStorage; + } catch { + // Storage the browser refuses: no reload can be remembered, so none is made. + } + let running = document.querySelector('script[type="module"][src]')?.getAttribute("src") + ?? undefined; + addEventListener("vite:preloadError", () => { + void recoverFromPreloadError({ + pending: hasPendingWork, + newer: () => + fetch(location.href, { cache: "no-store" }) + .then(response => response.ok ? response.text() : "") + .then(html => isNewerBuild(running, entryOf(html)), () => false), + reload: () => location.reload(), + now: Date.now, + storage, + }); + }); +} diff --git a/apps/web/src/question-page.tsx b/apps/web/src/question-page.tsx index 0855aaba7..2c671a08f 100644 --- a/apps/web/src/question-page.tsx +++ b/apps/web/src/question-page.tsx @@ -15,6 +15,7 @@ import { personName, QuestionnaireCard, runningFindings, + usePendingWork, useQuestionPageDraft, waitingNames, waitingOn, @@ -421,6 +422,7 @@ function Take( let [stance, setStance] = useState(); let [busy, setBusy] = useState(false); let [error, setError] = useState(); + usePendingWork(text.trim() !== "" || busy); let mine = take.ref.toLowerCase() === root.self?.toLowerCase(); let reply = async () => { if (!root.wire || !text.trim()) return; diff --git a/apps/web/src/question-why.tsx b/apps/web/src/question-why.tsx index 2ee54bc31..d1cbf29d2 100644 --- a/apps/web/src/question-why.tsx +++ b/apps/web/src/question-why.tsx @@ -5,6 +5,8 @@ import { useLayoutEffect, useRef, useState } from "react"; +import { usePendingWork } from "@chopin/editor"; + import type { Transport } from "@chopin/editor"; import type { Question } from "@chopin/protocol"; @@ -25,6 +27,9 @@ export function QuestionWhy( let field = useRef(null); let saved = useRef(why ?? ""); let [error, setError] = useState(); + // What is typed and not yet saved, which a reload would lose. + let [dirty, setDirty] = useState(false); + usePendingWork(dirty); let editable = canEdit && !!wire; // Another writer's change, or Chopin's draft arriving, replaces the line unless it is being edited. @@ -43,6 +48,7 @@ export function QuestionWhy( if (!element) return; let text = tidy(element.textContent ?? ""); element.textContent = text; + setDirty(false); if (text === saved.current) return; let previous = saved.current; saved.current = text; @@ -75,6 +81,7 @@ export function QuestionWhy( onInput={event => { // A emptied field can keep a stray line break, which would hide the placeholder. if (!event.currentTarget.textContent) event.currentTarget.replaceChildren(); + setDirty(tidy(event.currentTarget.textContent ?? "") !== saved.current); }} onKeyDown={event => { if (event.key === "Enter" && !event.nativeEvent.isComposing) { @@ -83,6 +90,7 @@ export function QuestionWhy( } if (event.key === "Escape") { if (field.current) field.current.textContent = saved.current; + setDirty(false); field.current?.blur(); } }} diff --git a/apps/web/src/room-workspace.tsx b/apps/web/src/room-workspace.tsx index 63992b70f..9fdc0c9f4 100644 --- a/apps/web/src/room-workspace.tsx +++ b/apps/web/src/room-workspace.tsx @@ -563,7 +563,7 @@ export function RoomWorkspace( return () => removeEventListener("title-edit", listen); }, [room]); // Controls dim only once a loss outlasts a blip; actions still read `status`. - let treatAsConnected = useConnectionNotice(status !== "connected") === "none"; + let treatAsConnected = useConnectionNotice(status !== "connected", wire?.restarting) === "none"; // A lost connection is said once, at the top of the page: decided here, above wherever the // notice is drawn, so a person switching between Chat and the document neither waits out the // grace period again nor hears of it twice. @@ -1682,7 +1682,11 @@ export function RoomWorkspace( )} ); - let machinePicker = status === "connected" ? machineSlot : undefined; + // It goes with a lost connection, but not while a restart is quiet: the page says nothing then, + // and a button that came and went would be something. + let machinePicker = status === "connected" || (!!wire?.restarting && treatAsConnected) + ? machineSlot + : undefined; let titleProps: TitleProps = { archivedAt: workspaceArchivedAt, canManage: effectiveCanManage, diff --git a/apps/web/src/wire-loss.test.ts b/apps/web/src/wire-loss.test.ts new file mode 100644 index 000000000..f44a67de1 --- /dev/null +++ b/apps/web/src/wire-loss.test.ts @@ -0,0 +1,83 @@ +import { describe, expect, it } from "bun:test"; + +import { RESTART_QUIET_MS } from "@chopin/protocol/restart"; + +import { + lossOf, + quietLeft, + refusalOf, + RESTART_JITTER_MS, + restartDelay, + restartRetryDelay, +} from "./wire-loss"; + +describe("lossOf", () => { + it("reads 1012 as a restart, which reconnects quietly", () => { + expect(lossOf(1012)).toBe("restart"); + }); + + it("reads 4404 as a deleted document, which nothing can reconnect to", () => { + expect(lossOf(4404)).toBe("deleted"); + }); + + it("reads every other close as an ordinary loss, 1001 and 1006 included", () => { + for (let code of [1000, 1001, 1006, 1011, 4401]) expect(lossOf(code)).toBe("lost"); + }); +}); + +describe("restartDelay", () => { + it("is spread over 0 to 1000 ms and never reaches the end", () => { + expect(restartDelay(0)).toBe(0); + expect(restartDelay(0.5)).toBe(500); + expect(restartDelay(0.999999)).toBeLessThan(RESTART_JITTER_MS); + expect(RESTART_JITTER_MS).toBe(1000); + }); +}); + +describe("restartRetryDelay", () => { + it("is about a second, spread over 250 to 1000 ms", () => { + expect(restartRetryDelay(0)).toBe(250); + expect(restartRetryDelay(0.5)).toBe(625); + expect(restartRetryDelay(0.999999)).toBeLessThan(1000); + }); +}); + +describe("quietLeft", () => { + it("counts a restart's quiet period down from the moment the connection was lost", () => { + expect(RESTART_QUIET_MS).toBe(5000); + expect(quietLeft(true, 1_000, 1_000)).toBe(RESTART_QUIET_MS); + expect(quietLeft(true, 1_000, 3_500)).toBe(2_500); + expect(quietLeft(true, 1_000, 1_000 + RESTART_QUIET_MS - 1)).toBe(1); + }); + + it("is over once the period has passed, when the ordinary rules and the outage take over", () => { + expect(quietLeft(true, 1_000, 1_000 + RESTART_QUIET_MS)).toBe(0); + expect(quietLeft(true, 1_000, 1_000 + RESTART_QUIET_MS * 3)).toBe(0); + }); + + it("is never there for a loss the server did not announce, or a connection that is not lost", () => { + expect(quietLeft(false, 1_000, 1_000)).toBe(0); + expect(quietLeft(true, undefined, 1_000)).toBe(0); + }); +}); + +describe("refusalOf", () => { + it("takes a 4xx as a refusal only when Chopin marked it", () => { + for (let status of [400, 403, 404, 429]) { + expect(refusalOf(status, true)).toBe("denied"); + expect(refusalOf(status, false)).toBeUndefined(); + } + }); + + it("takes a 401 as a request to sign in, marked or not", () => { + expect(refusalOf(401, true)).toBe("authentication"); + expect(refusalOf(401, false)).toBe("authentication"); + }); + + it("never takes anything but a 4xx for a refusal, so a proxy's 502 or an open door retries", () => { + for (let status of [200, 204, 302, 500, 502, 503, 504]) { + expect(refusalOf(status, true)).toBeUndefined(); + expect(refusalOf(status, false)).toBeUndefined(); + } + }); +}); diff --git a/apps/web/src/wire-loss.ts b/apps/web/src/wire-loss.ts new file mode 100644 index 000000000..129f246e1 --- /dev/null +++ b/apps/web/src/wire-loss.ts @@ -0,0 +1,68 @@ +/** + * What a lost socket and a refused probe mean for reconnecting, as plain decisions. + * + * Kept apart from `Wire` so they can be tested without a socket: the rules are the point, and + * a deploy that went wrong because one of them was misread is not a failure to reproduce in a + * browser. + */ + +import { RESTART_QUIET_MS, SERVICE_RESTART } from "@chopin/protocol/restart"; + +/** The document was deleted, the server is restarting, or the connection was simply lost. */ +export type Loss = "deleted" | "restart" | "lost"; + +/** + * How the socket's close code reads. 1012 is the server saying it is restarting, so the client + * comes back quietly; 4404 is a document that no longer exists, which no retry can reach. + */ +export function lossOf(code: number): Loss { + if (code === 4404) return "deleted"; + return code === SERVICE_RESTART ? "restart" : "lost"; +} + +/** + * Spread over this long so every tab of a restarted server does not return in the same instant. + * The zero-downtime upgrades design (Stage 1, `apps/web` `wire.ts`) says to add no restart jitter, + * as its proxy holds the first attempt until the new process is ready. It is added here on purpose: + * a restart with no such proxy would have every tab return in the same instant. + */ +export const RESTART_JITTER_MS = 1000; + +/** How long a restart waits before its first attempt, for a `random` in [0, 1). */ +export function restartDelay(random: number): number { + return Math.floor(random * RESTART_JITTER_MS); +} + +/** + * How long a client that has not got back in waits for its next attempt while a restart is + * quiet, for a `random` in [0, 1): about a second, spread, instead of the ladder's wait, which + * grows to several seconds and would find a server that is already back long after it was. + */ +export function restartRetryDelay(random: number): number { + return 250 + Math.floor(random * 750); +} + +/** + * How much of a restart's quiet period is left, `now` and `lostAt` being `Date.now()` readings. + * None for a loss that was not announced as a restart, none for a connection that is not lost, and + * none once the period has passed, which is when the ordinary rules take over: the backoff, and + * the outage the connection reports. + */ +export function quietLeft(restarting: boolean, lostAt: number | undefined, now: number): number { + if (!restarting || lostAt === undefined) return 0; + return Math.max(0, lostAt + RESTART_QUIET_MS - now); +} + +export type Denial = "authentication" | "denied"; + +/** + * Whether the probe's answer is a refusal, and which. + * + * A 401 always asks for a sign-in, even from a server that predates the marker. Any other 4xx + * is Chopin refusing only if it says so: a proxy with no upstream answers 4xx too, and reading + * that as a refusal would stop every open tab from ever reconnecting. + */ +export function refusalOf(status: number, marked: boolean): Denial | undefined { + if (status === 401) return "authentication"; + return status >= 400 && status < 500 && marked ? "denied" : undefined; +} diff --git a/apps/web/src/wire-quiet.test.ts b/apps/web/src/wire-quiet.test.ts new file mode 100644 index 000000000..851f2082f --- /dev/null +++ b/apps/web/src/wire-quiet.test.ts @@ -0,0 +1,197 @@ +/** + * What a restart's quiet period keeps from `onOutage`, on fake sockets and fake time. + * + * The period is five seconds long and what matters is where inside it each thing lands, so it is + * read from a clock the test moves rather than waited for: a real server would put a knock + * against the end of the period wherever its timers happened to fall. + */ + +import { RESTART_QUIET_MS, SERVICE_RESTART } from "@chopin/protocol/restart"; +import { afterEach, beforeEach, describe, expect, it, jest } from "bun:test"; + +import { Wire } from "./wire"; + +import type { Outage } from "./wire"; + +/** The part of a browser socket that `Wire` uses, ended and opened by the test. */ +class FakeSocket extends EventTarget { + static readonly CONNECTING = 0; + static readonly OPEN = 1; + static readonly CLOSING = 2; + static readonly CLOSED = 3; + static made: FakeSocket[] = []; + + readyState = FakeSocket.CONNECTING; + + constructor(readonly url: string) { + super(); + FakeSocket.made.push(this); + } + + send(): void {} + + close(): void { + this.readyState = FakeSocket.CLOSED; + } + + open(): void { + this.readyState = FakeSocket.OPEN; + this.dispatchEvent(new Event("open")); + } + + end(code: number): void { + this.readyState = FakeSocket.CLOSED; + this.dispatchEvent(Object.assign(new Event("close"), { code })); + } +} + +let real = { + WebSocket: globalThis.WebSocket, + fetch: globalThis.fetch, + location: globalThis.location, + random: Math.random, +}; +let wires: Wire[] = []; + +beforeEach(() => { + FakeSocket.made = []; + globalThis.WebSocket = FakeSocket as unknown as typeof WebSocket; + // Every probe says the door is not refused, which leaves the retry loop to go on. + globalThis.fetch = (() => + Promise.resolve(new Response(null, { status: 204 }))) as unknown as typeof fetch; + globalThis.location = { href: "http://chopin.test/", protocol: "http:" } as Location; + // The least jitter, and the shortest knock, so the timeline below is exact. + Math.random = () => 0; + jest.useFakeTimers(); +}); + +afterEach(() => { + for (let wire of wires) wire.dispose(); + wires = []; + jest.useRealTimers(); + globalThis.WebSocket = real.WebSocket; + globalThis.fetch = real.fetch; + globalThis.location = real.location; + Math.random = real.random; +}); + +/** Let the probe's promise settle, which is all a fake `fetch` needs. */ +async function settle(): Promise { + for (let turn = 0; turn < 20; turn++) await Promise.resolve(); +} + +/** Move the clock on, letting each probe answer as the timers it starts come due. */ +async function advance(ms: number): Promise { + let step = 50; + for (let moved = 0; moved < ms; moved += step) { + jest.advanceTimersByTime(Math.min(step, ms - moved)); + await settle(); + } +} + +function latest(): FakeSocket { + return FakeSocket.made.at(-1)!; +} + +/** A wire that has connected and then been told, with 1012, that the server is restarting. */ +async function restarted(): Promise<{ wire: Wire; outages: Array }> { + let outages: Array = []; + let wire = new Wire({ + channelId: "019c1234-1234-4123-8123-123456789abc", + onOutage: outage => outages.push(outage), + }); + wires.push(wire); + latest().open(); + expect(wire.status).toBe("connected"); + latest().end(SERVICE_RESTART); + await settle(); + expect(wire.restarting).toBe(true); + return { wire, outages }; +} + +describe("a restart's quiet period", () => { + it("raises no outage while it lasts, however many attempts fail in it", async () => { + let { outages } = await restarted(); + + // Each knock the server refuses, from the first attempt to the end of the quiet period. + for (let waited = 0; waited < RESTART_QUIET_MS - 1_000; waited += 100) { + await advance(100); + if (latest().readyState === FakeSocket.CONNECTING) latest().end(1006); + await settle(); + } + expect(FakeSocket.made.length).toBeGreaterThan(4); + expect(outages).toEqual([]); + }); + + it("says nothing at all, not even that an outage ended, when the server is back inside it", async () => { + let { wire, outages } = await restarted(); + + await advance(2_000); + latest().end(1006); + await advance(1_000); + latest().open(); + + expect(wire.status).toBe("connected"); + expect(wire.restarting).toBe(false); + await advance(RESTART_QUIET_MS * 2); + expect(outages).toEqual([]); + }); + + it("raises the outage when it has passed, though no attempt has failed by then", async () => { + let { outages } = await restarted(); + + // The first attempt is out, neither opened nor refused: nothing but time can end the quiet. + await advance(RESTART_QUIET_MS - 1); + expect(latest().readyState).toBe(FakeSocket.CONNECTING); + expect(outages).toEqual([]); + + await advance(1); + expect(outages).toEqual([{ failures: 0 }]); + }); + + it("does not count the attempts that failed inside it, and counts those after", async () => { + let { outages } = await restarted(); + + await advance(2_000); + latest().end(1006); + await advance(RESTART_QUIET_MS - 2_000); + // The attempt that followed the failure is still out, and the outage is raised without it. + expect(outages).toEqual([{ failures: 0 }]); + + latest().end(1006); + expect(outages).toEqual([{ failures: 0 }, { failures: 1 }]); + }); + + it("goes on as for any other loss once it has passed: the outage ends when the connection opens", async () => { + let { wire, outages } = await restarted(); + + await advance(RESTART_QUIET_MS); + expect(outages).toEqual([{ failures: 0 }]); + + latest().open(); + expect(wire.status).toBe("connected"); + expect(outages).toEqual([{ failures: 0 }, undefined]); + }); + + it("reports a loss that was not a restart at once, as before", async () => { + let outages: Array = []; + let wire = new Wire({ + channelId: "019c1234-1234-4123-8123-123456789abc", + onOutage: outage => outages.push(outage), + }); + wires.push(wire); + latest().open(); + latest().end(1006); + + expect(wire.restarting).toBe(false); + expect(outages).toEqual([{ failures: 0 }]); + }); + + it("is not raised for a wire that was disposed during it", async () => { + let { wire, outages } = await restarted(); + + wire.dispose(); + await advance(RESTART_QUIET_MS * 2); + expect(outages).toEqual([]); + }); +}); diff --git a/apps/web/src/wire.test.ts b/apps/web/src/wire.test.ts index 076c63eea..dd6440f76 100644 --- a/apps/web/src/wire.test.ts +++ b/apps/web/src/wire.test.ts @@ -7,9 +7,10 @@ * typing, so it being wrong is worse than it being absent. */ +import { REFUSAL_HEADER, RESTART_QUIET_MS } from "@chopin/protocol/restart"; import { afterEach, describe, expect, it } from "bun:test"; -import { Wire } from "./wire"; +import { Wire, WireLost } from "./wire"; import type { Server, ServerWebSocket } from "bun"; import type { Status } from "./wire"; @@ -30,7 +31,7 @@ function endpoint(accept: boolean): number { port: 0, hostname: "127.0.0.1", fetch(req, self) { - if (!accept) return new Response("nope", { status: 403 }); + if (!accept) return new Response("nope", { status: 403, headers: { [REFUSAL_HEADER]: "1" } }); return self.upgrade(req) ? undefined : new Response("no", { status: 400 }); }, websocket: { message() {} }, @@ -49,6 +50,17 @@ function unavailable(): number { return server.port!; } +/** What a proxy with no upstream answers: a 4xx that Chopin did not send. */ +function proxied(status: number): number { + let server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + fetch: () => new Response("no such upstream", { status }), + }); + servers.push(server); + return server.port!; +} + function unauthorized(): number { let server = Bun.serve({ port: 0, @@ -59,14 +71,22 @@ function unauthorized(): number { return server.port!; } -function restartable(): { port: number; deny: () => void; drop: () => void } { +function restartable(): { + port: number; + deny: () => void; + drop: () => void; + restart: () => void; + probes: () => number; +} { let denied = false; + let probes = 0; let sockets = new Set>(); let server = Bun.serve({ port: 0, hostname: "127.0.0.1", fetch(req, self) { if (req.headers.get("x-chopin-socket-probe") === "1") { + probes++; return new Response(null, { status: denied ? 401 : 204 }); } if (denied) return new Response("authentication required", { status: 401 }); @@ -88,6 +108,10 @@ function restartable(): { port: number; deny: () => void; drop: () => void } { drop: () => { for (let socket of sockets) socket.terminate(); }, + restart: () => { + for (let socket of sockets) socket.close(1012, "server restarting"); + }, + probes: () => probes, deny: () => { denied = true; for (let socket of sockets) socket.terminate(); @@ -139,8 +163,8 @@ function connect( return wire; } -async function until(check: () => boolean, label: string): Promise { - for (let i = 0; i < 200; i++) { +async function until(check: () => boolean, label: string, ms = 2_000): Promise { + for (let i = 0; i < ms / 10; i++) { if (check()) return; await Bun.sleep(10); } @@ -183,6 +207,20 @@ describe("status", () => { expect(seen).toContain("denied"); }); + it("retries a 4xx that did not come from Chopin, where it used to stop for good", async () => { + for (let status of [403, 404, 429]) { + let seen: Status[] = []; + let wire = connect(proxied(status), seen); + await until(() => wire.status === "reconnecting", `reconnecting after ${status}`); + await Bun.sleep(300); + expect(seen).not.toContain("denied"); + // Between attempts it reads reconnecting and during one connecting, so which it is + // depends on where the retry's wait has got to; what matters is that it is still trying. + expect(["connecting", "reconnecting"]).toContain(wire.status); + wire.dispose(); + } + }); + it("retries a temporarily unavailable admission", async () => { let seen: Status[] = []; let wire = connect(unavailable(), seen); @@ -227,6 +265,207 @@ describe("status", () => { expect(wire.status).toBe("connected"); }); + describe("a server that says it is restarting", () => { + it("is reconnected to within a second, without asking first whether the door is open", async () => { + let service = restartable(); + let seen: Status[] = []; + let wire = connect(service.port, seen); + await until(() => wire.status === "connected", "connected"); + expect(wire.restarting).toBe(false); + + let random = Math.random; + Math.random = () => 0.5; + try { + let lost = Date.now(); + service.restart(); + await until(() => wire.restarting, "marked as a restart"); + expect(wire.connected).toBe(false); + expect(wire.status).toBe("reconnecting"); + await until(() => seen.filter(status => status === "connected").length === 2, "back"); + let waited = Date.now() - lost; + expect(waited).toBeGreaterThanOrEqual(450); + expect(waited).toBeLessThan(1500); + } finally { + Math.random = random; + } + expect(wire.restarting).toBe(false); + expect(service.probes()).toBe(0); + }); + + it("spreads its first attempt over a second, so tabs do not return together", async () => { + let service = restartable(); + let seen: Status[] = []; + let wire = connect(service.port, seen); + await until(() => wire.status === "connected", "connected"); + + let random = Math.random; + Math.random = () => 0.999; + try { + let lost = Date.now(); + service.restart(); + await until(() => seen.filter(status => status === "connected").length === 2, "back"); + expect(Date.now() - lost).toBeGreaterThanOrEqual(950); + } finally { + Math.random = random; + } + }); + + it( + "keeps knocking about once a second while the restart is quiet, so a server that is back is found at once", + async () => { + let attempts: number[] = []; + let refusing = false; + let sockets = new Set>(); + let server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + fetch(request, self) { + if (request.headers.get("x-chopin-socket-probe") === "1") { + return new Response("try again", { status: 503 }); + } + attempts.push(Date.now()); + if (refusing) return new Response("try again", { status: 503 }); + return self.upgrade(request) ? undefined : new Response("no", { status: 400 }); + }, + websocket: { + open: socket => { + sockets.add(socket); + }, + message() {}, + }, + }); + servers.push(server); + let seen: Status[] = []; + let wire = connect(server.port!, seen); + await until(() => wire.status === "connected", "connected"); + attempts.length = 0; + + // The widest jitter, so that only the quiet knocking, not the ladder, can pass for it. + let random = Math.random; + Math.random = () => 0.999; + try { + refusing = true; + let lost = Date.now(); + for (let socket of sockets) socket.close(1012, "server restarting"); + await until(() => attempts.length >= 4, "four attempts in the quiet", 5_000); + expect(Date.now() - lost).toBeLessThan(4_500); + expect(wire.restarting).toBe(true); + expect(seen).not.toContain("denied"); + // The ladder alone, from its second step, would be 1.5 s and then 3 s apart. + for (let index = 1; index < attempts.length; index++) { + expect(attempts[index]! - attempts[index - 1]!).toBeLessThan(1_300); + } + + refusing = false; + let back = Date.now(); + await until( + () => seen.filter(status => status === "connected").length === 2, + "back", + 3_000, + ); + expect(Date.now() - back).toBeLessThan(1_500); + expect(wire.restarting).toBe(false); + } finally { + Math.random = random; + } + }, + 15_000, + ); + + it("goes on the ordinary backoff, from where it was, once the quiet has passed", async () => { + // The ladder's steps are timed on the real clock, but the quiet period is read from + // `Date.now`, which is moved past it inside the third probe's answer, before the wire + // reads it. So where a knock happens to land against the five seconds never matters. + let attempts: number[] = []; + let sockets = new Set>(); + let refusing = false; + let probes = 0; + let skew = 0; + let clock = Date.now; + let server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + fetch(request, self) { + if (request.headers.get("x-chopin-socket-probe") === "1") { + if (refusing && ++probes === 3) skew = RESTART_QUIET_MS + 1_000; + return new Response("try again", { status: 503 }); + } + if (refusing) { + attempts.push(performance.now()); + return new Response("try again", { status: 503 }); + } + return self.upgrade(request) ? undefined : new Response("no", { status: 400 }); + }, + websocket: { + open: socket => { + sockets.add(socket); + }, + message() {}, + }, + }); + servers.push(server); + let wire = connect(server.port!, []); + await until(() => wire.status === "connected", "connected"); + + let random = Math.random; + Math.random = () => 0.999; + Date.now = () => clock() + skew; + try { + refusing = true; + for (let socket of sockets) socket.close(1012, "server restarting"); + await until(() => attempts.length >= 5, "five attempts", 15_000); + // The first attempt and two knocks fall in the quiet. Then the ladder goes on from its + // second step: 1.5 s, and 3 s after that. Counting the knocks towards it would make the + // first wait 6 s, and starting it from the beginning would make it 0.75 s. + let gaps = [attempts[3]! - attempts[2]!, attempts[4]! - attempts[3]!]; + expect(gaps[0]).toBeGreaterThan(1_300); + expect(gaps[0]).toBeLessThan(3_000); + expect(gaps[1]).toBeGreaterThan(2_700); + expect(gaps[1]).toBeLessThan(5_000); + } finally { + Date.now = clock; + Math.random = random; + } + }, 20_000); + + it("tells a request that was in flight that a restart cut it off, so it can be made again quietly", async () => { + let service = restartable(); + let wire = connect(service.port, []); + await until(() => wire.status === "connected", "connected"); + + let outcome = wire.ask("chat:send", { text: "hello" }).then(() => undefined, error => error); + service.restart(); + let error = await outcome; + expect(error).toBeInstanceOf(WireLost); + expect((error as WireLost).restarting).toBe(true); + expect(wire.restarting).toBe(true); + }); + + it("does not call a request lost to a plain drop a restart", async () => { + let service = restartable(); + let wire = connect(service.port, []); + await until(() => wire.status === "connected", "connected"); + + let outcome = wire.ask("chat:send", { text: "hello" }).then(() => undefined, error => error); + service.drop(); + let error = await outcome; + expect(error).toBeInstanceOf(WireLost); + expect((error as WireLost).restarting).toBe(false); + expect((error as WireLost).message).toBe("connection lost"); + }); + + it("is not a restart when the connection is merely dropped", async () => { + let service = restartable(); + let seen: Status[] = []; + let wire = connect(service.port, seen); + await until(() => wire.status === "connected", "connected"); + service.drop(); + await until(() => wire.status === "reconnecting", "reconnecting"); + expect(wire.restarting).toBe(false); + await until(() => wire.status === "connected", "back"); + }); + }); + it("reconnects immediately without leaving a second backoff attempt", async () => { let service = restartable(); let seen: Status[] = []; diff --git a/apps/web/src/wire.ts b/apps/web/src/wire.ts index 1dc9c012c..a47173674 100644 --- a/apps/web/src/wire.ts +++ b/apps/web/src/wire.ts @@ -10,7 +10,10 @@ * same build work on localhost, over a LAN address, and through a tunnel. */ +import { REFUSAL_HEADER } from "@chopin/protocol/restart"; + import { activeActor } from "./active-actor"; +import { lossOf, quietLeft, refusalOf, restartDelay, restartRetryDelay } from "./wire-loss"; import type { Session } from "@chopin/protocol"; @@ -35,6 +38,21 @@ export class WireRefusal extends Error { } } +/** + * A request that was waiting for its answer when its connection went. `restarting` says the server + * closed it because it is restarting, which ends within moments: a caller whose request can be made + * again safely may keep its work and do so, without saying anything. + */ +export class WireLost extends Error { + readonly restarting: boolean; + + constructor(message: string, restarting = false) { + super(message); + this.name = "WireLost"; + this.restarting = restarting; + } +} + type Frame = { kind: string; ts: number; rid?: string; sender?: string }; type Listener = (frame: never) => void; @@ -55,6 +73,9 @@ export type WireOptions = WireTarget & { * Told when an outage begins, each time an attempt in it fails, and, with `undefined`, when the * connection opens again. Automatic attempts and a person's both count, and neither a manual * reconnect nor the network coming back resets the count: only a connection that opens does. + * A loss the server announced as a restart (close code 1012) raises none for `RESTART_QUIET_MS`, + * and the attempts that fail meanwhile are not counted: a restart that is over by then is never + * reported, and one that is not is reported then, as any other loss. */ onOutage?: (outage: Outage | undefined) => void; onAuthenticationRequired?: () => void; @@ -111,6 +132,10 @@ export class Wire { #timer: ReturnType | undefined; /** Whether the probe after a loss is still awaiting its answer. */ #probing = false; + /** Whether `onOutage` has been told of this outage, so that it is told when it ends. */ + #outageTold = false; + /** Raises the outage of a restart that outlasts its quiet period. */ + #quietTimer: ReturnType | undefined; /** * Undefined until the first transition, so the first one always announces. * @@ -123,6 +148,8 @@ export class Wire { */ #status: Status | undefined; #everConnected = false; + /** Whether the current outage began with the server saying it is restarting. */ + #restarting = false; #disposed = false; #terminal = false; #connectionGeneration = 0; @@ -171,6 +198,17 @@ export class Wire { return this.#status ?? "connecting"; } + /** + * Whether the connection was lost because the server is restarting, until it is back. + * + * A restart is expected to end on its own within moments, so what is shown about it may + * wait longer than for any other loss. What has to act on the connection must not: read + * `status` and `connected` for that. + */ + get restarting(): boolean { + return this.#restarting; + } + /** * Whether a request can be carried right now. * @@ -222,11 +260,15 @@ export class Wire { socket.addEventListener("open", () => { if (this.#socket !== socket) return; opened = true; - let recovered = this.#lostAt !== undefined; + let recovered = this.#outageTold; + this.#outageTold = false; this.#everConnected = true; this.#attempts = 0; this.#failures = 0; this.#lostAt = undefined; + this.#restarting = false; + clearTimeout(this.#quietTimer); + this.#quietTimer = undefined; if (recovered) this.#options.onOutage?.(undefined); this.#set("connected"); }); @@ -241,20 +283,69 @@ export class Wire { // A socket superseded by a reconnect must not drive status or retries. if (this.#socket !== socket) return; this.#socket = undefined; - if (event.code === 4404) { + let loss = lossOf(event.code); + if (loss === "deleted") { this.#deleted(); return; } this.#lostAt ??= Date.now(); - if (!opened) this.#failures++; - this.#options.onOutage?.({ failures: this.#failures }); - this.#abandon("connection lost"); + // A restart is expected to end on its own within moments: until its quiet has passed, an + // attempt that fails is not counted and no outage is raised for it (`#quietLeft`). + if (loss === "restart") this.#restarting = true; + if (!opened && this.#quietLeft() <= 0) this.#failures++; + this.#raiseOutage(); + this.#abandon("connection lost", loss === "restart"); // Said now, not once the probe has answered: until then nothing else says it is lost. this.#set("reconnecting"); - void this.#retry(); + if (loss === "restart") this.#restart(); + else void this.#retry(); }); } + #quietLeft(): number { + return quietLeft(this.#restarting, this.#lostAt, Date.now()); + } + + /** + * Tell `onOutage` the connection is lost, with how many attempts have failed. + * + * Not while a restart is quiet: the server said it is coming back, and nothing is to show for + * that yet, which is what the outage is for. If the restart outlasts its quiet period the + * outage is raised then, whether or not an attempt has failed by that time, and goes on as + * for any other loss. + */ + #raiseOutage(): void { + let quiet = this.#quietLeft(); + if (quiet <= 0) { + this.#outageTold = true; + this.#options.onOutage?.({ failures: this.#failures }); + return; + } + if (this.#quietTimer !== undefined) return; + this.#quietTimer = setTimeout(() => { + this.#quietTimer = undefined; + if (this.#lostAt === undefined || this.#disposed || this.#terminal) return; + this.#outageTold = true; + this.#options.onOutage?.({ failures: this.#failures }); + }, quiet); + } + + /** + * Come back after the server said it is restarting. + * + * Not asked first whether the door is open: it is about to be shut, and the first attempt + * is what finds out. It waits a jittered moment so that every tab of the server does not + * return in the same instant. An attempt that fails is followed by another soon while the + * restart is quiet, and after that the ordinary ladder goes on from where it was. + */ + #restart(): void { + this.#attempts++; + this.#timer = setTimeout(() => { + this.#timer = undefined; + this.#connect(); + }, restartDelay(Math.random())); + } + /** * Work out why a connection attempt was refused. * @@ -269,11 +360,9 @@ export class Wire { headers: { "x-chopin-socket-probe": "1" }, signal: AbortSignal.timeout(PROBE_TIMEOUT), }); - if (response.status >= 400 && response.status < 500) { - return { - reason: await response.text(), - authentication: response.status === 401, - }; + let refusal = refusalOf(response.status, response.headers.get(REFUSAL_HEADER) === "1"); + if (refusal) { + return { reason: await response.text(), authentication: refusal === "authentication" }; } } catch { // Unreachable, or too slow to say, rather than refused; the retry loop is the right answer. @@ -294,12 +383,17 @@ export class Wire { return this.#set("denied", refusal.reason); } - let delay = Math.min(BASE_DELAY * 2 ** this.#attempts, MAX_DELAY) * (0.5 + Math.random()); - if (this.#lostAt !== undefined) { + // While a restart is quiet the next attempt is never far off, so that a server that is + // back is found at once, and those attempts do not count towards the backoff that follows. + let quiet = this.#quietLeft() > 0; + let delay = quiet + ? restartRetryDelay(Math.random()) + : Math.min(BASE_DELAY * 2 ** this.#attempts, MAX_DELAY) * (0.5 + Math.random()); + if (!quiet && this.#lostAt !== undefined) { let rescue = this.#lostAt + RESCUE_AT - Date.now(); if (rescue > 0) delay = Math.min(delay, rescue); } - this.#attempts++; + if (!quiet) this.#attempts++; this.#timer = setTimeout(() => { this.#timer = undefined; this.#connect(); @@ -351,6 +445,8 @@ export class Wire { this.#unlisten(); if (this.#timer) clearTimeout(this.#timer); this.#timer = undefined; + clearTimeout(this.#quietTimer); + this.#quietTimer = undefined; let socket = this.#socket; this.#socket = undefined; this.#abandon("document deleted"); @@ -360,8 +456,8 @@ export class Wire { } /** Reject everything still waiting; a reply cannot survive its connection. */ - #abandon(reason: string): void { - for (let waiting of this.#pending.values()) waiting.reject(new Error(reason)); + #abandon(reason: string, restarting = false): void { + for (let waiting of this.#pending.values()) waiting.reject(new WireLost(reason, restarting)); this.#pending.clear(); } @@ -402,6 +498,8 @@ export class Wire { this.#disposed = true; this.#unlisten(); if (this.#timer) clearTimeout(this.#timer); + clearTimeout(this.#quietTimer); + this.#quietTimer = undefined; this.#abandon("disposed"); this.#socket?.close(); this.#socket = undefined; diff --git a/docs/agent-connection.md b/docs/agent-connection.md index b9b4867f7..1afc79fec 100644 --- a/docs/agent-connection.md +++ b/docs/agent-connection.md @@ -913,6 +913,7 @@ needs no feature check: an older Chopin shows the same text unformatted. | 1008 | `error` with `malformed-message` or `protocol` | Unparseable, invalid, or out-of-order message | | 1009 | `error` with `message-too-large` | A frame over 2 MiB | | 1011 | `refused` with `unavailable` | The server could not check the document | +| 1012 | nothing | The server is restarting | | 4401 | `refused` | `sign-in-required` | | 4403 | `refused` | `repository-forbidden` or `access-revoked` | | 4404 | `refused` | `document-unavailable` or `document-archived` | @@ -1032,8 +1033,8 @@ not apply to it. The server keeps sessions, holds and open requests in memory only, so a restart ends what a disconnected link kept, and a graceful one ends it exactly as a crash does. The -server tells every link it is going down before it closes any socket, and no link -withdraws a question on the way out, so the Decisions the Planner had asked stay open +server tells every link it is going down before it closes any socket, closes each with 1012, +and no link withdraws a question on the way out, so the Decisions the Planner had asked stay open and answerable in the document, as after a crash. A returning client's `resume` finds nothing: `resumed` lists no sessions, the client calls its `destroy` handler for each session it held and settles every question it still waited on as `cancelled`. The run @@ -1163,7 +1164,7 @@ pull request (see to `CHOPIN_DOCUMENT`. - On HTTP 401 at the upgrade, or when the connection closes for any reason other than its own `detach`, a backend that did not offer `resumable-sessions` exits. Until the - credential's 15 minutes have passed it may instead reattach with it after 1001 or a + credential's 15 minutes have passed it may instead reattach with it after 1001, 1012 or a dropped socket; after that the credential is refused and it must exit. Chopin forgets a credential once it has expired. - A backend that offers `resumable-sessions` does not exit when its link drops diff --git a/docs/self-hosting.md b/docs/self-hosting.md index 3d4e60755..1db70fd79 100644 --- a/docs/self-hosting.md +++ b/docs/self-hosting.md @@ -581,6 +581,9 @@ The proxy must: - proxy WebSocket upgrades on `/ws`, and on `/api/v1/agent` under `HARNESS=remote`; - proxy `/mcp` and the `/api/v1/agent` upgrade without removing their `Authorization` header, nor the upgrade's `Sec-WebSocket-Protocol` header; +- pass Chopin's response headers through unchanged, so a browser can tell Chopin's + refusal of a WebSocket handshake (`x-chopin-refusal: 1`) from the proxy's own + 4xx while no upstream answers; - serve Chopin at `/`, not below a path prefix; and - redirect alternate hosts to the canonical origin before application traffic, except the paths a [moved deployment](#moving-to-a-new-origin) keeps serving. @@ -700,6 +703,20 @@ The image health check calls `/api/session`. It confirms HTTP liveness but does not prove that a browser bundle is present, a new database transaction can complete, GitHub is reachable, or the Planner can start. +Two unauthenticated endpoints answer with a status word and nothing else, for a +proxy or platform that routes on them: + +- `GET /api/health/live` answers `200 live` while the process runs. +- `GET /api/health/ready` answers `200 ready` once the process holds the writer + lease and has finished startup recovery. It answers `503 starting` before + that, and `503 draining` from the moment a shutdown begins. + +Send new connections only to a ready process. Readiness does not check that +GitHub is reachable or that a model answers. Both endpoints answer `HEAD` the +same way. +They are the first step of Stage 1 of the zero-downtime upgrades design, +invisible restarts on one server. + ## Operations ### Backups and restore @@ -721,6 +738,43 @@ Browser sessions and Planner ownership are cleared at startup. migrations before serving. 4. Repeat the first-start checks after migrations or authentication changes. +A stop signal makes Chopin close every WebSocket with code 1012 (service +restart) before it exits, rather than the 1001 that stopping the listener +sends. An open browser tab takes that as a restart: it reconnects after a +random moment of up to a second and goes on retrying about once a second, says +nothing about it for the first 5 seconds, and keeps its unacknowledged edits, +drafts and answers to send once connected. In those 5 seconds the connection +reports no outage, so the notice for a lost connection does not wait on a +shorter count, and the attempts that fail meanwhile are not counted towards +the backoff or towards offering Retry. A restart that lasts longer is then +treated as any other lost connection, with the same moment of grace before +the notice. A refusal Chopin makes of a handshake carries +`x-chopin-refusal: 1`, and a tab stops retrying on a 4xx only when it does (a +401 still asks for a sign-in). Without it the 4xx is a proxy's and the tab keeps +retrying. This is the client half of Stage 1 of the zero-downtime upgrades +design, invisible restarts on one server; it does not make the gap between the +old process stopping and the new one listening any shorter. Two things differ +from that design's current revision, which adds no restart jitter and keeps the +existing 1.5 seconds of silence for a restart: the random moment before the first +attempt is added so that tabs do not all return together, and the silence is 5 +seconds, because a restart that stops the old process before it starts the new +one takes longer than 1.5 seconds. The design's 1.5 seconds assumes a second +process already waiting to take over. + +A tab opened before an upgrade asks for the hashed `/assets/*` files of its own +build, which the new image does not carry, since a build here has only its own +files. Chopin answers a missing file under `/assets/` with 404 and never with the +app's HTML. When a part of the page fails to load for that reason (Vite's +`vite:preloadError`), the tab reloads to fetch the new build, but only while the +server answers with a page of another build than the one running (it asks for the +page and compares the entry script, so a failure with the server away, or with no +upgrade behind it, is left to the page's own recovery) and it holds nothing the +server has not taken: no unacknowledged edit, unsent message, answer or draft. +Otherwise it leaves the failure as it was and a later one reloads it, and a reload +that did not cure the failure is not repeated within ten seconds. Keeping the previous build's assets in the new image would let such a tab +carry on without reloading, but it needs the previous image, which nothing in this +repository's build has. + Migration files are checksummed once applied. Never edit an applied migration. Because migrations are forward-only, returning to an older application image may require restoring the pre-upgrade database rather than merely changing the diff --git a/e2e/restart.e2e.ts b/e2e/restart.e2e.ts new file mode 100644 index 000000000..bb3852afa --- /dev/null +++ b/e2e/restart.e2e.ts @@ -0,0 +1,665 @@ +/** + * A deploy under an open tab: the server stops and a new one starts while a person is typing, and + * the tab shows nothing, loses nothing and goes on. Restarting needs a server of its own, so this + * file starts one against a database of its own. + */ +import { spawn } from "node:child_process"; + +import { SQL } from "bun"; + +import { RESTART_QUIET_MS } from "../packages/protocol/restart"; +import { chatInput, expectChatValue } from "./chat-input"; +import { + createChannel, + readSource, + registerTestDatabase, + seedChannel, + testChannelPath, +} from "./database"; +import { authenticate, content, expect, ready, status, test } from "./room"; +import { HOST, ROOT } from "./servers"; + +import type { Page } from "@playwright/test"; + +const PORT = 8796; +const ORIGIN = `http://${HOST}:${PORT}`; +const DATABASE = "chopin_restart_quiet"; +const PREFETCH = `Failed to fetch dynamically imported module: ${ORIGIN}/assets/dialog-prefetch-`; + +function databaseUrls() { + let cluster = new URL( + process.env.E2E_DATABASE_URL_3 + || "postgresql://chopin:chopin@127.0.0.1:5436/chopin?sslmode=disable", + ); + let own = new URL(cluster); + own.pathname = `/${DATABASE}`; + return { cluster: cluster.href, own: own.href }; +} + +async function freshDatabase(): Promise { + let { cluster, own } = databaseUrls(); + let admin = new SQL(cluster); + try { + await admin.unsafe(`DROP DATABASE IF EXISTS ${DATABASE} WITH (FORCE)`); + await admin.unsafe(`CREATE DATABASE ${DATABASE}`); + } finally { + await admin.close(); + } + return own; +} + +const ENV = { + SERVER_HOST: HOST, + PORT: String(PORT), + APP_ORIGIN: ORIGIN, + STORAGE_DRIVER: "postgres", + JEV_API_KEY: "e2e-jev-only", + GITHUB_APP_SLUG: "chopin-e2e", + GITHUB_APP_CLIENT_ID: "e2e", + GITHUB_APP_CLIENT_SECRET: "e2e", + GITHUB_ALLOWED_USERS: "", + GITHUB_ALLOWED_ORGANIZATIONS: "githubnext", + SESSION_ENCRYPTION_KEY: process.env.SESSION_ENCRYPTION_KEY!, + AGENT: "off", + BACKGROUND_JOBS: "off", + WEB_RESEARCH: "off", + CONVERSATION_PLAN: "off", + DEV_QUESTIONS: "", + DEV_COMMENTS: "", +}; + +type Running = { child: ReturnType; log: { text: string } }; + +let processes: Running[] = []; + +// One disposable database and application port for the file. +test.describe.configure({ mode: "default" }); + +/** SIGTERM, as a deploy sends it, and what the process exited with. */ +async function stop(running: Running): Promise { + processes = processes.filter(value => value !== running); + let { child } = running; + if (child.exitCode !== null || child.pid === undefined) return child.exitCode; + let exit = new Promise(resolve => child.once("close", code => resolve(code))); + process.kill(-child.pid, "SIGTERM"); + return exit; +} + +async function health(path: string): Promise<[number, string]> { + let response = await fetch(`${ORIGIN}${path}`); + return [response.status, await response.text()]; +} + +/** Start Chopin on the file's database and wait until it says it is ready for traffic. */ +async function server(database: string): Promise { + let child = spawn(process.execPath, ["--preload", "./e2e/github.ts", "apps/server/src/main.ts"], { + cwd: ROOT, + detached: true, + env: { ...process.env, ...ENV, DATABASE_URL: database }, + stdio: ["ignore", "pipe", "pipe"], + }); + let log = { text: "" }; + let read = (chunk: Buffer) => { + log.text = (log.text + String(chunk)).slice(-20_000); + }; + child.stdout?.on("data", read); + child.stderr?.on("data", read); + let running = { child, log }; + processes.push(running); + for (let index = 0; index < 300; index++) { + if (child.exitCode !== null) { + throw new Error(`server exited with ${child.exitCode}: ${log.text.slice(-800)}`); + } + try { + let [code, word] = await health("/api/health/ready"); + if (code === 200 && word === "ready") return running; + } catch { /* Not bound yet. */ } + await new Promise(resolve => setTimeout(resolve, 50)); + } + throw new Error("server did not become ready"); +} + +async function migrated(database: string) { + let migration = Bun.spawn(["bun", "apps/server/src/storage/migrate.ts"], { + cwd: ROOT, + env: { ...process.env, ...ENV, DATABASE_URL: database }, + stdout: "ignore", + stderr: "pipe", + }); + if (await migration.exited !== 0) { + throw new Error( + `migration failed: ${(await new Response(migration.stderr).text()).slice(-800)}`, + ); + } +} + +type Probe = { + closes: number[]; + violations: string[]; + timeline: string[]; + watching: boolean; + /** Whether the page may say it is offline, as it does once a restart has outlasted its quiet. */ + allowOffline: boolean; + /** When, by `performance.now()`, the first 1012 arrived, the first thing said and the notice. */ + lostAt?: number; + saidAt?: number; + noticeAt?: number; +}; + +/** + * Record, in the page itself, every code a socket closes with and anything the page says about + * the connection while watching: sampled every 25 ms, so no notice can come and go unseen. The + * document's one offline notice and its polite word once it has gone are among what is watched. + */ +async function watch(page: Page) { + await page.addInitScript(() => { + let probe: Probe = { + closes: [], + violations: [], + timeline: [], + watching: false, + allowOffline: false, + }; + let at = () => Math.round(performance.now()); + (window as unknown as { probe: Probe }).probe = probe; + let Native = window.WebSocket; + window.WebSocket = class extends Native { + constructor(url: string | URL, protocols?: string | string[]) { + super(url, protocols); + this.addEventListener("open", () => probe.timeline.push(`${at()} open`)); + this.addEventListener("close", event => { + probe.closes.push(event.code); + probe.timeline.push(`${at()} close ${event.code}`); + if (event.code === 1012) probe.lostAt ??= at(); + }); + } + } as typeof WebSocket; + setInterval(() => { + if (!probe.watching) return; + let say = (what: string) => { + probe.violations.push(what); + probe.saidAt ??= at(); + probe.timeline.push(`${at()} ${what}`); + }; + let level = document.querySelector(".plan-status")?.getAttribute("data-level"); + if (level && level !== "hidden") say(`status ${level}`); + // The error line, such as "Message not sent", as opposed to the composer's quiet notes. + let failure = document.querySelector(".composer-notice.motion-feedback")?.textContent?.trim(); + if (failure) say(`composer error ${failure}`); + if (document.querySelector(".plan[data-plan-offline]")) say("locked"); + if (document.querySelector(".plan-content")?.getAttribute("contenteditable") === "false") { + say("not editable"); + } + // The one notice for a lost connection, and the word said when it has gone. + if (document.querySelector("[data-offline-notice]")) { + probe.noticeAt ??= at(); + if (!probe.allowOffline) say("offline notice"); + } + let spoken = document.querySelector("[data-offline-status]")?.textContent?.trim(); + if (spoken && !probe.allowOffline) say(`offline status ${spoken}`); + }, 25); + }); +} + +function probe(page: Page) { + return page.evaluate(() => (window as unknown as { probe: Probe }).probe); +} + +/** + * What the page throws, sorted. While no server answers, the sidebar's idle warm-up of its + * dialogs (`dialog-prefetch`) asks for a chunk and fails, which only the console sees: it is + * kept apart, with when it happened, and nothing else is allowed. Any other error, a different + * chunk included, is unexpected. + */ +function collectErrors(page: Page) { + let all: Array<{ at: number; message: string }> = []; + page.on("pageerror", error => all.push({ at: Date.now(), message: error.message })); + let prefetch = (message: string) => + message.startsWith(PREFETCH) && /^[\w-]+\.js$/.test(message.slice(PREFETCH.length)); + return { + get prefetch() { + return all.filter(error => prefetch(error.message)).map(error => error.at); + }, + unexpected: () => all.filter(error => !prefetch(error.message)).map(error => error.message), + }; +} + +/** + * A deploy, as a tab can tell: it asks the server for its page, and the page names another entry + * script than the one the tab runs. Only that answer is changed, never the page the tab loads, and + * a server that does not answer is not made to. + */ +async function deployed(page: Page) { + await page.route(`${ORIGIN}/**`, async route => { + let request = route.request(); + let own = new URL(page.url()); + own.hash = ""; + if (request.resourceType() !== "fetch" || request.url() !== own.href) return route.fallback(); + try { + let response = await route.fetch(); + let html = (await response.text()).replace( + /(]*type="module"[^>]*src=")[^"]+/, + "$1/assets/index-NextBuild1.js", + ); + await route.fulfill({ response, body: html }); + } catch { + await route.abort(); + } + }); +} + +test.afterEach(async () => { + for (let running of [...processes].toReversed()) await stop(running); +}); + +test("a restart under an open tab shows nothing, loses no edit and keeps the draft", async ({ page }) => { + test.setTimeout(120_000); + let database = await freshDatabase(); + await migrated(database); + let first = await server(database); + registerTestDatabase(PORT, database); + let room = crypto.randomUUID(); + await createChannel(PORT, room); + await seedChannel(PORT, room, "# Restart tracer\n\nSteady text.\n"); + + let loads = 0; + let errors = collectErrors(page); + page.on("load", () => loads++); + await watch(page); + await authenticate(page, `ana-${crypto.randomUUID().slice(0, 8)}`, ORIGIN); + await page.goto(`${ORIGIN}${testChannelPath(room)}`); + await ready(page); + expect(await health("/api/health/live")).toEqual([200, "live"]); + expect(await health("/api/health/ready")).toEqual([200, "ready"]); + + let chat = page.getByRole("complementary", { name: "Chat", exact: true }); + let draft = "A message for the team, not sent yet"; + await chatInput(chat).fill(draft); + await content(page).click(); + await content(page).press("ControlOrMeta+End"); + await page.evaluate(() => { + (window as unknown as { tab: string }).tab = "kept"; + (window as unknown as { probe: Probe }).probe.watching = true; + }); + + // Typing goes on all the while: numbered words, each of which must arrive exactly once. + const WORDS = 50; + let typing = (async () => { + for (let word = 0; word < WORDS; word++) { + await page.keyboard.type(`w${word} `); + await page.waitForTimeout(100); + } + })(); + await page.waitForTimeout(1_000); + let stopped = Date.now(); + let code = await stop(first); + // The gap a platform leaves between stopping the old container and starting the new one. + await new Promise(resolve => setTimeout(resolve, 1_000)); + let second = await server(database); + let outage = Date.now() - stopped; + let back = Date.now(); + await typing; + + expect(code, first.log.text).toBe(0); + expect(await health("/api/health/ready")).toEqual([200, "ready"]); + // Longer than a blip, which the page waits out in 1.5 s, and quicker than it waits for a restart. + expect(outage, second.log.text).toBeGreaterThan(1_500); + expect(outage, second.log.text).toBeLessThan(4_000); + + await expect(status(page)).toHaveAttribute("data-level", "hidden"); + await expect(content(page)).toHaveAttribute("contenteditable", "true"); + let seen = await probe(page); + // The server said it was restarting, in so many words, rather than going away. + expect(seen.closes).toContain(1012); + expect(seen.closes).not.toContain(1001); + expect(seen.violations, seen.timeline.join("\n")).toEqual([]); + // Every error the page raised was the one idle prefetch, and only while no server answered. + expect(errors.unexpected()).toEqual([]); + expect(errors.prefetch.length).toBeLessThanOrEqual(1); + for (let at of errors.prefetch) { + expect(at).toBeGreaterThanOrEqual(stopped); + expect(at).toBeLessThanOrEqual(back); + } + expect(loads).toBe(1); + expect(await page.evaluate(() => (window as unknown as { tab?: string }).tab)).toBe("kept"); + await expectChatValue(chatInput(chat), draft); + + await expect.poll(() => readSource(PORT, room), { timeout: 20_000 }).toContain(`w${WORDS - 1}`); + let words = [...(await readSource(PORT, room)).matchAll(/\bw(\d+)\b/g)].map(match => + Number(match[1]) + ); + expect(words).toEqual(Array.from({ length: WORDS }, (_, index) => index)); +}); + +test("a restart that outlasts its quiet period is then told as any other lost connection, and that goes when the server is back", async ({ page }) => { + test.setTimeout(120_000); + let database = await freshDatabase(); + await migrated(database); + let first = await server(database); + registerTestDatabase(PORT, database); + let room = crypto.randomUUID(); + await createChannel(PORT, room); + await seedChannel(PORT, room, "# A long restart\n\nSteady text.\n"); + + await watch(page); + await authenticate(page, `ana-${crypto.randomUUID().slice(0, 8)}`, ORIGIN); + await page.goto(`${ORIGIN}${testChannelPath(room)}`); + await ready(page); + await content(page).click(); + await content(page).press("ControlOrMeta+End"); + await page.evaluate(() => { + let { probe } = window as unknown as { probe: Probe }; + probe.allowOffline = true; + probe.watching = true; + }); + + let code = await stop(first); + expect(code, first.log.text).toBe(0); + + // Nothing is said for the quiet period, and after it the document's one notice, as for any + // other loss, once its own moment of grace has passed. + let notice = page.locator("[data-offline-notice]"); + await expect(notice).toBeVisible({ timeout: 20_000 }); + let seen = await probe(page); + expect(seen.closes).toContain(1012); + expect(seen.lostAt).toBeDefined(); + expect(seen.saidAt! - seen.lostAt!, seen.timeline.join("\n")).toBeGreaterThanOrEqual( + RESTART_QUIET_MS, + ); + expect(seen.noticeAt! - seen.lostAt!, seen.timeline.join("\n")).toBeLessThan( + RESTART_QUIET_MS + 4_000, + ); + + let second = await server(database); + await expect(notice).toHaveCount(0, { timeout: 20_000 }); + await expect(page.locator("[data-offline-status]")).toHaveText("Reconnected"); + await expect(content(page)).toHaveAttribute("contenteditable", "true"); + expect(await health("/api/health/ready"), second.log.text).toEqual([200, "ready"]); +}); + +/** How many times `text` is in what the room has stored, chat included. */ +async function stored(database: string, room: string, text: string): Promise { + let connection = new SQL(database); + try { + let [row] = await connection<{ body: string }[]>` + SELECT sidecar::text AS body FROM channel_state WHERE channel_id = ${room} + `; + return row ? row.body.split(text).length - 1 : 0; + } finally { + await connection.close(); + } +} + +for (let reached of [true, false]) { + test(`a message being sent as the server restarts, ${reached ? "one it had already taken" : "one it never read"}, is sent again once it is back and arrives once`, async ({ page }) => { + test.setTimeout(120_000); + let database = await freshDatabase(); + await migrated(database); + let first = await server(database); + registerTestDatabase(PORT, database); + let room = crypto.randomUUID(); + await createChannel(PORT, room); + await seedChannel(PORT, room, "# In flight\n\nSteady text.\n"); + + // Between the page and the server: while `cut` holds, a send either goes unread, or is + // taken and its answer is lost with the connection, as when the server stops mid-request. + let cut = { on: false }; + let sent: Array<{ requestId?: string }> = []; + let lost = 0; + await page.routeWebSocket("**/ws?**", socket => { + let upstream = socket.connectToServer(); + socket.onMessage(message => { + let frame = typeof message === "string" + ? JSON.parse(message) as { kind?: string; requestId?: string } + : {}; + if (frame.kind === "chat:send") { + sent.push(frame); + if (cut.on && !reached) return; + } + upstream.send(message); + }); + upstream.onMessage(message => { + let kind = typeof message === "string" + ? (JSON.parse(message) as { kind?: string }).kind + : ""; + if (cut.on && reached && kind === "chat:send") { + lost++; + return; + } + socket.send(message); + }); + socket.onClose((code, reason) => void upstream.close({ code, reason })); + upstream.onClose((code, reason) => void socket.close({ code, reason })); + }); + + let loads = 0; + let errors = collectErrors(page); + page.on("load", () => loads++); + await watch(page); + await authenticate(page, `ana-${crypto.randomUUID().slice(0, 8)}`, ORIGIN); + await page.goto(`${ORIGIN}${testChannelPath(room)}`); + await ready(page); + + let chat = page.getByRole("complementary", { name: "Chat", exact: true }); + let message = `Sent as the server restarted ${crypto.randomUUID().slice(0, 8)}`; + await page.evaluate(() => { + (window as unknown as { probe: Probe }).probe.watching = true; + }); + cut.on = true; + await chatInput(chat).fill(message); + await chatInput(chat).press("Enter"); + await expect.poll(() => reached ? lost : sent.length, { timeout: 10_000 }).toBe(1); + if (reached) expect(await stored(database, room, message)).toBe(1); + + let stopped = Date.now(); + let code = await stop(first); + cut.on = false; + await new Promise(resolve => setTimeout(resolve, 1_000)); + let second = await server(database); + let outage = Date.now() - stopped; + + // The message is sent again, under the same request, and the draft goes once it is taken. + await expectChatValue(chatInput(chat), ""); + await expect(chat.getByText(message, { exact: true })).toHaveCount(1); + expect(code, first.log.text).toBe(0); + expect(outage, second.log.text).toBeLessThan(4_000); + expect(sent).toHaveLength(2); + expect(sent[1]!.requestId).toBeTruthy(); + expect(sent[1]!.requestId).toBe(sent[0]!.requestId); + expect(await stored(database, room, message)).toBe(1); + + await expect(status(page)).toHaveAttribute("data-level", "hidden"); + let seen = await probe(page); + expect(seen.closes).toContain(1012); + expect(seen.closes).not.toContain(1001); + expect(seen.violations, seen.timeline.join("\n")).toEqual([]); + expect(errors.unexpected()).toEqual([]); + expect(loads).toBe(1); + }); +} + +test("an asset of a build that is gone answers 404, and a tab reloads for it only when it holds nothing", async ({ page }) => { + test.setTimeout(90_000); + let database = await freshDatabase(); + await migrated(database); + await server(database); + registerTestDatabase(PORT, database); + let room = crypto.randomUUID(); + await createChannel(PORT, room); + await seedChannel(PORT, room, "# Old build\n\nSteady text.\n"); + + let gone = await page.request.get(`${ORIGIN}/assets/old-chunk-Ab12Cd34.js`); + expect(gone.status()).toBe(404); + expect(gone.headers()["content-type"]).not.toContain("text/html"); + let entry = (await (await page.request.get(ORIGIN)).text()).match(/\/assets\/[^"]+\.js/)![0]; + let present = await page.request.get(`${ORIGIN}${entry}`); + expect(present.status()).toBe(200); + expect(present.headers()["cache-control"]).toContain("immutable"); + + let loads = 0; + page.on("load", () => loads++); + await authenticate(page, `ana-${crypto.randomUUID().slice(0, 8)}`, ORIGIN); + await page.goto(`${ORIGIN}${testChannelPath(room)}`); + await ready(page); + await page.evaluate(() => { + (window as unknown as { tab: string }).tab = "kept"; + }); + let preloadError = () => + page.evaluate(() => { + window.dispatchEvent(new Event("vite:preloadError", { cancelable: true })); + }); + + // With the same build still served there is nothing to fetch, so a failure is left to the + // page's own recovery, whatever the reason for it. + await preloadError(); + await page.waitForTimeout(1_000); + expect(loads).toBe(1); + + // An unsent message is work a reload would lose, so the failure is left to run its course. + await deployed(page); + let chat = page.getByRole("complementary", { name: "Chat", exact: true }); + await chatInput(chat).fill("A message not sent yet"); + await preloadError(); + await page.waitForTimeout(1_000); + expect(loads).toBe(1); + expect(await page.evaluate(() => (window as unknown as { tab?: string }).tab)).toBe("kept"); + + // Once nothing is held, the next failure reloads the tab onto the new build. + await chatInput(chat).fill(""); + let reloaded = page.waitForEvent("load"); + await preloadError(); + await reloaded; + await ready(page); + expect(loads).toBe(2); + expect(await page.evaluate(() => (window as unknown as { tab?: string }).tab)).toBeUndefined(); + + // A failure straight after is not answered with another reload, which would never end. + await preloadError(); + await page.waitForTimeout(1_000); + expect(loads).toBe(2); +}); + +test("a tab holding a reply or a research question typed into the document is not reloaded for a chunk that is gone", async ({ page }) => { + test.setTimeout(90_000); + let database = await freshDatabase(); + await migrated(database); + await server(database); + registerTestDatabase(PORT, database); + let room = crypto.randomUUID(); + await createChannel(PORT, room); + await seedChannel(PORT, room, "# Held\n\nThe rollout goes team by team.\n"); + + let loads = 0; + page.on("load", () => loads++); + await authenticate(page, `ana-${crypto.randomUUID().slice(0, 8)}`, ORIGIN); + await page.goto(`${ORIGIN}${testChannelPath(room)}`); + await ready(page); + await deployed(page); + let preloadError = () => + page.evaluate(() => { + window.dispatchEvent(new Event("vite:preloadError", { cancelable: true })); + }); + + // A comment, and a reply to it begun and not sent. + await page.waitForFunction(() => document.getAnimations().every(a => a.playState !== "running")); + await content(page).locator("p", { hasText: "team by team" }).first().selectText(); + await page.getByRole("button", { name: "Comment on this passage", exact: true }).click(); + let card = page.getByRole("dialog", { name: "New comment" }); + await card.getByRole("textbox", { name: "Comment" }).fill("Who goes first?"); + await card.getByRole("button", { name: "Post comment", exact: true }).click(); + await expect(card).toHaveCount(0); + await page.getByRole("button", { name: /^Comment on “The rollout goes team by team/ }).click(); + let thread = page.getByRole("dialog", { name: "Comment thread" }); + let reply = thread.getByRole("textbox", { name: "Reply", exact: true }); + await expect(thread).toContainText("Who goes first?"); + await reply.fill("The docs team, I think."); + await preloadError(); + await page.waitForTimeout(1_000); + expect(loads).toBe(1); + + // Sent, it is the server's, and the field holds nothing. + await reply.press("Enter"); + await expect(thread).toContainText("The docs team, I think."); + await expect(reply).toHaveValue(""); + + // A research question typed into the document and not started. + await content(page).focus(); + await page.keyboard.press("ControlOrMeta+End"); + await page.keyboard.press("Enter"); + await page.keyboard.type("/research"); + await page.getByRole("listbox", { name: "Insert block" }) + .getByRole("option", { name: "Research", exact: true }) + .click(); + let question = page.getByRole("region", { name: "Research question", exact: true }) + .getByRole("textbox", { name: "Research question", exact: true }); + await question.fill("Which teams have shipped this before?"); + await preloadError(); + await page.waitForTimeout(1_000); + expect(loads).toBe(1); + + // Put away, nothing is held and the next failure reloads the tab onto the new build. + await question.press("Escape"); + await expect(question).toHaveCount(0); + let reloaded = page.waitForEvent("load"); + await preloadError(); + await reloaded; + expect(loads).toBe(2); +}); + +test("a chunk that fails while the server is away does not reload the tab into the outage", async ({ page }) => { + test.setTimeout(90_000); + let database = await freshDatabase(); + await migrated(database); + let first = await server(database); + registerTestDatabase(PORT, database); + let room = crypto.randomUUID(); + await createChannel(PORT, room); + await seedChannel(PORT, room, "# Away\n\nSteady text.\n"); + + let loads = 0; + page.on("load", () => loads++); + await authenticate(page, `ana-${crypto.randomUUID().slice(0, 8)}`, ORIGIN); + await page.goto(`${ORIGIN}${testChannelPath(room)}`); + await ready(page); + await page.evaluate(() => { + (window as unknown as { tab: string }).tab = "kept"; + }); + + // Nothing is unsent and another build would be served, but the server is gone: a reload now + // would land on the browser's error page. + await deployed(page); + await stop(first); + await page.evaluate(() => { + window.dispatchEvent(new Event("vite:preloadError", { cancelable: true })); + }); + await page.waitForTimeout(2_000); + expect(loads).toBe(1); + expect(await page.evaluate(() => (window as unknown as { tab?: string }).tab)).toBe("kept"); + + await server(database); + await expect(status(page)).toHaveAttribute("data-level", "hidden", { timeout: 20_000 }); + await expect(content(page)).toHaveAttribute("contenteditable", "true"); + expect(loads).toBe(1); +}); + +test("a chunk a deploy removed reloads the tab once and not for ever", async ({ page }) => { + test.setTimeout(90_000); + let database = await freshDatabase(); + await migrated(database); + await server(database); + registerTestDatabase(PORT, database); + + // What the server answers for a file of the build before it, which this one does not carry. + await deployed(page); + await page.route( + "**/assets/project-sidebar-*.js", + route => route.fulfill({ status: 404, contentType: "text/plain", body: "not found" }), + ); + let loads = 0; + page.on("load", () => loads++); + await authenticate(page, `ana-${crypto.randomUUID().slice(0, 8)}`, ORIGIN); + await page.goto(ORIGIN); + await expect.poll(() => loads, { timeout: 15_000 }).toBe(2); + await page.waitForTimeout(3_000); + expect(loads).toBe(2); +}); diff --git a/packages/editor/src/comment-layer.tsx b/packages/editor/src/comment-layer.tsx index 0876168b1..7846000cd 100644 --- a/packages/editor/src/comment-layer.tsx +++ b/packages/editor/src/comment-layer.tsx @@ -21,6 +21,7 @@ import { CommentSheet, commentSheetHeading } from "./comment-sheet"; import { useCommentSheetReveal } from "./comment-sheet-reveal"; import { $rangeOf } from "./marks"; import { blockElement } from "./scroll"; +import { usePendingWork } from "./pending-work"; import { COARSE_POINTER_QUERY, PRIMARY_COARSE_POINTER_QUERY, usePhone } from "./pointer"; import { handlesOpenComment, @@ -441,6 +442,7 @@ export function CommentLayer({ store }: { store: ThreadStore }) { compactRef.current = compact; let immediately = options.motionImmediately?.() ?? false; let draft = canEdit ? state.draft : undefined; + usePendingWork(!!draft); let measurePreview = useCallback((request: PreviewRequest, height: number) => { if (height <= 0) { setPreviewMeasurement(undefined); diff --git a/packages/editor/src/comments.test.tsx b/packages/editor/src/comments.test.tsx index 19a5f540e..42876f619 100644 --- a/packages/editor/src/comments.test.tsx +++ b/packages/editor/src/comments.test.tsx @@ -1,7 +1,7 @@ import { describe, expect, it } from "bun:test"; import { renderToStaticMarkup } from "react-dom/server"; -import { composerKey, DraftCard, mention, stamp, ThreadCard } from "./comments"; +import { composerKey, composerUnsent, DraftCard, mention, stamp, ThreadCard } from "./comments"; import { displayName } from "./display-name"; import type { Comment } from "@chopin/protocol"; @@ -259,3 +259,19 @@ describe("mention", () => { expect(mention("mail@chopin ", 12)).toBeUndefined(); }); }); + +describe("composerUnsent", () => { + it("holds nothing for an empty field or one of spaces", () => { + expect(composerUnsent("", false)).toBe(false); + expect(composerUnsent(" \n ", false)).toBe(false); + }); + + it("holds words typed and not sent", () => { + expect(composerUnsent("Keep the rollout reversible.", false)).toBe(true); + }); + + it("holds a send that has left the field and has not been answered, which a refusal puts back", () => { + expect(composerUnsent("", true)).toBe(true); + expect(composerUnsent("Keep the rollout reversible.", true)).toBe(true); + }); +}); diff --git a/packages/editor/src/comments.tsx b/packages/editor/src/comments.tsx index 8ce1ba817..dd79246e6 100644 --- a/packages/editor/src/comments.tsx +++ b/packages/editor/src/comments.tsx @@ -18,6 +18,7 @@ import { limits } from "@chopin/dialect"; import { ChopinMark } from "./chopin-mark"; import { displayName } from "./display-name"; import { Face } from "./face"; +import { usePendingWork } from "./pending-work"; import { PRIMARY_COARSE_POINTER_QUERY } from "./pointer"; import type { KeyboardEvent, ReactNode } from "react"; @@ -291,6 +292,15 @@ export function mention( return { text, caret: Math.max(0, Math.min(text.length, caret - removed)) }; } +/** + * Whether a comment composer holds words a reload would lose: text typed and not sent, or a send + * whose answer has not come. The words leave the field when sent and return if the send is refused, + * so the field alone does not say. + */ +export function composerUnsent(text: string, sending: boolean): boolean { + return sending || text.trim() !== ""; +} + /** * The field and its send, shared by new comments and replies. * @@ -320,6 +330,7 @@ function Composer({ onTyping?: (writing: boolean) => void; }) { let [text, setText] = useState(""); + let [sending, setSending] = useState(false); let [to, setTo] = useState(initial); let [focused, setFocused] = useState(false); let [coarse, setCoarse] = useState(false); @@ -329,6 +340,7 @@ function Composer({ let chip = useRef(null); let caret = useRef(undefined); let checkbox = useId(); + usePendingWork(composerUnsent(text, sending)); useEffect(() => { setCoarse(matchMedia(PRIMARY_COARSE_POINTER_QUERY).matches); @@ -387,13 +399,14 @@ function Composer({ setRefusal(undefined); onTyping?.(false); let sent = onSend(value, addressed ? "planner" : undefined); + if (sent) setSending(true); // Refused: put the words back, unless something new was typed meanwhile, and say why. void sent?.then(result => { if (result.ok) return; setRefusal(result.message); setText(current => current ? current : value); setTo(current => current || addressed); - }); + }).finally(() => setSending(false)); // A clicked send disables itself and drops focus without a blur, which would leave // the field believing it still had focus; the field takes it back instead. ref.current?.focus({ preventScroll: true }); diff --git a/packages/editor/src/connection-notice.test.ts b/packages/editor/src/connection-notice.test.ts index 4bab797f3..6142b3e34 100644 --- a/packages/editor/src/connection-notice.test.ts +++ b/packages/editor/src/connection-notice.test.ts @@ -1,12 +1,16 @@ import { describe, expect, test } from "bun:test"; +import { RESTART_QUIET_MS } from "@chopin/protocol/restart"; + import { CONNECTION_GRACE, CONNECTION_STALL, + connectionGrace, connectionNotice, isOffline, offersRetry, offlineAnnouncement, + RESTART_GRACE, RETRY_AFTER_FAILURES, } from "./connection-notice"; @@ -22,6 +26,28 @@ describe("connectionNotice", () => { expect(connectionNotice(CONNECTION_GRACE + CONNECTION_STALL - 1)).toBe("reconnecting"); expect(connectionNotice(CONNECTION_GRACE + CONNECTION_STALL)).toBe("offline"); }); + + test("stays silent longer when the server said it is restarting", () => { + expect(RESTART_GRACE).toBeGreaterThan(CONNECTION_GRACE); + // The quiet period, in which the connection reports no outage, and then the ordinary grace. + expect(RESTART_GRACE).toBe(RESTART_QUIET_MS + CONNECTION_GRACE); + expect(connectionNotice(CONNECTION_GRACE, true)).toBe("none"); + expect(connectionNotice(RESTART_GRACE - 1, true)).toBe("none"); + expect(connectionNotice(RESTART_GRACE, true)).toBe("reconnecting"); + expect(connectionNotice(RESTART_GRACE + CONNECTION_STALL - 1, true)).toBe("reconnecting"); + expect(connectionNotice(RESTART_GRACE + CONNECTION_STALL, true)).toBe("offline"); + }); + + test("names the grace a send held through a blip may wait, the same as the silence", () => { + expect(connectionGrace()).toBe(CONNECTION_GRACE); + expect(connectionGrace(false)).toBe(CONNECTION_GRACE); + expect(connectionGrace(true)).toBe(RESTART_GRACE); + }); + + test("keeps its own grace for a loss the server did not announce", () => { + expect(connectionNotice(CONNECTION_GRACE, false)).toBe("reconnecting"); + expect(connectionNotice(CONNECTION_GRACE)).toBe("reconnecting"); + }); }); describe("isOffline", () => { diff --git a/packages/editor/src/connection-notice.ts b/packages/editor/src/connection-notice.ts index 837ad660d..d2b4d0a0d 100644 --- a/packages/editor/src/connection-notice.ts +++ b/packages/editor/src/connection-notice.ts @@ -1,9 +1,19 @@ /** When a lost connection is worth mentioning, shared by every surface that mentions it. */ +import { RESTART_QUIET_MS } from "@chopin/protocol/restart"; import { useEffect, useState } from "react"; /** How long a lost connection stays silent, so a blip that recovers shows nothing. */ export const CONNECTION_GRACE = 1500; +/** + * How long a loss stays silent when the server said it is restarting, which ends within + * moments on its own, so a notice would only alarm. The connection raises no outage for such a + * loss until `RESTART_QUIET_MS` has passed (see `Wire`), and what is shown follows the ordinary + * grace from there, so every surface speaks at the same moment, as for any other loss. The + * zero-downtime upgrades design keeps the 1.5 s grace for a restart; see `RESTART_QUIET_MS` for + * why it cannot here. + */ +export const RESTART_GRACE = RESTART_QUIET_MS + CONNECTION_GRACE; /** How much longer it reads as reconnecting before it is called offline. */ export const CONNECTION_STALL = 5000; /** Failed reconnect attempts in one outage before a person is offered a Retry of their own. */ @@ -15,10 +25,19 @@ export const RETRY_AFTER_FAILURES = 4; */ export type ConnectionNotice = "none" | "reconnecting" | "offline"; +/** How long a lost connection stays silent, `restarting` when the server said it is restarting. */ +export function connectionGrace(restarting = false): number { + return restarting ? RESTART_GRACE : CONNECTION_GRACE; +} + /** The notice for a connection that has been lost for `elapsed` milliseconds. */ -export function connectionNotice(elapsed: number | undefined): ConnectionNotice { - if (elapsed === undefined || elapsed < CONNECTION_GRACE) return "none"; - return elapsed < CONNECTION_GRACE + CONNECTION_STALL ? "reconnecting" : "offline"; +export function connectionNotice( + elapsed: number | undefined, + restarting = false, +): ConnectionNotice { + let grace = connectionGrace(restarting); + if (elapsed === undefined || elapsed < grace) return "none"; + return elapsed < grace + CONNECTION_STALL ? "reconnecting" : "offline"; } /** @@ -60,17 +79,18 @@ export function offlineAnnouncement(wasShown: boolean, shown: boolean): string { * as reopening the document or sending, must keep reading it directly: a * reconnect inside the grace period never changes this value. */ -export function useConnectionNotice(lost: boolean): ConnectionNotice { +export function useConnectionNotice(lost: boolean, restarting = false): ConnectionNotice { let [notice, setNotice] = useState("none"); useEffect(() => { setNotice("none"); if (!lost) return; - let timers = [CONNECTION_GRACE, CONNECTION_GRACE + CONNECTION_STALL].map(delay => - setTimeout(() => setNotice(connectionNotice(delay)), delay) + let grace = connectionGrace(restarting); + let timers = [grace, grace + CONNECTION_STALL].map(delay => + setTimeout(() => setNotice(connectionNotice(delay, restarting)), delay) ); return () => { for (let timer of timers) clearTimeout(timer); }; - }, [lost]); + }, [lost, restarting]); return lost ? notice : "none"; } diff --git a/packages/editor/src/index.ts b/packages/editor/src/index.ts index e491e2a6c..f36092f01 100644 --- a/packages/editor/src/index.ts +++ b/packages/editor/src/index.ts @@ -7,6 +7,7 @@ export type { CollaborationOptions } from "./collaboration"; export { CONNECTION_GRACE, CONNECTION_STALL, + connectionGrace, isOffline, offersRetry, offlineAnnouncement, @@ -46,6 +47,7 @@ export { AgentFace, Face } from "./face"; export type { FaceProps } from "./face"; export { GitHubReferencesProvider } from "./github-references"; export type { Refusal } from "./history"; +export { usePendingWork } from "./pending-work"; export { isDemoHandle, personName, setFaceProfiles } from "./people"; export type { FaceProfile } from "./people"; export { PlanEditor } from "./plan-editor"; diff --git a/packages/editor/src/pending-work.ts b/packages/editor/src/pending-work.ts new file mode 100644 index 000000000..3b7f5b821 --- /dev/null +++ b/packages/editor/src/pending-work.ts @@ -0,0 +1,2 @@ +// Defined beside the questionnaire views, which are below the editor and need it too. +export { usePendingWork } from "@chopin/question/react"; diff --git a/packages/editor/src/plan-editor.tsx b/packages/editor/src/plan-editor.tsx index 1e7ba0331..aaa833105 100644 --- a/packages/editor/src/plan-editor.tsx +++ b/packages/editor/src/plan-editor.tsx @@ -383,8 +383,10 @@ export function PlanEditor( }, [presence, connection]); // Locking waits out a blip, and edits made meanwhile wait in the outbox. - let offline = useConnectionNotice(connection !== undefined && connection !== "connected") - !== "none"; + let offline = useConnectionNotice( + connection !== undefined && connection !== "connected", + wire?.restarting, + ) !== "none"; let blocked = offline || !!busy || !!readOnly || !state.synced; let locked = blocked || editMode === "reading"; diff --git a/packages/editor/src/provider.test.ts b/packages/editor/src/provider.test.ts index 05bc5d114..3e71e9dea 100644 --- a/packages/editor/src/provider.test.ts +++ b/packages/editor/src/provider.test.ts @@ -11,6 +11,7 @@ * layout, and happy-dom returns zero for every measurement. */ +import { hasPendingWork } from "@chopin/protocol/pending-work"; import { afterEach, describe, expect, it, jest } from "bun:test"; import * as Y from "yjs"; @@ -433,6 +434,24 @@ describe("delivering local edits", () => { expect(room.doc.getText("typed").toString()).toBe("xx"); }); + it("holds the page from reloading while an edit is unacknowledged, and lets go once it is", async () => { + jest.useFakeTimers(); + let room = server(index => index === 0); + let doc = new Y.Doc(); + let provider = new PlanProvider({ wire: room.transport, doc }); + await provider.connect(); + expect(provider.unsaved).toBe(false); + + typing(doc, 1); + jest.advanceTimersByTime(100); + expect(provider.unsaved).toBe(true); + expect(hasPendingWork()).toBe(true); + + // The resend is acknowledged. + jest.advanceTimersByTime(10_000); + expect(provider.unsaved).toBe(false); + }); + it("leaves acknowledged edits alone", async () => { jest.useFakeTimers(); let room = server(() => false); diff --git a/packages/editor/src/provider.ts b/packages/editor/src/provider.ts index 64bf96026..4f0e3b2c4 100644 --- a/packages/editor/src/provider.ts +++ b/packages/editor/src/provider.ts @@ -13,6 +13,7 @@ import { applyAwarenessUpdate, Awareness, encodeAwarenessUpdate } from "y-protocols/awareness"; import * as Y from "yjs"; +import { trackPendingWork } from "@chopin/protocol/pending-work"; import type { Plan } from "@chopin/protocol"; import type { Provider } from "@lexical/yjs"; import type { Transport } from "./transport"; @@ -154,6 +155,12 @@ export class PlanProvider implements Provider { this.#wire = options.wire; this.#doc = options.doc; this.awareness = new Awareness(options.doc) as PlanProvider["awareness"]; + trackPendingWork(this); + } + + /** Whether an update has not been acknowledged yet, which a reload would lose. */ + get unsaved(): boolean { + return this.#outbox.size > 0; } get epoch(): string | undefined { diff --git a/packages/editor/src/research-draft.test.ts b/packages/editor/src/research-draft.test.ts index 2e37f23e8..cade47914 100644 --- a/packages/editor/src/research-draft.test.ts +++ b/packages/editor/src/research-draft.test.ts @@ -1,6 +1,9 @@ +import { hasPendingWork } from "@chopin/protocol/pending-work"; import { describe, expect, it } from "bun:test"; import * as Y from "yjs"; +import { ResearchDraftStore } from "./research-draft"; + import type { Research } from "@chopin/protocol"; const REQUEST: Research.RequestView = { @@ -240,3 +243,41 @@ describe("research draft lifecycle", () => { expect(store.get()?.position).toBe(original); }); }); + +describe("a research draft holds the page from reloading", () => { + it("holds nothing until something is typed, and lets go when it is cleared or dismissed", () => { + let store = new ResearchDraftStore("investigation"); + store.open(ANCHOR, position("typed")); + expect(store.unsaved).toBe(false); + + store.change("Which index is faster?"); + expect(store.unsaved).toBe(true); + expect(hasPendingWork()).toBe(true); + + store.change(" "); + expect(store.unsaved).toBe(false); + store.change("Which index is faster?"); + expect(store.dismiss()).toBe(true); + expect(store.unsaved).toBe(false); + }); + + it("holds through the request and until its reference is placed, whatever was typed", async () => { + let store = new ResearchDraftStore(); + store.open(ANCHOR, position("created")); + store.change(REQUEST.question); + let resolve!: (request: Research.RequestView) => void; + let running = store.start(() => new Promise(done => resolve = done)); + expect(store.unsaved).toBe(true); + + resolve(REQUEST); + await running; + // Created, but with no place for its reference yet: leaving would leave it loose. + expect(store.get()?.created).toEqual(REQUEST); + expect(store.unsaved).toBe(true); + + expect(store.place(position("placed"))).toBe(false); + store.attachPlacement(() => true); + expect(store.get()).toBeUndefined(); + expect(store.unsaved).toBe(false); + }); +}); diff --git a/packages/editor/src/research-draft.ts b/packages/editor/src/research-draft.ts index 7f2a81a29..8514fb4e3 100644 --- a/packages/editor/src/research-draft.ts +++ b/packages/editor/src/research-draft.ts @@ -1,3 +1,4 @@ +import { trackPendingWork } from "@chopin/protocol/pending-work"; import type { RelativePosition } from "yjs"; import type { DOMRectLike } from "./toolbar/placement"; @@ -61,6 +62,16 @@ export class ResearchDraftStore { /** `noun` names the work in messages: "research" or "investigation". */ constructor(noun = "research") { this.#noun = noun; + trackPendingWork(this); + } + + /** + * Whether the draft holds something a reload would lose: a question typed and not yet started, + * or a request in flight or created whose reference has not been placed in the document. + */ + get unsaved(): boolean { + let draft = this.#draft; + return !!draft && (draft.phase !== "editing" || draft.question.trim() !== ""); } subscribe(listener: () => void): () => void { diff --git a/packages/editor/src/transport.ts b/packages/editor/src/transport.ts index ebf04a6ea..4a3a017ca 100644 --- a/packages/editor/src/transport.ts +++ b/packages/editor/src/transport.ts @@ -23,6 +23,11 @@ export type Transport = { * Absent means "assume it can", which is what a stub wants. */ readonly connected?: boolean; + /** + * Whether the connection was lost because the server said it is restarting. What is shown + * about such a loss waits longer; what acts on the connection reads `connected`. + */ + readonly restarting?: boolean; }; /** Mirrors the client's connection states, for read-only and status chrome. */ diff --git a/packages/editor/src/widgets/explore.tsx b/packages/editor/src/widgets/explore.tsx index 28d598724..b7a735c02 100644 --- a/packages/editor/src/widgets/explore.tsx +++ b/packages/editor/src/widgets/explore.tsx @@ -19,6 +19,7 @@ import { useEffect, useId, useRef, useState, useSyncExternalStore } from "react" import type { ReactNode } from "react"; import { Face } from "../face"; +import { usePendingWork } from "../pending-work"; import type { Question } from "@chopin/protocol"; import type { Transport } from "../transport"; @@ -194,6 +195,7 @@ export function ExploreMenu( let [focus, setFocus] = useState(""); let [busy, setBusy] = useState(false); let [error, setError] = useState(); + usePendingWork(focus.trim() !== "" || busy); let first = useRef(null); let details = useRef>>({}); let peopleList = useRef(null); diff --git a/packages/protocol/package.json b/packages/protocol/package.json index c2d954d7f..51819ac1f 100644 --- a/packages/protocol/package.json +++ b/packages/protocol/package.json @@ -37,6 +37,14 @@ "types": "./agents.ts", "default": "./agents.ts" }, + "./pending-work": { + "types": "./pending-work.ts", + "default": "./pending-work.ts" + }, + "./restart": { + "types": "./restart.ts", + "default": "./restart.ts" + }, "./run-stages": { "types": "./run-stages.ts", "default": "./run-stages.ts" diff --git a/packages/protocol/pending-work.test.ts b/packages/protocol/pending-work.test.ts new file mode 100644 index 000000000..71a13cdd8 --- /dev/null +++ b/packages/protocol/pending-work.test.ts @@ -0,0 +1,45 @@ +import { beforeEach, expect, test } from "bun:test"; + +import { PendingWork } from "./pending-work"; + +let work = new PendingWork(); + +beforeEach(() => { + work = new PendingWork(); +}); + +test("nothing is pending until a source says it holds work, and again once it lets go", () => { + expect(work.any()).toBe(false); + let source = { unsaved: false }; + work.track(source); + expect(work.any()).toBe(false); + source.unsaved = true; + expect(work.any()).toBe(true); + source.unsaved = false; + expect(work.any()).toBe(false); +}); + +test("one source with work is enough, however many hold none", () => { + let idle = [{ unsaved: false }, { unsaved: false }]; + let busy = { unsaved: true }; + for (let source of [...idle, busy]) work.track(source); + expect(work.any()).toBe(true); + busy.unsaved = false; + expect(work.any()).toBe(false); +}); + +test("a source reads its work when asked, not when it was tracked", () => { + let outbox: string[] = []; + let source = { + get unsaved() { + return outbox.length > 0; + }, + }; + work.track(source); + expect(work.any()).toBe(false); + outbox.push("update"); + expect(work.any()).toBe(true); + outbox.length = 0; + expect(work.any()).toBe(false); + expect(source.unsaved).toBe(false); +}); diff --git a/packages/protocol/pending-work.ts b/packages/protocol/pending-work.ts new file mode 100644 index 000000000..67bde620d --- /dev/null +++ b/packages/protocol/pending-work.ts @@ -0,0 +1,42 @@ +/** + * Work this page holds that nobody else has yet: edits the server has not acknowledged, + * answers and drafts it has not taken, a message not yet sent. + * + * A page that is about to replace itself, reloading for a new build, asks here first. Owners + * are held weakly, so one that is thrown away needs no way to say so, and one that is still + * reachable with work in it keeps the page from reloading, which is the safe mistake. + */ + +export type PendingWorkSource = { readonly unsaved: boolean }; + +/** The sources of one page. */ +export class PendingWork { + readonly #sources = new Set>(); + + /** Count `source` while its `unsaved` is true. */ + track(source: PendingWorkSource): void { + this.#sources.add(new WeakRef(source)); + } + + /** Whether any tracked source holds work that a reload would lose. */ + any(): boolean { + for (let reference of this.#sources) { + let source = reference.deref(); + if (!source) this.#sources.delete(reference); + else if (source.unsaved) return true; + } + return false; + } +} + +const page = new PendingWork(); + +/** Count `source` on this page while its `unsaved` is true. */ +export function trackPendingWork(source: PendingWorkSource): void { + page.track(source); +} + +/** Whether anything on this page holds work that a reload would lose. */ +export function hasPendingWork(): boolean { + return page.any(); +} diff --git a/packages/protocol/restart.ts b/packages/protocol/restart.ts new file mode 100644 index 000000000..173766962 --- /dev/null +++ b/packages/protocol/restart.ts @@ -0,0 +1,32 @@ +/** + * What a client needs to tell a server that is restarting from one that refused it. + * + * Both ends need these and must not disagree: the server closes with the one and marks the + * other, and the client's reconnect is quiet only for a close and an answer it recognizes. + */ + +/** + * The close code of a socket the server ends because it is restarting (IANA "Service Restart"). + * The client reconnects quietly and keeps what it has not sent. + */ +export const SERVICE_RESTART = 1012; + +/** + * Set to `1` on every answer Chopin itself gives to a socket handshake it refuses. A 4xx without + * it came from something in front of Chopin, such as a proxy with no upstream, and says nothing + * about this person's access. + */ +export const REFUSAL_HEADER = "x-chopin-refusal"; + +/** + * How long, in milliseconds, a restart is treated as a moment that will pass: the page says + * nothing about it for this long, and a client that has not been let back in keeps trying about + * once a second, so a server that is back is found at once, not at the end of a long backoff. + * + * This departs from the zero-downtime upgrades design (Stage 1, "What changes, by package", + * `packages/editor`), which keeps the existing 1.5 s silence for a restart. That figure assumes a + * second process already waiting to take over, which is not part of this change. A restart that + * stops the old process before it starts the new one takes longer than 1.5 s, and a notice + * within that time is not allowed to show. + */ +export const RESTART_QUIET_MS = 5000; diff --git a/packages/question/src/react/criteria.tsx b/packages/question/src/react/criteria.tsx index 7d5607f7b..56095606a 100644 --- a/packages/question/src/react/criteria.tsx +++ b/packages/question/src/react/criteria.tsx @@ -14,6 +14,7 @@ import { markKey } from "../draft"; import { MAX_CRITERIA, MAX_CRITERION_LABEL } from "../limits"; import { plainInlineText } from "./inline-segments"; import { Key } from "./option-key"; +import { usePendingWork } from "./use-pending-work"; import type { ComponentType, KeyboardEvent, ReactNode } from "react"; import type { IconProps } from "@chopin/icons"; @@ -208,6 +209,8 @@ function useAdd( ? criteria.find(criterion => criterion.label.trim().toLowerCase() === label.toLowerCase()) : undefined; let ready = !!label && !duplicate; + // Typed text is kept until the criterion is added, so a reload would lose it. + usePendingWork(!!label || pending); let hint = duplicate ? `Already a criterion: ${duplicate.label}` : failure ?? ""; return { text, diff --git a/packages/question/src/react/index.ts b/packages/question/src/react/index.ts index b7f5c0ce2..33be15042 100644 --- a/packages/question/src/react/index.ts +++ b/packages/question/src/react/index.ts @@ -34,5 +34,6 @@ export type { } from "./question-view"; export { cardRelation, NOT_LINKED, RelationNote } from "./relation-note"; export { ResolvedActions } from "./resolved-actions"; +export { usePendingWork } from "./use-pending-work"; export { forget, useQuestionnaire } from "./use-questionnaire"; export type { QuestionnaireOptions, QuestionnaireState, Transport } from "./use-questionnaire"; diff --git a/packages/question/src/react/question-view.tsx b/packages/question/src/react/question-view.tsx index 0cde56c52..bd52a2316 100644 --- a/packages/question/src/react/question-view.tsx +++ b/packages/question/src/react/question-view.tsx @@ -25,6 +25,7 @@ import { projectSuggestion, reduceSuggestionEditState } from "./project-suggesti import { cardRelation, RelationNote } from "./relation-note"; import { ResolvedActions } from "./resolved-actions"; import { TextAnswer } from "./text-answer"; +import { usePendingWork } from "./use-pending-work"; import type { VisibleSuggestion } from "./project-suggestion"; import type { Relation } from "../relation"; @@ -395,6 +396,8 @@ function AddOption( let label = text?.trim() ?? ""; let duplicate = label ? duplicateOf(question, label) : undefined; let ready = !!label && !duplicate && !disabled; + // Typed text is kept until the option is added, so a reload would lose it. + usePendingWork(!!label || pending !== null); useEffect(() => { let target = focus.current; diff --git a/packages/question/src/react/questionnaire-controller.ts b/packages/question/src/react/questionnaire-controller.ts index d96a08bb9..2b37d4089 100644 --- a/packages/question/src/react/questionnaire-controller.ts +++ b/packages/question/src/react/questionnaire-controller.ts @@ -1,3 +1,4 @@ +import { trackPendingWork } from "@chopin/protocol/pending-work"; import { derive } from "../answer"; import { crdt, setMark } from "../draft"; import type { Model as DraftModel } from "../draft"; @@ -111,6 +112,12 @@ export class QuestionnaireController { focus: undefined, closed: false, }; + trackPendingWork(this); + } + + /** Whether a draft edit or an answer has not reached the server, which a reload would lose. */ + get unsaved(): boolean { + return this.#outbox.length > 0 || this.#snapshot.submitting; } getSnapshot = (): Snapshot => this.#snapshot; diff --git a/packages/question/src/react/use-pending-work.ts b/packages/question/src/react/use-pending-work.ts new file mode 100644 index 000000000..c5b04f2a1 --- /dev/null +++ b/packages/question/src/react/use-pending-work.ts @@ -0,0 +1,17 @@ +import { trackPendingWork } from "@chopin/protocol/pending-work"; +import { useEffect } from "react"; + +/** + * Count this component's work as unsaved, such as a message not yet sent, while `unsaved` + * holds, so that a page about to reload for a new build waits for it. + */ +export function usePendingWork(unsaved: boolean): void { + useEffect(() => { + if (!unsaved) return; + let source = { unsaved: true }; + trackPendingWork(source); + return () => { + source.unsaved = false; + }; + }, [unsaved]); +} diff --git a/packages/question/src/react/use-questionnaire-unsaved.test.ts b/packages/question/src/react/use-questionnaire-unsaved.test.ts new file mode 100644 index 000000000..42a144d89 --- /dev/null +++ b/packages/question/src/react/use-questionnaire-unsaved.test.ts @@ -0,0 +1,44 @@ +import { hasPendingWork } from "@chopin/protocol/pending-work"; +import { describe, expect, it } from "bun:test"; +import { create } from "../index"; +import { QuestionnaireController } from "./use-questionnaire"; +import type { Transport } from "./use-questionnaire"; +import { DEFINITION } from "./use-questionnaire.test-fixtures"; + +describe("QuestionnaireController unsaved work", () => { + it("holds the page from reloading while an answer edit has not reached the server", async () => { + let model = create(DEFINITION); + let settleEdit!: (reply: unknown) => void; + let wire = { + ask(kind: string) { + if (kind === "question:open") { + return Promise.resolve({ + open: true, + definition: DEFINITION, + model: [...model.toBinary()], + revision: 0, + presence: [], + }); + } + if (kind === "question:edit") return new Promise(resolve => settleEdit = resolve); + throw new Error(`Unexpected ${kind}`); + }, + send() {}, + on: () => () => {}, + } as unknown as Transport; + let controller = new QuestionnaireController(wire, "w", DEFINITION, true); + let off = controller.subscribe(() => {}); + await Bun.sleep(0); + expect(controller.unsaved).toBe(false); + + controller.change("q0", { choice: "o0" }); + await Bun.sleep(0); + expect(controller.unsaved).toBe(true); + expect(hasPendingWork()).toBe(true); + + settleEdit({ open: true, accepted: true, revision: 1 }); + await Bun.sleep(0); + expect(controller.unsaved).toBe(false); + off(); + }); +}); diff --git a/scripts/design-contract/exceptions/dynamic-editor.json b/scripts/design-contract/exceptions/dynamic-editor.json index 6c0c6a693..d875a3d36 100644 --- a/scripts/design-contract/exceptions/dynamic-editor.json +++ b/scripts/design-contract/exceptions/dynamic-editor.json @@ -228,7 +228,7 @@ 1 ] ], - "sourceHash": "b407049da3918eccef652e8f37d3364690df1cd7df62b7262b022f43cf7028db" + "sourceHash": "89297a705c3d7ec8edb4ffc3e8cef460f029e5a6ff192f6ce6cf35d90d550e83" }, { "file": "packages/editor/src/comment-layer.tsx", @@ -256,7 +256,7 @@ 1 ] ], - "sourceHash": "b407049da3918eccef652e8f37d3364690df1cd7df62b7262b022f43cf7028db" + "sourceHash": "89297a705c3d7ec8edb4ffc3e8cef460f029e5a6ff192f6ce6cf35d90d550e83" }, { "file": "packages/editor/src/table/rails.tsx", @@ -375,7 +375,7 @@ 2 ] ], - "sourceHash": "766613371657f491993aa08f98e4a2853a54a7a57f6728889892df46e496468d" + "sourceHash": "517caf207fcd039e7de67a79d05d0da523eb001997ff0703e721c5187110769d" }, { "file": "packages/editor/src/collaboration.tsx", diff --git a/scripts/design-contract/exceptions/dynamic-packages.json b/scripts/design-contract/exceptions/dynamic-packages.json index 8b31131aa..ab5bc49f9 100644 --- a/scripts/design-contract/exceptions/dynamic-packages.json +++ b/scripts/design-contract/exceptions/dynamic-packages.json @@ -32,7 +32,7 @@ 1 ] ], - "sourceHash": "400b3dbeb9c578e95c94386baaebc06e5903e92f2b5a4671a7a5b486b29c1dc7" + "sourceHash": "e71d33dc6cd28e340980cb16fb935a54de7e78fab279da38c268b8f3cb3659cd" }, { "file": "packages/question/src/react/resolved-actions.tsx", diff --git a/scripts/design-contract/exceptions/dynamic-web.json b/scripts/design-contract/exceptions/dynamic-web.json index cec15934d..319c8626a 100644 --- a/scripts/design-contract/exceptions/dynamic-web.json +++ b/scripts/design-contract/exceptions/dynamic-web.json @@ -73,7 +73,7 @@ 1 ] ], - "sourceHash": "7116df5ad4fc3d55f905cdc426c4eb8de3a7c9fb9903f75a1bf4657b351704ac" + "sourceHash": "63213489ee99add048d63095e34c02e5342e9e1bb8c9714cc3746f97404d6eee" }, { "file": "apps/web/src/chat/markdown.tsx",