From bc862418579a03a176cccc2a5cddc0ef93848494 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 30 Oct 2026 18:22:20 +0100 Subject: [PATCH] fix(terminal): bound backpressure recovery history --- docs/terminal-performance.md | 2 +- .../46-daemon-terminal-worker-upgrade.test.ts | 42 +++- .../src/terminal/terminal-restore.test.ts | 6 +- .../server/src/terminal/terminal-restore.ts | 13 +- .../terminal-session-controller.test.ts | 44 +++- .../terminal/terminal-session-controller.ts | 8 +- .../terminal/worker-terminal-manager.test.ts | 190 +++++++++++++++++- 7 files changed, 289 insertions(+), 16 deletions(-) diff --git a/docs/terminal-performance.md b/docs/terminal-performance.md index 4c85642968..c757d004ec 100644 --- a/docs/terminal-performance.md +++ b/docs/terminal-performance.md @@ -27,7 +27,7 @@ Terminal frames share the daemon main event loop with all agent traffic. The `ev - **Replay geometry belongs to the parser queue.** Native measurements and web fits must leave the captured dimensions intact through every chunk's commit. Apply the latest measured size after replay, before live bytes; clear, replacement and disposal must release that geometry ownership. A layout effect must not resize the renderer directly during replay. - **Coalesced output carries the LAST chunk's revision.** Snapshot replay dedup (`replayTerminalOutputAfterSnapshot`) skips buffered output with `revision <= replayRevision`; a merged batch with a lower revision would be wrongly skipped (lost output). - **The input-mode tracker runs once per process boundary, not per hop.** The worker commits its authoritative tracker with xterm parsing before publishing the corresponding revision; the daemon caches current modes from ordered worker events and adoption. A replay response carries capture-time modes for its viewer; it must not overwrite the shared cache after newer live output. Apply that viewer's preamble before its buffered live bytes. Do not reintroduce a per-chunk `feed()` on the daemon main loop. -- **Snapshot catch-up is backpressure-gated.** A stream falls back to a full snapshot only when `outputBytesSinceSnapshot > MAX_TERMINAL_OUTPUT_FRAME_BYTES` (256KB) **and** the client transport reports `bufferedAmount > MAX_CLIENT_BUFFERED_BYTES` (4MB). A client that keeps draining streams continuously, no matter how much output is produced. Before this gate existed, every 256KB of build output dropped a frame and forced a full JSON cell-grid snapshot (~200k objects across IPC) — the historical source of spiky lag and GC hitches. +- **Snapshot catch-up is backpressure-gated.** After more than 256 KiB of output, a stream catches up only when its transport reports more than 4 MiB queued, or exposes no backpressure reading. A draining client streams continuously. Pressure recovery sends at most 500 recent history rows plus the screen, preserving smaller explicit demands including zero; even a live-only subscription needs this resynchronization. It does not change the subscription's configured demand: later attach/reconnect restores still request the full configured history. The recent window avoids repeatedly serializing all retained history for a slow viewer. - **Plugin daemon sessions report IPC queue bytes.** Their virtual socket increments `bufferedAmount` before `process.send` and decrements it only in the send callback. Text and binary frames share that ordered queue, so the normal snapshot catch-up and physical high-water gates remain valid for server-side plugin SDK traffic. - **Client output writes are not serialized per frame.** The emulator runtime drains contiguous plain writes straight into xterm (which buffers internally). Only barrier ops (`clear`, `snapshot`, `suppressInput` writes) wait — behind a zero-length sentinel write — so resets can't interleave with in-flight output. - **Find preserves the inspected xterm viewport, including at the live bottom.** Keep its output anchor inside the existing batched write path. Serializing output to preserve a search result would add terminal latency. diff --git a/packages/cli/tests/46-daemon-terminal-worker-upgrade.test.ts b/packages/cli/tests/46-daemon-terminal-worker-upgrade.test.ts index d6c3d7f3b7..9090456406 100644 --- a/packages/cli/tests/46-daemon-terminal-worker-upgrade.test.ts +++ b/packages/cli/tests/46-daemon-terminal-worker-upgrade.test.ts @@ -34,7 +34,7 @@ import { import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { getAvailablePort } from "./helpers/network.ts"; +import { createServer } from "node:net"; if (process.platform === "win32") { console.log("Skipping terminal worker upgrade CLI test on Windows (POSIX shell terminals)."); @@ -49,7 +49,6 @@ const paseoHome = join(root, "home"); const cliHome = join(root, "os-home"); const workDir = join(root, "work"); for (const dir of [paseoHome, cliHome, workDir]) mkdirSync(dir, { recursive: true }); -const port = await getAvailablePort(); const currentProtocolVersion = 5; const zsh = ["/bin/zsh", "/usr/bin/zsh"].find((candidate) => existsSync(candidate)) ?? null; @@ -336,7 +335,18 @@ async function terminalIds(installation: Installation): Promise { } async function startDaemon(installation: Installation, extraEnv: Record = {}) { - await paseo(installation, ["daemon", "start"], extraEnv); + const result = await paseo(installation, ["daemon", "start", "--json"], extraEnv); + lastListen = (JSON.parse(result.stdout) as { listen: string }).listen; + assert.match(lastListen, /^127\.0\.0\.1:[1-9]\d*$/, "start publishes the actual bound port"); + if (competingListener.listening) { + const competitor = competingListener.address(); + assert(competitor && typeof competitor !== "string"); + assert.notEqual( + Number(lastListen.split(":").at(-1)), + competitor.port, + "each daemon binds its own endpoint while the released port belongs to a peer", + ); + } } async function stopDaemon(installation: Installation, keepTerminals = false) { @@ -371,10 +381,14 @@ const a = createInstallation("A", "0.98.0"); const b = createInstallation("B", "0.99.0"); let restoreA: (() => void) | null = null; let passed = false; +let lastListen = ""; +const competingListener = createServer(); try { for (const [field, value] of [ - ["daemon.listen", `127.0.0.1:${port}`], + // Let the listening daemon reserve its port atomically. Clients discover the + // actual endpoint from this home's PID record after every start. + ["daemon.listen", "127.0.0.1:0"], ["daemon.relay.enabled", "false"], ["features.dictation.enabled", "false"], ["features.voiceMode.enabled", "false"], @@ -433,6 +447,15 @@ try { restoreA = null; console.log("✓ closed and cleaned up\n"); + // A released endpoint belongs to nobody. Reproduce another process taking it + // between starts; worker adoption is keyed by the home, not by a stable TCP port. + const releasedPort = Number(lastListen.split(":").at(-1)); + await new Promise((resolve, reject) => { + competingListener.once("error", reject); + competingListener.listen(releasedPort, "127.0.0.1", resolve); + }); + console.log(`Competing listener owns released port ${releasedPort}`); + // There is no supported previous wire at protocol 5. Future previous-version classification // is covered by terminal-worker-protocol.test.ts; do not simulate an unreleased drain adapter. for (const protocolVersion of [0, 1, 2, 3, 4]) { @@ -514,6 +537,17 @@ try { } } } + if (competingListener.listening) { + await new Promise((resolve, reject) => + competingListener.close((error) => { + if (error) { + reject(error); + return; + } + resolve(); + }), + ); + } restoreA?.(); for (const installation of [b, a]) { await run(installation, ["daemon", "stop", "--home", paseoHome, "--force"]); diff --git a/packages/server/src/terminal/terminal-restore.test.ts b/packages/server/src/terminal/terminal-restore.test.ts index ab222ffc13..868b7fe155 100644 --- a/packages/server/src/terminal/terminal-restore.test.ts +++ b/packages/server/src/terminal/terminal-restore.test.ts @@ -26,12 +26,14 @@ describe("terminal restore policy", () => { ).toEqual({ scrollbackLines: 500 }); }); - test("promotes live restore to visible restore after output overflow", () => { + test("bounds pressure recovery for every restore mode", () => { expect(resolveRestoreAfterOutputOverflow({ mode: "live" })).toEqual({ mode: "visible-snapshot", + scrollbackLines: 500, }); expect(resolveRestoreAfterOutputOverflow({ mode: "full-snapshot" })).toEqual({ - mode: "full-snapshot", + mode: "visible-snapshot", + scrollbackLines: 500, }); }); diff --git a/packages/server/src/terminal/terminal-restore.ts b/packages/server/src/terminal/terminal-restore.ts index 0533adfa9c..4320f450ee 100644 --- a/packages/server/src/terminal/terminal-restore.ts +++ b/packages/server/src/terminal/terminal-restore.ts @@ -23,11 +23,14 @@ export type TerminalRestoreOptions = NonNullable { getClientBufferedAmount: () => number | null, chunked = false, pace?: () => Promise, + restore?: TerminalRestoreOptions, ): Promise<{ messages: SessionOutboundMessage[]; pushOutput: (data: string) => void; frames: TerminalStreamFrame[]; + replayOptions: Array; + ready: () => void; }> { let terminalListener: ((message: ServerMessage) => void) | null = null; const terminal: TerminalSession = { @@ -712,12 +717,16 @@ describe("terminal-session-controller backpressure snapshot fallback", () => { kill: vi.fn(), killAndWait: vi.fn(), }; + const replayOptions: Array = []; const terminalManager: TerminalManager = { getTerminals: vi.fn(), createTerminal: vi.fn(), registerCwdEnv: vi.fn(), getTerminal: vi.fn(() => terminal), - replayTerminal: replayFixture(async () => terminalReplay("live", 1)), + replayTerminal: (id, options, sink) => { + replayOptions.push(options); + return replayFixture(async () => terminalReplay("live", 1))(id, options, sink); + }, setTerminalTitle: vi.fn(), killTerminal: vi.fn(), killTerminalAndWait: vi.fn(), @@ -757,6 +766,7 @@ describe("terminal-session-controller backpressure snapshot fallback", () => { type: "subscribe_terminal_request", terminalId: "term-1", requestId: "req-1", + restore, }); await waitForCoalescerFlush(); // Drop the initial subscribe snapshot frame so each test only sees frames @@ -764,12 +774,44 @@ describe("terminal-session-controller backpressure snapshot fallback", () => { frames.length = 0; return { + replayOptions, + ready: () => terminalListener?.({ type: "snapshotReady" }), pushOutput: (data) => terminalListener?.({ type: "output", data, revision: 2 }), frames, messages, }; } + test.each([false, true])( + "bounds recovery without changing attach demand (chunked=%s)", + async (chunked) => { + for (const [restore, history] of [ + [{ mode: "visible-snapshot", scrollbackLines: 10000 }, 500], + [{ mode: "full-snapshot" }, 500], + [{ mode: "visible-snapshot", scrollbackLines: 5 }, 5], + [{ mode: "visible-snapshot", scrollbackLines: 0 }, 0], + [{ mode: "full-snapshot", scrollbackLines: 0 }, 0], + [{ mode: "live", scrollbackLines: 0 }, 0], + [{ mode: "live" }, 500], + [undefined, 500], + ] as const) { + const fixture = await setup(() => 8 * 1024 * 1024, chunked, undefined, restore); + const initial = fixture.replayOptions.slice(); + fixture.pushOutput("pressure".repeat(40000)); + await waitForCoalescerFlush(); + expect(fixture.replayOptions.at(-1)).toEqual({ scrollbackLines: history }); + expect(fixture.replayOptions).toHaveLength(initial.length + 1); + fixture.ready(); // Worker adoption must use the original demand, even on the same stream. + await waitForCoalescerFlush(); + expect(fixture.replayOptions.slice(initial.length + 1)).toEqual(initial); + fixture.frames.length = 0; + fixture.pushOutput("after-catch-up"); + await waitForCoalescerFlush(); + expect(fixture.frames.map((frame) => frame.opcode)).toEqual([TerminalStreamOpcode.Output]); + } + }, + ); + test("streams all output without a snapshot when the client keeps up", async () => { const { pushOutput, frames } = await setup(() => 0); diff --git a/packages/server/src/terminal/terminal-session-controller.ts b/packages/server/src/terminal/terminal-session-controller.ts index 2cc548ba72..7c919839d1 100644 --- a/packages/server/src/terminal/terminal-session-controller.ts +++ b/packages/server/src/terminal/terminal-session-controller.ts @@ -929,7 +929,6 @@ export class TerminalSessionController { // The snapshot replaces this batch only after it succeeds. If the // terminal disappears during the read, completion still owns these bytes. activeStream.snapshotOutput = payload; - activeStream.restore = resolveRestoreAfterOutputOverflow(activeStream.restore); activeStream.needsSnapshot = true; void this.trySendSnapshot(activeStream); return; @@ -1029,7 +1028,12 @@ export class TerminalSessionController { activeStream.outputCoalescer.flush(); activeStream.snapshotInFlight = true; try { - const restore = activeStream.restore; + // A pressure catch-up replaces the skipped batch with a recent window. + // Keep the subscription demand intact for later adoption/full restores. + const restore = + activeStream.snapshotOutput !== undefined + ? resolveRestoreAfterOutputOverflow(activeStream.restore) + : activeStream.restore; const snapshotResult = restore ? await this.emitRestoreSnapshot(activeStream, terminalManager, restore) : await this.emitRestoreSnapshot(activeStream, terminalManager, { mode: "full-snapshot" }); diff --git a/packages/server/src/terminal/worker-terminal-manager.test.ts b/packages/server/src/terminal/worker-terminal-manager.test.ts index dadf8d398a..ad066327a2 100644 --- a/packages/server/src/terminal/worker-terminal-manager.test.ts +++ b/packages/server/src/terminal/worker-terminal-manager.test.ts @@ -1958,6 +1958,7 @@ function terminalViewers( terminalManager: DaemonTerminalManager, terminalId: string, chunked = true, + getClientBufferedAmount?: (source: object) => number | null, ) { const frames: Array<{ source: object; frame: TerminalStreamFrame }> = []; const messages: Array<{ source: object; message: SessionOutboundMessage }> = []; @@ -1990,6 +1991,7 @@ function terminalViewers( }, hasBinaryChannel: () => true, supportsChunkedReplay: () => chunked, + getClientBufferedAmount, isPathWithinRoot: () => true, sessionLogger: pino({ level: "silent" }), }); @@ -2354,6 +2356,8 @@ it.each([3000, 12050])( "retains ten thousand history rows and restores exact demands after %s printed rows", async (count) => { manager = await startRealWorkerManager(); + const gate = createTemporaryHome(); + const releaseNewline = join(gate, "newline"); const session = trackTerminal( await manager.createTerminal({ cwd: createTemporaryHome(), @@ -2361,7 +2365,14 @@ it.each([3000, 12050])( rows: 4, cols: 80, ...nodeTerminalCommand( - `for(let i=0;i<${count};i++) process.stdout.write("\\x1b[31mrow-"+String(i).padStart(5,"0")+" 界é🙂\\x1b[0m\\r\\n");setInterval(()=>{},1000)`, + `const fs=require("node:fs"); + const watcher=fs.watch(${JSON.stringify(gate)},()=>{ + if(fs.existsSync(${JSON.stringify(releaseNewline)})){ + watcher.close();process.stdout.write("\\r\\nROWS-COMPLETE"); + } + }); + for(let i=0;i<${count};i++) process.stdout.write("\\x1b[31mrow-"+String(i).padStart(5,"0")+" 界é🙂\\x1b[0m"+(i===${count}-1?"":"\\r\\n")); + setInterval(()=>{},1000)`, ), }), ); @@ -2372,6 +2383,14 @@ it.each([3000, 12050])( .includes("row-" + String(count - 1).padStart(5, "0")), 15000, ); + // A visible final label can precede its trailing CRLF. Hold that transition + // deliberately, then observe content on the following row before capture. + writeFileSync(releaseNewline, "release"); + await waitForCondition( + async () => + (await manager!.captureTerminal(session.id, { start: -1 })).lines[0] === "ROWS-COMPLETE", + 15000, + ); for (const history of [0, 200, 1000, 10000]) { const chunks: Uint8Array[] = []; const metadata = await manager.replayTerminal( @@ -2399,6 +2418,8 @@ it.each([3000, 12050])( const labels = Array.from({ length: buffer.length }, (_, i) => buffer.getLine(i)!.translateToString(true), ).filter((line) => line.includes("row-")); + expect({ row: buffer.cursorY, col: buffer.cursorX }).toEqual({ row: 3, col: 13 }); + expect(buffer.getLine(buffer.length - 1)!.translateToString(true)).toBe("ROWS-COMPLETE"); expect(labels).toHaveLength(Math.min(count, history + 3)); expect(labels[0]).toContain("row-" + String(count - labels.length).padStart(5, "0")); expect(labels.at(-1)).toContain("row-" + String(count - 1).padStart(5, "0") + " 界é🙂"); @@ -2409,3 +2430,170 @@ it.each([3000, 12050])( }, 30000, ); + +it.each([false, true])( + "shrinks real ten-thousand-row pressure recovery and preserves reattach (chunked=%s)", + async (chunked) => { + manager = await startRealWorkerManager(); + const session = trackTerminal( + await manager.createTerminal({ + cwd: createTemporaryHome(), + workspaceId: "pressure-history", + rows: 4, + cols: 120, + ...nodeTerminalCommand(` + process.stdin.setRawMode(true); + let row = 0; + function print(count) { + for(let i=0;i { + if(data.toString().includes("BURST")) { print(4000); process.stdout.write("BURST-DONE\\r\\n"); } + else process.stdout.write("ECHO:" + data.toString() + "\\r\\n"); + }); + setInterval(()=>{},1000); + `), + }), + ); + await waitForCondition( + async () => + (await manager!.captureTerminal(session.id, { start: -5 })).lines + .join("") + .includes("READY"), + 15000, + ); + const desktop = {}, + phone = {}, + fresh = {}; + let pressure = false; + const viewers = terminalViewers(manager, session.id, chunked, (source) => + source === phone && pressure ? 8 * 1024 * 1024 : 0, + ); + const restore = { mode: "full-snapshot" as const, scrollbackLines: 10000 }; + function payloads(source: object): Buffer[] { + const result: Buffer[] = []; + let chunks: Uint8Array[] = []; + for (const { source: owner, frame } of viewers.frames) { + if (owner !== source) continue; + if (frame.opcode === TerminalStreamOpcode.Restore) result.push(Buffer.from(frame.payload)); + if (frame.opcode === TerminalStreamOpcode.ReplayBegin) chunks = []; + if (frame.opcode === TerminalStreamOpcode.ReplayData) chunks.push(frame.payload); + if (frame.opcode === TerminalStreamOpcode.ReplayEnd) result.push(Buffer.concat(chunks)); + } + return result; + } + async function parse(payload: Uint8Array) { + const receiver = new xterm.Terminal({ + rows: 4, + cols: 120, + scrollback: 10000, + allowProposedApi: true, + }); + try { + await new Promise((resolve) => receiver.write(payload, resolve)); + const buffer = receiver.buffer.active; + return { + history: buffer.baseY, + text: Array.from({ length: buffer.length }, (_, i) => + buffer.getLine(i)!.translateToString(true), + ).join("\n"), + }; + } finally { + receiver.dispose(); + } + } + try { + await viewers.attach(desktop, "healthy", restore); + await viewers.attach(phone, "pressured", restore); + await waitForCondition( + () => payloads(phone).length === 1 && payloads(desktop).length === 1, + 15000, + ); + const initial = payloads(phone)[0]!; + expect((await parse(initial)).history).toBe(10000); + pressure = true; + session.send({ type: "input", data: "BURST" }); + await waitForCondition(() => payloads(phone).length >= 2, 15000); + pressure = false; + // The final burst marker may belong to the replay cut or to its live tail. + // Read the same replay/output exchange a renderer sees rather than assuming + // worker parsing and daemon request scheduling cut the burst at one point. + async function currentView(source: object) { + const frames = viewers.frames + .filter((row) => row.source === source) + .map((row) => row.frame); + const start = frames.findLastIndex( + (frame) => + frame.opcode === TerminalStreamOpcode.Restore || + frame.opcode === TerminalStreamOpcode.ReplayBegin, + ); + return parse( + Buffer.concat( + frames + .slice(start) + .filter( + (frame) => + frame.opcode === TerminalStreamOpcode.Restore || + frame.opcode === TerminalStreamOpcode.ReplayData || + frame.opcode === TerminalStreamOpcode.Output, + ) + .map((frame) => frame.payload), + ), + ); + } + await waitForCondition( + async () => + (await currentView(phone)).text.includes("BURST-DONE") && + (await currentView(desktop)).text.includes("BURST-DONE"), + 15000, + ); + const recovery = payloads(phone)[1]!; + const recovered = await parse(recovery); + expect(recovered.history).toBe(500); + expect(recovered.text).toContain("界é🙂"); + for (const source of [phone, desktop]) { + const indexes = [...(await currentView(source)).text.matchAll(/row-(\d+)/g)].map((match) => + Number(match[1]), + ); + expect(indexes.at(-1)).toBe(16049); + expect(indexes).toEqual(Array.from({ length: indexes.length }, (_, i) => indexes[0]! + i)); + } + expect(recovery.byteLength).toBeLessThan(initial.byteLength / 10); + expect(recovery.byteLength).toBeLessThan(256 * 1024); + expect(payloads(desktop)).toHaveLength(1); // The healthy peer keeps its full replay and live stream. + session.send({ type: "input", data: "AFTER-CATCH-UP" }); + await waitForCondition( + () => + viewers.text(phone).includes("ECHO:AFTER-CATCH-UP") && + viewers.text(desktop).includes("ECHO:AFTER-CATCH-UP"), + 10000, + ); + expect(viewers.text(phone).match(/ECHO:AFTER-CATCH-UP/g)).toHaveLength(1); + expect(viewers.text(desktop).match(/ECHO:AFTER-CATCH-UP/g)).toHaveLength(1); + await viewers.delivery.detach(phone); + await viewers.attach(fresh, "reattach", restore); + await waitForCondition(() => payloads(fresh).length === 1, 15000); + const reattached = await parse(payloads(fresh)[0]!); + expect(reattached.history).toBe(10000); + expect(reattached.text).toContain("row-16049"); + expect(reattached.text).toContain("ECHO:AFTER-CATCH-UP"); + expect(viewers.ended(desktop)).toHaveLength(0); + expect(viewers.ended(phone)).toHaveLength(0); + process.stdout.write( + JSON.stringify({ + chunked, + initialBytes: initial.byteLength, + recoveryBytes: recovery.byteLength, + reattachBytes: payloads(fresh)[0]!.byteLength, + retainedHistory: reattached.history, + recoveryHistory: recovered.history, + }) + "\n", + ); + } finally { + await viewers.delivery.close(); + } + }, + 45000, +);