diff --git a/src/main/terminal/local-terminal-session.test.ts b/src/main/terminal/local-terminal-session.test.ts index ec3ddf54..5e9f271c 100644 --- a/src/main/terminal/local-terminal-session.test.ts +++ b/src/main/terminal/local-terminal-session.test.ts @@ -258,30 +258,215 @@ describe('LocalTerminalSession', () => { }) it('pauses at the pending-event boundary and resumes after ACK', async () => { - const pty = new FakePty() - const session = await LocalTerminalSession.create({ - ...baseOptions, - dependencies: sessionDependencies(pty) - }) - let lastSequence = 0 - session.onEvent((event) => { - lastSequence = event.sequence - }) + vi.useFakeTimers() + try { + const pty = new FakePty() + const session = await LocalTerminalSession.create({ + ...baseOptions, + dependencies: sessionDependencies(pty) + }) + let lastSequence = 0 + session.onEvent((event) => { + lastSequence = event.sequence + }) - for ( - let index = 0; - index < TERMINAL_LIMITS.maximumPendingEvents; - index += 1 - ) { - pty.emitData('x') - if (pty.pause.mock.calls.length > 0) { - break + for ( + let index = 0; + index < TERMINAL_LIMITS.maximumPendingEvents; + index += 1 + ) { + pty.emitData('x') + // Each flush window yields one event. + vi.advanceTimersByTime(20) + if (pty.pause.mock.calls.length > 0) { + break + } } - } - expect(pty.pause).toHaveBeenCalledTimes(1) - session.acknowledge(lastSequence) - expect(pty.resume).toHaveBeenCalledTimes(1) + expect(pty.pause).toHaveBeenCalledTimes(1) + expect(session.snapshot().state).toBe('running') + session.acknowledge(lastSequence) + expect(pty.resume).toHaveBeenCalledTimes(1) + } finally { + vi.useRealTimers() + } + }) + + it('coalesces bursts into few ordered output events', async () => { + vi.useFakeTimers() + try { + const pty = new FakePty() + const session = await LocalTerminalSession.create({ + ...baseOptions, + dependencies: sessionDependencies(pty) + }) + const events: TerminalEvent[] = [] + session.onEvent((event) => events.push(event)) + const expected: string[] = [] + for (let index = 0; index < 1_000; index += 1) { + const chunk = `line ${index} 你好\r\n` + expected.push(chunk) + pty.emitData(chunk) + if (index % 100 === 0) { + vi.advanceTimersByTime(8) + } + } + vi.advanceTimersByTime(20) + + const output = events.filter((event) => event.type === 'output') + expect(output.map((event) => event.data).join('')).toBe(expected.join('')) + expect(output.length).toBeLessThan(25) + const sequences = events.map((event) => event.sequence) + expect(sequences).toEqual([...sequences].sort((a, b) => a - b)) + expect(new Set(sequences).size).toBe(sequences.length) + await session.close() + } finally { + vi.useRealTimers() + } + }) + + it('flushes buffered output before exit events and stops its timer', async () => { + vi.useFakeTimers() + try { + const pty = new FakePty() + const session = await LocalTerminalSession.create({ + ...baseOptions, + dependencies: sessionDependencies(pty) + }) + const events: TerminalEvent[] = [] + session.onEvent((event) => events.push(event)) + pty.emitData('first ') + pty.emitData('trailing output') + pty.emitData(' tail') + pty.emitExit(0) + + expect( + events.slice(2).map((event) => + event.type === 'output' ? event.data : event.type + ) + ).toEqual(['first ', 'trailing output tail', 'exit', 'state']) + expect(vi.getTimerCount()).toBe(0) + vi.advanceTimersByTime(100) + expect(events).toHaveLength(6) + } finally { + vi.useRealTimers() + } + }) + + it('flushes buffered output before the closing state on kill', async () => { + vi.useFakeTimers() + try { + const pty = new FakePty() + const session = await LocalTerminalSession.create({ + ...baseOptions, + dependencies: sessionDependencies(pty, { + terminate: vi.fn(async () => undefined) + }) + }) + const events: TerminalEvent[] = [] + session.onEvent((event) => events.push(event)) + pty.emitData('a') + pty.emitData('b') + await session.close() + + expect( + events.slice(2).map((event) => + event.type === 'output' + ? event.data + : event.type === 'state' + ? event.state + : event.type + ) + ).toEqual(['a', 'b', 'closing', 'exit', 'exited']) + expect(vi.getTimerCount()).toBe(0) + } finally { + vi.useRealTimers() + } + }) + + it('counts buffered and in-flight bytes for backpressure and keeps them bounded', async () => { + vi.useFakeTimers() + try { + const pty = new FakePty() + const session = await LocalTerminalSession.create({ + ...baseOptions, + dependencies: sessionDependencies(pty) + }) + const outputs: Extract[] = [] + session.onEvent((event) => { + if (event.type === 'output') { + outputs.push(event) + } + }) + const block = 'q'.repeat(16 * 1024) + let produced = 0 + // A renderer that never ACKs: output must pause well before the hard + // 4 MiB bound and never exceed it. + while (pty.pause.mock.calls.length === 0) { + pty.emitData(block) + produced += block.length + expect(produced).toBeLessThan(TERMINAL_LIMITS.maximumBufferedOutputBytes) + } + expect(session.snapshot().state).toBe('running') + const emittedBytes = outputs.reduce((sum, event) => sum + event.data.length, 0) + expect(produced - emittedBytes).toBeLessThan(TERMINAL_LIMITS.maximumEventBytes) + expect( + outputs.every( + (event) => Buffer.byteLength(event.data) <= TERMINAL_LIMITS.maximumEventBytes + ) + ).toBe(true) + + // In-flight data after pause is absorbed by the headroom. + for (let index = 0; index < 16; index += 1) { + pty.emitData(block) + produced += block.length + } + vi.advanceTimersByTime(20) + expect(session.snapshot().state).toBe('running') + + // A partial ACK that leaves more than half the budget pending must not + // resume; ACKing the rest must. + session.acknowledge(outputs[0]!.sequence) + expect(pty.resume).not.toHaveBeenCalled() + session.acknowledge(outputs.at(-1)!.sequence) + expect(pty.resume).toHaveBeenCalledTimes(1) + expect( + outputs.map((event) => event.data).join('').length + ).toBe(produced) + await session.close() + } finally { + vi.useRealTimers() + } + }) + + it('fails with output-limit-exceeded when the renderer never ACKs and the PTY ignores pause', async () => { + vi.useFakeTimers() + try { + const pty = new FakePty() + const session = await LocalTerminalSession.create({ + ...baseOptions, + dependencies: sessionDependencies(pty, { + terminate: vi.fn(async () => undefined) + }) + }) + const block = 'q'.repeat(64 * 1024) + for ( + let produced = 0; + produced <= 2 * TERMINAL_LIMITS.maximumBufferedOutputBytes && + session.snapshot().state === 'running'; + produced += block.length + ) { + pty.emitData(block) + } + expect(session.snapshot()).toMatchObject({ + state: 'failed', + error: { code: 'output-limit-exceeded' } + }) + expect(vi.getTimerCount()).toBe(0) + await session.close() + } finally { + vi.useRealTimers() + } }) it('reports launch failures with stable failed state', async () => { diff --git a/src/main/terminal/local-terminal-session.ts b/src/main/terminal/local-terminal-session.ts index 2264ff0d..e7c775fa 100644 --- a/src/main/terminal/local-terminal-session.ts +++ b/src/main/terminal/local-terminal-session.ts @@ -16,6 +16,10 @@ import { } from '../../shared/terminal-contracts' import { requestProcessTreeTermination } from '../agent/child-process-termination' import { buildCredentialFilteredUserEnvironment } from '../agent/process-environment' +import { + TerminalOutputCoalescer, + type TerminalOutputCoalescerOptions +} from './terminal-output-coalescer' export type LocalTerminalTarget = Extract< TerminalTarget, @@ -52,6 +56,7 @@ export type LocalTerminalDependencies = { directoryExists?: (path: string) => Promise spawn?: LocalPtySpawn terminate?: (pty: IPty, platform: NodeJS.Platform) => Promise + outputCoalescing?: Omit } export type LocalTerminalSessionOptions = { @@ -72,11 +77,25 @@ type UnsequencedTerminalEvent = T extends TerminalEvent type LocalTerminalEvent = UnsequencedTerminalEvent const CONTROL_EVENT_RESERVE = 2 -const PAUSE_PENDING_EVENT_COUNT = +/** Hard limits: exceeding them fails the session (renderer stopped ACKing). */ +const MAXIMUM_PENDING_OUTPUT_EVENTS = TERMINAL_LIMITS.maximumPendingEvents - CONTROL_EVENT_RESERVE -const RESUME_PENDING_EVENT_COUNT = Math.floor( - PAUSE_PENDING_EVENT_COUNT / 2 +/** + * Soft limits: pause the PTY with headroom below the hard limits, because a + * paused PTY may still deliver data that was already in flight. Coalesced + * events are up to 64 KiB, so the byte limit is normally reached first. + */ +const PAUSE_PENDING_EVENT_COUNT = Math.floor( + (MAXIMUM_PENDING_OUTPUT_EVENTS * 3) / 4 ) +const PAUSE_PENDING_OUTPUT_BYTES = Math.floor( + (TERMINAL_LIMITS.maximumBufferedOutputBytes * 3) / 4 +) +const RESUME_PENDING_EVENT_COUNT = Math.floor( + MAXIMUM_PENDING_OUTPUT_EVENTS / 2 +) +const RESUME_PENDING_OUTPUT_BYTES = + TERMINAL_LIMITS.maximumBufferedOutputBytes / 2 const CLOSE_WAIT_MS = 1_500 const FORCE_WAIT_MS = 250 @@ -248,32 +267,6 @@ export async function resolveLocalTerminalLaunch( } } -function splitBoundedOutput(data: string): string[] { - const encoded = Buffer.from(data) - const chunks: string[] = [] - let offset = 0 - while (offset < encoded.byteLength) { - let end = Math.min( - offset + TERMINAL_LIMITS.maximumEventBytes, - encoded.byteLength - ) - if (end < encoded.byteLength) { - while ( - end > offset && - (encoded[end]! & 0xc0) === 0x80 - ) { - end -= 1 - } - } - if (end === offset) { - end = Math.min(offset + 1, encoded.byteLength) - } - chunks.push(encoded.subarray(offset, end).toString('utf8')) - offset = end - } - return chunks -} - async function waitUntil( predicate: () => boolean, milliseconds: number @@ -369,12 +362,17 @@ export class LocalTerminalSession { private exit: TerminalExit | null = null private error: TerminalError | null = null private closePromise: Promise | undefined + private readonly output: TerminalOutputCoalescer private constructor(options: LocalTerminalSessionOptions) { this.options = options this.dependencies = options.dependencies ?? {} this.sessionId = options.sessionId ?? randomUUID() this.size = { ...options.size } + this.output = new TerminalOutputCoalescer({ + ...this.dependencies.outputCoalescing, + emit: (data) => this.emitOutput(data) + }) this.emit({ type: 'state', state: 'starting' }) } @@ -487,8 +485,8 @@ export class LocalTerminalSession { if ( this.paused && this.pendingEvents.length <= RESUME_PENDING_EVENT_COUNT && - this.pendingOutputBytes <= - TERMINAL_LIMITS.maximumBufferedOutputBytes / 2 + this.pendingOutputBytes + this.output.bufferedBytes <= + RESUME_PENDING_OUTPUT_BYTES ) { this.paused = false this.pty?.resume() @@ -545,31 +543,39 @@ export class LocalTerminalSession { if (this.state !== 'running' || data.length === 0) { return } - for (const chunk of splitBoundedOutput(data)) { - if ( - this.pendingEvents.length >= PAUSE_PENDING_EVENT_COUNT || - this.pendingOutputBytes + Buffer.byteLength(chunk) > - TERMINAL_LIMITS.maximumBufferedOutputBytes - ) { - this.fail( - 'output-limit-exceeded', - 'Terminal output exceeded the bounded Main process buffer' - ) - void this.close() - return - } - this.pendingOutputBytes += Buffer.byteLength(chunk) - this.emit({ type: 'output', data: chunk }) - this.updateBackpressure() + // Bytes count against backpressure as soon as they are buffered, so a slow + // renderer pauses the PTY regardless of the coalescing window. + this.output.push(data) + this.updateBackpressure() + } + + /** Called by the coalescer with ordered, bounded (<= 64 KiB) chunks. */ + private emitOutput(chunk: string): void { + const bytes = Buffer.byteLength(chunk) + if ( + this.pendingEvents.length >= MAXIMUM_PENDING_OUTPUT_EVENTS || + this.pendingOutputBytes + bytes > + TERMINAL_LIMITS.maximumBufferedOutputBytes + ) { + this.fail( + 'output-limit-exceeded', + 'Terminal output exceeded the bounded Main process buffer' + ) + void this.close() + return } + this.pendingOutputBytes += bytes + this.emit({ type: 'output', data: chunk }) + this.updateBackpressure() } private updateBackpressure(): void { if ( !this.paused && + this.state === 'running' && (this.pendingEvents.length >= PAUSE_PENDING_EVENT_COUNT || - this.pendingOutputBytes >= - TERMINAL_LIMITS.maximumBufferedOutputBytes) + this.pendingOutputBytes + this.output.bufferedBytes >= + PAUSE_PENDING_OUTPUT_BYTES) ) { this.paused = true this.pty?.pause() @@ -599,8 +605,10 @@ export class LocalTerminalSession { this.exit.signal = 'unknown' } this.state = 'exited' + // emit() flushes buffered output first, so trailing output precedes exit. this.emit({ type: 'exit', exit: this.exit }) this.emit({ type: 'state', state: 'exited' }) + this.output.dispose() } private fail( @@ -618,6 +626,9 @@ export class LocalTerminalSession { retryable: code === 'launch-failed' } this.state = 'failed' + // Failure is terminal: drop buffered output (it is what exceeded the + // bound) and cancel the flush timer before announcing the error. + this.output.dispose() this.emit({ type: 'error', error: this.error }) this.emit({ type: 'state', state: 'failed' }) } @@ -625,6 +636,10 @@ export class LocalTerminalSession { private emit( event: LocalTerminalEvent ): void { + if (event.type !== 'output') { + // Control events must follow every byte the PTY produced before them. + this.output.flush() + } if (this.lastSequence >= Number.MAX_SAFE_INTEGER) { this.state = 'failed' this.error = { diff --git a/src/main/terminal/ssh-terminal-session.test.ts b/src/main/terminal/ssh-terminal-session.test.ts index 45f2892a..350b09d0 100644 --- a/src/main/terminal/ssh-terminal-session.test.ts +++ b/src/main/terminal/ssh-terminal-session.test.ts @@ -185,21 +185,83 @@ describe("SSH terminal session", () => { }); it("emits ordered bounded output and pauses until acknowledged", async () => { - const harness = createHarness(); - const session = await openSession(harness); + vi.useFakeTimers(); + try { + const harness = createHarness(); + const session = await openSession(harness); - for (let index = 0; index < 255; index += 1) { - harness.channel.emit("data", `chunk-${index}`); + for (let index = 0; index < 255; index += 1) { + harness.channel.emit("data", `chunk-${index}`); + // One flush window per chunk keeps one event per chunk. + vi.advanceTimersByTime(20); + } + + expect(harness.events).toHaveLength(256); + expect(harness.events.map((event) => event.sequence)).toEqual( + Array.from({ length: 256 }, (_, index) => index + 1), + ); + expect(harness.channel.pause).toHaveBeenCalledOnce(); + session.acknowledge(256); + expect(harness.channel.resume).toHaveBeenCalledOnce(); + session.close(); + } finally { + vi.useRealTimers(); } + }); - expect(harness.events).toHaveLength(256); - expect(harness.events.map((event) => event.sequence)).toEqual( - Array.from({ length: 256 }, (_, index) => index + 1), - ); - expect(harness.channel.pause).toHaveBeenCalledOnce(); - session.acknowledge(256); - expect(harness.channel.resume).toHaveBeenCalledOnce(); - session.close(); + it("coalesces bursts in order and flushes trailing output before exit", async () => { + vi.useFakeTimers(); + try { + const harness = createHarness(); + await openSession(harness); + const expected: string[] = []; + for (let index = 0; index < 500; index += 1) { + const chunk = `chunk-${index};`; + expected.push(chunk); + harness.channel.emit("data", Buffer.from(chunk)); + } + // Split multi-byte character across two SSH packets. + const bytes = Buffer.from("你"); + harness.channel.emit("data", bytes.subarray(0, 1)); + harness.channel.emit("data", bytes.subarray(1)); + expected.push("你"); + harness.channel.emit("exit", 0); + harness.channel.emit("close"); + + const output = harness.events.filter((event) => event.type === "output"); + expect(output.map((event) => event.data).join("")).toBe(expected.join("")); + expect(output.length).toBeLessThanOrEqual(2); + expect(harness.events.slice(-2).map((event) => event.type)).toEqual([ + "exit", + "state", + ]); + const count = harness.events.length; + vi.advanceTimersByTime(100); + expect(harness.events).toHaveLength(count); + } finally { + vi.useRealTimers(); + } + }); + + it("flushes buffered output before closing on kill", async () => { + vi.useFakeTimers(); + try { + const harness = createHarness(); + const session = await openSession(harness); + harness.channel.emit("data", "a"); + harness.channel.emit("data", "b"); + session.close(); + expect( + harness.events.slice(1).map((event) => + event.type === "output" ? event.data : event.type === "state" ? event.state : event.type, + ), + ).toEqual(["a", "b", "closing", "exit", "exited"]); + const count = harness.events.length; + vi.advanceTimersByTime(100); + expect(harness.events).toHaveLength(count); + } finally { + vi.useRealTimers(); + } }); it("marks a disconnected SSH transport interrupted immediately and rejects late I/O", async () => { diff --git a/src/main/terminal/ssh-terminal-session.ts b/src/main/terminal/ssh-terminal-session.ts index 94c03c3f..6addc204 100644 --- a/src/main/terminal/ssh-terminal-session.ts +++ b/src/main/terminal/ssh-terminal-session.ts @@ -16,6 +16,10 @@ import { type SshConnectionPoolTarget, type SshTerminalConnectionLease, } from "../ssh/ssh-connection-pool"; +import { + TerminalOutputCoalescer, + type TerminalOutputCoalescerOptions, +} from "./terminal-output-coalescer"; const SSH_TERMINAL_TERM = "xterm-256color"; @@ -71,29 +75,13 @@ function eventByteLength(event: TerminalEvent): number { } } -function splitUtf8(text: string, maximumBytes: number): string[] { - const chunks: string[] = []; - let chunk = ""; - let chunkBytes = 0; - for (const character of text) { - const characterBytes = Buffer.byteLength(character); - if (chunk && chunkBytes + characterBytes > maximumBytes) { - chunks.push(chunk); - chunk = ""; - chunkBytes = 0; - } - chunk += character; - chunkBytes += characterBytes; - } - if (chunk) { - chunks.push(chunk); - } - return chunks; -} +const PAUSE_PENDING_BYTES = + TERMINAL_LIMITS.maximumBufferedOutputBytes - TERMINAL_LIMITS.maximumEventBytes; export async function createSshTerminalSession( pool: SshConnectionPool, options: SshTerminalSessionOptions, + coalescing?: Omit, ): Promise { const initialSize = terminalSizeSchema.parse(options.size); const workingDirectoryCommand = options.workingDirectory && !options.command @@ -128,7 +116,22 @@ export async function createSshTerminalSession( const decoder = new StringDecoder("utf8"); let running: RunningSession | undefined; + const updateBackpressure = (): void => { + if ( + !paused && + (pendingEvents.length >= TERMINAL_LIMITS.maximumPendingEvents || + pendingBytes + output.bufferedBytes >= PAUSE_PENDING_BYTES) + ) { + paused = true; + channel.pause(); + } + }; + const emit = (event: TerminalEventPayload): void => { + if (event.type !== "output") { + // Control events must follow every byte received before them. + output.flush(); + } sequence += 1; const sequenced = { ...event, @@ -139,19 +142,18 @@ export async function createSshTerminalSession( pendingEvents.push({ event: sequenced, bytes }); pendingBytes += bytes; options.onEvent(sequenced); - if ( - !paused && - (pendingEvents.length >= TERMINAL_LIMITS.maximumPendingEvents || - pendingBytes >= - TERMINAL_LIMITS.maximumBufferedOutputBytes - - TERMINAL_LIMITS.maximumEventBytes) - ) { - paused = true; - channel.pause(); - } + updateBackpressure(); }; + const output = new TerminalOutputCoalescer({ + ...coalescing, + emit: (data) => emit({ type: "output", data }), + }); + const release = (): void => { + // Every terminal path calls release() after its final events; stop the + // flush timer and drop anything that can no longer be delivered. + output.dispose(); const active = running; running = undefined; active?.removeDisconnectListener(); @@ -183,12 +185,8 @@ export async function createSshTerminalSession( } const text = typeof chunk === "string" ? chunk : decoder.write(Buffer.from(chunk)); - for (const data of splitUtf8( - text, - TERMINAL_LIMITS.maximumEventBytes, - )) { - emit({ type: "output", data }); - } + output.push(text); + updateBackpressure(); }); channel.once( "exit", @@ -210,15 +208,8 @@ export async function createSshTerminalSession( release(); return; } - const finalText = decoder.end(); - if (finalText) { - for (const data of splitUtf8( - finalText, - TERMINAL_LIMITS.maximumEventBytes, - )) { - emit({ type: "output", data }); - } - } + output.push(decoder.end()); + output.flush(); if (!lease.isUsable()) { interrupt(); return; @@ -279,9 +270,7 @@ export async function createSshTerminalSession( if ( paused && pendingEvents.length < TERMINAL_LIMITS.maximumPendingEvents && - pendingBytes < - TERMINAL_LIMITS.maximumBufferedOutputBytes - - TERMINAL_LIMITS.maximumEventBytes + pendingBytes + output.bufferedBytes < PAUSE_PENDING_BYTES ) { paused = false; channel.resume(); diff --git a/src/main/terminal/terminal-output-coalescer.test.ts b/src/main/terminal/terminal-output-coalescer.test.ts new file mode 100644 index 00000000..54e0bed2 --- /dev/null +++ b/src/main/terminal/terminal-output-coalescer.test.ts @@ -0,0 +1,165 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { TERMINAL_LIMITS } from '../../shared/terminal-contracts' +import { + TERMINAL_OUTPUT_FLUSH_INTERVAL_MS, + TerminalOutputCoalescer, + splitUtf8Bounded +} from './terminal-output-coalescer' + +describe('TerminalOutputCoalescer', () => { + beforeEach(() => { + vi.useFakeTimers() + }) + afterEach(() => { + vi.useRealTimers() + }) + + it('emits the first chunk immediately and merges a burst into one flush', () => { + const emitted: string[] = [] + const coalescer = new TerminalOutputCoalescer({ + emit: (data) => emitted.push(data) + }) + + coalescer.push('$ ') + expect(emitted).toEqual(['$ ']) + for (let index = 0; index < 100; index += 1) { + coalescer.push(`${index},`) + } + expect(emitted).toHaveLength(1) + vi.advanceTimersByTime(TERMINAL_OUTPUT_FLUSH_INTERVAL_MS) + expect(emitted).toHaveLength(2) + expect(emitted.join('')).toBe( + '$ ' + Array.from({ length: 100 }, (_, index) => `${index},`).join('') + ) + coalescer.dispose() + }) + + it('preserves exact order across timer, size-cap and explicit flushes', () => { + const emitted: string[] = [] + const coalescer = new TerminalOutputCoalescer({ + emit: (data) => emitted.push(data), + maximumBatchBytes: 16 + }) + const expected: string[] = [] + for (let index = 0; index < 500; index += 1) { + const chunk = index % 7 === 0 ? `你好${index}` : `#${index};` + expected.push(chunk) + coalescer.push(chunk) + if (index % 13 === 0) { + vi.advanceTimersByTime(TERMINAL_OUTPUT_FLUSH_INTERVAL_MS) + } + if (index % 97 === 0) { + coalescer.flush() + } + } + coalescer.flush() + + expect(emitted.join('')).toBe(expected.join('')) + expect(emitted.every((chunk) => Buffer.byteLength(chunk) <= 16)).toBe(true) + expect(coalescer.bufferedBytes).toBe(0) + coalescer.dispose() + }) + + it('emits full batches at the size cap without waiting for the timer', () => { + const emitted: string[] = [] + const coalescer = new TerminalOutputCoalescer({ + emit: (data) => emitted.push(data) + }) + const cap = TERMINAL_LIMITS.maximumEventBytes + coalescer.push('x') + const block = 'y'.repeat(4096) + for (let index = 0; index < 40; index += 1) { + coalescer.push(block) + } + + // 160 KiB buffered inside one window: two full 64 KiB batches go out + // immediately; the remainder waits for the timer. + expect(emitted.slice(1).map((chunk) => Buffer.byteLength(chunk))).toEqual([ + cap, + cap + ]) + expect(coalescer.bufferedBytes).toBe(40 * 4096 - 2 * cap) + expect(coalescer.bufferedBytes).toBeLessThan(cap) + vi.advanceTimersByTime(TERMINAL_OUTPUT_FLUSH_INTERVAL_MS) + expect(coalescer.bufferedBytes).toBe(0) + expect(emitted.join('')).toBe('x' + block.repeat(40)) + coalescer.dispose() + }) + + it('never splits a multi-byte UTF-8 character across batches', () => { + const emitted: string[] = [] + const coalescer = new TerminalOutputCoalescer({ + emit: (data) => emitted.push(data), + maximumBatchBytes: 5 + }) + coalescer.push('a') + coalescer.push('你好世界🙂') + coalescer.flush() + expect(emitted.join('')).toBe('a你好世界🙂') + for (const chunk of emitted) { + expect(Buffer.from(chunk).toString('utf8')).toBe(chunk) + expect(chunk).not.toContain('\uFFFD') + } + coalescer.dispose() + }) + + it('dispose cancels the timer and drops later input', () => { + const emit = vi.fn() + const coalescer = new TerminalOutputCoalescer({ emit }) + coalescer.push('a') + coalescer.push('b') + coalescer.dispose() + expect(vi.getTimerCount()).toBe(0) + coalescer.push('c') + coalescer.flush() + vi.advanceTimersByTime(100) + expect(emit.mock.calls).toEqual([['a']]) + }) + + it('stops scheduling once idle', () => { + const coalescer = new TerminalOutputCoalescer({ emit: vi.fn() }) + coalescer.push('a') + vi.advanceTimersByTime(TERMINAL_OUTPUT_FLUSH_INTERVAL_MS) + expect(vi.getTimerCount()).toBe(0) + coalescer.dispose() + }) + + it('reduces IPC events by orders of magnitude for a 20 MB flood', () => { + // Model node-pty delivering ~4 KiB reads; compare the number of output + // events (one IPC send each) before and after coalescing. + const totalBytes = 20 * 1024 * 1024 + const readBytes = 4096 + const read = 'z'.repeat(readBytes) + let coalescedEvents = 0 + let coalescedBytes = 0 + const coalescer = new TerminalOutputCoalescer({ + emit: (data) => { + coalescedEvents += 1 + coalescedBytes += data.length + } + }) + const reads = totalBytes / readBytes + for (let index = 0; index < reads; index += 1) { + coalescer.push(read) + if (index % 64 === 0) { + vi.advanceTimersByTime(1) + } + } + coalescer.flush() + coalescer.dispose() + + expect(coalescedBytes).toBe(totalBytes) + // Uncoalesced: one event per read (5120). Coalesced: ~64 KiB per event. + expect(coalescedEvents).toBeLessThanOrEqual( + Math.ceil(totalBytes / TERMINAL_LIMITS.maximumEventBytes) + 100 + ) + expect(coalescedEvents).toBeLessThan(reads / 10) + }) +}) + +describe('splitUtf8Bounded', () => { + it('returns no chunks for empty input and a single chunk when small', () => { + expect(splitUtf8Bounded('', 4)).toEqual([]) + expect(splitUtf8Bounded('abc', 4)).toEqual(['abc']) + }) +}) diff --git a/src/main/terminal/terminal-output-coalescer.ts b/src/main/terminal/terminal-output-coalescer.ts new file mode 100644 index 00000000..5952bd6b --- /dev/null +++ b/src/main/terminal/terminal-output-coalescer.ts @@ -0,0 +1,161 @@ +import { TERMINAL_LIMITS } from '../../shared/terminal-contracts' + +/** + * Minimum spacing between timer-driven flushes. Output arriving while the + * coalescer is idle is flushed immediately (leading edge), so interactive + * echo gets no added latency; bursts inside the window are merged. + */ +export const TERMINAL_OUTPUT_FLUSH_INTERVAL_MS = 8 + +/** A full batch is flushed without waiting for the timer. */ +export const TERMINAL_OUTPUT_MAXIMUM_BATCH_BYTES = + TERMINAL_LIMITS.maximumEventBytes + +type TimerHandle = ReturnType + +export type TerminalOutputCoalescerOptions = { + /** Receives ordered chunks of at most `maximumBatchBytes` UTF-8 bytes. */ + emit: (data: string) => void + flushIntervalMs?: number + maximumBatchBytes?: number +} + +/** + * Splits text into ordered chunks of at most `maximumBytes` UTF-8 bytes without + * cutting a multi-byte sequence. + */ +export function splitUtf8Bounded( + data: string, + maximumBytes: number +): string[] { + const encoded = Buffer.from(data) + if (encoded.byteLength <= maximumBytes) { + return encoded.byteLength === 0 ? [] : [data] + } + const chunks: string[] = [] + let offset = 0 + while (offset < encoded.byteLength) { + let end = Math.min(offset + maximumBytes, encoded.byteLength) + if (end < encoded.byteLength) { + while (end > offset && (encoded[end]! & 0xc0) === 0x80) { + end -= 1 + } + } + if (end === offset) { + end = Math.min(offset + 1, encoded.byteLength) + } + chunks.push(encoded.subarray(offset, end).toString('utf8')) + offset = end + } + return chunks +} + +/** + * Per-terminal output buffer in Main. It never holds more than one batch + * (`maximumBatchBytes`), so a slow renderer is handled by the session's + * ACK-based backpressure rather than by growing this buffer. + */ +export class TerminalOutputCoalescer { + private readonly emitChunk: (data: string) => void + private readonly flushIntervalMs: number + private readonly maximumBatchBytes: number + private chunks: string[] = [] + private bytes = 0 + private timer: TimerHandle | undefined + private disposed = false + + constructor(options: TerminalOutputCoalescerOptions) { + this.emitChunk = options.emit + this.flushIntervalMs = + options.flushIntervalMs ?? TERMINAL_OUTPUT_FLUSH_INTERVAL_MS + this.maximumBatchBytes = + options.maximumBatchBytes ?? TERMINAL_OUTPUT_MAXIMUM_BATCH_BYTES + } + + /** UTF-8 bytes received but not yet emitted. */ + get bufferedBytes(): number { + return this.bytes + } + + get isDisposed(): boolean { + return this.disposed + } + + push(data: string): void { + if (this.disposed || data.length === 0) { + return + } + this.chunks.push(data) + this.bytes += Buffer.byteLength(data) + if (this.timer === undefined) { + this.flush() + this.startWindow() + return + } + if (this.bytes >= this.maximumBatchBytes) { + this.emitFullBatches() + } + } + + /** Emits everything buffered, in order. Safe to call re-entrantly. */ + flush(): void { + if (this.disposed || this.bytes === 0) { + return + } + const data = this.take() + for (const chunk of splitUtf8Bounded(data, this.maximumBatchBytes)) { + if (this.disposed) { + return + } + this.emitChunk(chunk) + } + } + + /** Drops buffered data and cancels the timer. Further input is ignored. */ + dispose(): void { + this.disposed = true + this.chunks = [] + this.bytes = 0 + if (this.timer !== undefined) { + clearTimeout(this.timer) + this.timer = undefined + } + } + + private emitFullBatches(): void { + const parts = splitUtf8Bounded(this.take(), this.maximumBatchBytes) + const last = parts.at(-1) + if (last !== undefined && Buffer.byteLength(last) < this.maximumBatchBytes) { + parts.pop() + this.chunks = [last] + this.bytes = Buffer.byteLength(last) + } + for (const chunk of parts) { + if (this.disposed) { + return + } + this.emitChunk(chunk) + } + } + + private take(): string { + const data = this.chunks.length === 1 ? this.chunks[0]! : this.chunks.join('') + this.chunks = [] + this.bytes = 0 + return data + } + + private startWindow(): void { + if (this.disposed) { + return + } + this.timer = setTimeout(() => { + this.timer = undefined + if (this.bytes > 0) { + this.flush() + this.startWindow() + } + }, this.flushIntervalMs) + this.timer.unref?.() + } +} diff --git a/src/renderer/src/TerminalPanel.test.tsx b/src/renderer/src/TerminalPanel.test.tsx index 098e19e1..8aff0f06 100644 --- a/src/renderer/src/TerminalPanel.test.tsx +++ b/src/renderer/src/TerminalPanel.test.tsx @@ -8,7 +8,9 @@ import type { } from '../../shared/terminal-contracts' import { TerminalPanel, + resetTerminalRendererWarningsForTests, resolveTerminalFontFamily, + warnTerminalRendererFallback, type TerminalAdapter, type TerminalClipboardAdapter, type TerminalEmulator, @@ -483,6 +485,120 @@ describe('TerminalPanel', () => { ) }) + it('batches output into one write per frame and ACKs after the write callback', async () => { + const frames = new Set<() => void>() + const frameScheduler = (callback: () => void): (() => void) => { + frames.add(callback) + return () => frames.delete(callback) + } + const runFrame = (): void => { + const current = [...frames] + frames.clear() + current.forEach((callback) => callback()) + } + const backend = adapterHarness() + const emulator = emulatorHarness() + const callbacks: Array<() => void> = [] + emulator.write.mockImplementation((_: string, callback?: () => void) => { + if (callback) { + callbacks.push(callback) + } + }) + render( + + ) + resizeObserver.trigger() + await waitFor(() => + expect(backend.adapter.create).toHaveBeenCalledOnce() + ) + + for (let sequence = 1; sequence <= 5; sequence += 1) { + backend.emit({ sessionId, sequence, type: 'output', data: `${sequence};` }) + } + await waitFor(() => expect(frames.size).toBe(1)) + expect(emulator.write).not.toHaveBeenCalled() + runFrame() + expect(emulator.write).toHaveBeenCalledOnce() + expect(emulator.write).toHaveBeenCalledWith('1;2;3;4;5;', expect.any(Function)) + expect(backend.adapter.ack).not.toHaveBeenCalled() + callbacks.shift()?.() + expect(backend.adapter.ack).toHaveBeenCalledOnce() + expect(backend.adapter.ack).toHaveBeenCalledWith({ sessionId, sequence: 5 }) + + // A control event flushes pending output first and is applied after it. + backend.emit({ sessionId, sequence: 6, type: 'output', data: 'bye' }) + backend.emit({ + sessionId, + sequence: 7, + type: 'exit', + exit: { exitCode: 0, signal: null } + }) + await waitFor(() => + expect(emulator.write).toHaveBeenLastCalledWith('bye', expect.any(Function)) + ) + expect(screen.queryByText('已退出')).not.toBeInTheDocument() + callbacks.shift()?.() + await waitFor(() => + expect(backend.adapter.ack).toHaveBeenLastCalledWith({ sessionId, sequence: 7 }) + ) + expect(backend.adapter.ack).toHaveBeenCalledWith({ sessionId, sequence: 6 }) + expect(emulator.write).toHaveBeenCalledTimes(2) + }) + + it('keeps writing output for a hidden terminal tab', async () => { + const backend = adapterHarness() + const emulator = emulatorHarness() + render( + + ) + await waitFor(() => + expect(backend.adapter.getSnapshot).toHaveBeenCalled() + ) + await screen.findByText('PowerShell 7') + backend.emit({ sessionId, sequence: 1, type: 'output', data: 'a' }) + backend.emit({ sessionId, sequence: 2, type: 'output', data: 'b' }) + await waitFor(() => + expect(emulator.write).toHaveBeenCalledWith('ab', expect.any(Function)) + ) + await waitFor(() => + expect(backend.adapter.ack).toHaveBeenCalledWith({ sessionId, sequence: 2 }) + ) + }) + + it('logs a bounded, content-free warning when the WebGL renderer falls back', () => { + resetTerminalRendererWarningsForTests() + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined) + try { + for (let index = 0; index < 20; index += 1) { + warnTerminalRendererFallback( + index % 2 === 0 ? 'webgl-load-failed' : 'webgl-context-lost', + new TypeError('secret user content') + ) + } + expect(warn).toHaveBeenCalledTimes(5) + expect(warn).toHaveBeenCalledWith( + '[terminal] webgl-load-failed; falling back to the DOM renderer', + { error: 'TypeError' } + ) + expect(JSON.stringify(warn.mock.calls)).not.toContain('secret') + } finally { + warn.mockRestore() + resetTerminalRendererWarningsForTests() + } + }) + it('uses the resolved cross-platform monospace font stack', () => { document.documentElement.style.setProperty( '--font-family-mono', diff --git a/src/renderer/src/TerminalPanel.tsx b/src/renderer/src/TerminalPanel.tsx index 63b43982..3b5f4cc9 100644 --- a/src/renderer/src/TerminalPanel.tsx +++ b/src/renderer/src/TerminalPanel.tsx @@ -28,6 +28,10 @@ import type { TerminalTarget } from '../../shared/terminal-contracts' import './terminal-panel.css' +import { + TerminalOutputBatcher, + type TerminalFrameScheduler +} from './terminal-output-batcher' export type TerminalAdapter = Omit & { subscribe: DesktopApi['terminal']['onEvent'] @@ -74,6 +78,8 @@ export type TerminalPanelProps = { sessionId?: TerminalSessionId title?: string terminalFactory?: TerminalFactory + /** Test seam for the per-frame output batching scheduler. */ + frameScheduler?: TerminalFrameScheduler onSessionChange?: (snapshot: TerminalSnapshot) => void onRename?: (title: string) => void } @@ -105,6 +111,31 @@ export function resolveTerminalFontFamily(): string { return configured || fallbackTerminalFontFamily } +/** Diagnostics are bounded per renderer so a flapping GPU cannot spam logs. */ +const MAXIMUM_RENDERER_WARNINGS = 5 +let rendererWarningCount = 0 + +export function warnTerminalRendererFallback( + reason: 'webgl-load-failed' | 'webgl-context-lost', + cause?: unknown +): void { + if (rendererWarningCount >= MAXIMUM_RENDERER_WARNINGS) { + return + } + rendererWarningCount += 1 + // Only the error class is logged: messages could echo environment details. + const errorName = + cause instanceof Error ? cause.name.slice(0, 64) : undefined + console.warn( + `[terminal] ${reason}; falling back to the DOM renderer`, + errorName ? { error: errorName } : {} + ) +} + +export function resetTerminalRendererWarningsForTests(): void { + rendererWarningCount = 0 +} + function createDefaultTerminal(): TerminalFactoryResult { const terminal = new Terminal({ cursorBlink: true, @@ -122,14 +153,22 @@ function createDefaultTerminal(): TerminalFactoryResult { fitAddon, searchAddon, activateRenderer: () => { - const webglAddon = new WebglAddon() + let webglAddon: WebglAddon + try { + webglAddon = new WebglAddon() + } catch (cause) { + warnTerminalRendererFallback('webgl-load-failed', cause) + return + } const contextLossSubscription = webglAddon.onContextLoss(() => { + warnTerminalRendererFallback('webgl-context-lost') contextLossSubscription.dispose() webglAddon.dispose() }) try { terminal.loadAddon(webglAddon) - } catch { + } catch (cause) { + warnTerminalRendererFallback('webgl-load-failed', cause) contextLossSubscription.dispose() webglAddon.dispose() } @@ -157,6 +196,7 @@ export function TerminalPanel({ sessionId, title, terminalFactory = createDefaultTerminal, + frameScheduler, onSessionChange, onRename }: TerminalPanelProps): React.JSX.Element { @@ -176,6 +216,9 @@ export function TerminalPanel({ const mountedRef = useRef(false) const acceptedSequenceRef = useRef(0) const eventQueueRef = useRef(Promise.resolve()) + const outputBatcherRef = useRef( + undefined + ) const lastSizeRef = useRef< { cols: number; rows: number } | undefined >(undefined) @@ -213,6 +256,17 @@ export function TerminalPanel({ } }, []) + const reportAckFailure = useCallback((reason: unknown): void => { + if (mountedRef.current) { + setError( + failureMessage( + reason, + translationRef.current('sidebar.terminal.errors.ack') + ) + ) + } + }, []) + const consumeEvent = useCallback( async (event: TerminalEvent): Promise => { if ( @@ -225,33 +279,42 @@ export function TerminalPanel({ acceptedSequenceRef.current = event.sequence if (event.type === 'output') { - await new Promise((resolve) => { - emulatorRef.current?.write(event.data, resolve) - if (!emulatorRef.current) { - resolve() - } - }) - } else { - const current = snapshotRef.current - if (current) { - if (event.type === 'state') { - publishSnapshot({ ...current, state: event.state }) - } else if (event.type === 'exit') { - publishSnapshot({ - ...current, - state: 'exited', - exit: event.exit, - error: null - }) - } else { - publishSnapshot({ - ...current, - state: 'failed', - error: event.error, - exit: null - }) - setError(event.error.message) - } + // Merged with neighbours and written once per frame; the batcher ACKs + // the batch's last sequence after xterm has processed it. + outputBatcherRef.current?.push( + event.sessionId, + event.sequence, + event.data + ) + return + } + // Control events apply only after all earlier output has been written. + await outputBatcherRef.current?.flush() + if ( + !mountedRef.current || + event.sessionId !== activeSessionRef.current + ) { + return + } + const current = snapshotRef.current + if (current) { + if (event.type === 'state') { + publishSnapshot({ ...current, state: event.state }) + } else if (event.type === 'exit') { + publishSnapshot({ + ...current, + state: 'exited', + exit: event.exit, + error: null + }) + } else { + publishSnapshot({ + ...current, + state: 'failed', + error: event.error, + exit: null + }) + setError(event.error.message) } } @@ -262,18 +325,11 @@ export function TerminalPanel({ sequence: event.sequence }) } catch (reason) { - if (mountedRef.current) { - setError( - failureMessage( - reason, - translationRef.current('sidebar.terminal.errors.ack') - ) - ) - } + reportAckFailure(reason) } } }, - [adapter, publishSnapshot] + [adapter, publishSnapshot, reportAckFailure] ) useEffect(() => { @@ -291,6 +347,19 @@ export function TerminalPanel({ terminal.open(hostRef.current) activateRenderer?.() } + const outputBatcher = new TerminalOutputBatcher({ + write: (data, done) => terminal.write(data, done), + acknowledge: (ackSessionId, sequence) => { + if (!mountedRef.current) { + return + } + adapter + .ack({ sessionId: ackSessionId, sequence }) + .catch(reportAckFailure) + }, + ...(frameScheduler ? { schedule: frameScheduler } : {}) + }) + outputBatcherRef.current = outputBatcher const inputSubscription = terminal.onData((data) => { const current = snapshotRef.current @@ -322,12 +391,14 @@ export function TerminalPanel({ operationGenerationRef.current += 1 unsubscribe() inputSubscription.dispose() + outputBatcher.dispose() + outputBatcherRef.current = undefined terminal.dispose() emulatorRef.current = undefined fitAddonRef.current = undefined searchAddonRef.current = undefined } - }, [adapter, consumeEvent, terminalFactory]) + }, [adapter, consumeEvent, frameScheduler, reportAckFailure, terminalFactory]) const attachSession = useCallback( async ( @@ -548,6 +619,7 @@ export function TerminalPanel({ if (!mountedRef.current) { return } + outputBatcherRef.current?.reset() emulatorRef.current?.clear() activeSessionRef.current = undefined acceptedSequenceRef.current = 0 diff --git a/src/renderer/src/terminal-output-batcher.test.ts b/src/renderer/src/terminal-output-batcher.test.ts new file mode 100644 index 00000000..a5f22195 --- /dev/null +++ b/src/renderer/src/terminal-output-batcher.test.ts @@ -0,0 +1,164 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { + TERMINAL_FRAME_FALLBACK_MS, + TerminalOutputBatcher, + scheduleTerminalFrame, + type TerminalFrameScheduler +} from './terminal-output-batcher' + +function manualScheduler(): { + schedule: TerminalFrameScheduler + runFrame: () => void + pendingFrames: () => number +} { + const callbacks = new Set<() => void>() + return { + schedule: (callback) => { + callbacks.add(callback) + return () => callbacks.delete(callback) + }, + runFrame: () => { + const current = [...callbacks] + callbacks.clear() + for (const callback of current) { + callback() + } + }, + pendingFrames: () => callbacks.size + } +} + +describe('TerminalOutputBatcher', () => { + afterEach(() => { + vi.useRealTimers() + vi.unstubAllGlobals() + }) + + it('writes once per frame in order and ACKs the last sequence after the write callback', async () => { + const frames = manualScheduler() + const writes: Array<{ data: string; done: () => void }> = [] + const acks: number[] = [] + const batcher = new TerminalOutputBatcher({ + write: (data, done) => writes.push({ data, done }), + acknowledge: (_, sequence) => acks.push(sequence), + schedule: frames.schedule + }) + batcher.push('s', 3, 'a') + batcher.push('s', 4, 'b') + batcher.push('s', 5, 'c') + expect(frames.pendingFrames()).toBe(1) + expect(writes).toHaveLength(0) + + frames.runFrame() + expect(writes.map((write) => write.data)).toEqual(['abc']) + expect(acks).toEqual([]) + writes[0]!.done() + expect(acks).toEqual([5]) + + batcher.push('s', 6, 'd') + frames.runFrame() + writes[1]!.done() + writes[1]!.done() + expect(writes.map((write) => write.data)).toEqual(['abc', 'd']) + expect(acks).toEqual([5, 6]) + batcher.dispose() + }) + + it('flush writes pending output immediately and waits for the emulator', async () => { + const frames = manualScheduler() + let release: (() => void) | undefined + const batcher = new TerminalOutputBatcher({ + write: (_, done) => { + release = done + }, + acknowledge: vi.fn(), + schedule: frames.schedule + }) + batcher.push('s', 1, 'x') + let flushed = false + const flush = batcher.flush().then(() => { + flushed = true + }) + expect(frames.pendingFrames()).toBe(0) + await Promise.resolve() + expect(flushed).toBe(false) + release?.() + await flush + expect(flushed).toBe(true) + batcher.dispose() + }) + + it('reset drops unwritten output but ACKs it so Main is not left paused', () => { + const frames = manualScheduler() + const write = vi.fn() + const acknowledge = vi.fn() + const batcher = new TerminalOutputBatcher({ + write, + acknowledge, + schedule: frames.schedule + }) + batcher.push('old', 9, 'stale') + batcher.reset() + frames.runFrame() + expect(write).not.toHaveBeenCalled() + expect(acknowledge).toHaveBeenCalledWith('old', 9) + batcher.dispose() + }) + + it('does not ACK after dispose', () => { + const frames = manualScheduler() + let done: (() => void) | undefined + const acknowledge = vi.fn() + const batcher = new TerminalOutputBatcher({ + write: (_, callback) => { + done = callback + }, + acknowledge, + schedule: frames.schedule + }) + batcher.push('s', 1, 'x') + frames.runFrame() + batcher.dispose() + done?.() + expect(acknowledge).not.toHaveBeenCalled() + }) + + it('keeps flushing hidden terminals via the timer fallback when frames stop', () => { + vi.useFakeTimers() + // Hidden/minimized windows never run animation frames. + vi.stubGlobal('requestAnimationFrame', vi.fn(() => 1)) + vi.stubGlobal('cancelAnimationFrame', vi.fn()) + const writes: string[] = [] + const acks: number[] = [] + const batcher = new TerminalOutputBatcher({ + write: (data, done) => { + writes.push(data) + done() + }, + acknowledge: (_, sequence) => acks.push(sequence) + }) + batcher.push('s', 1, 'hidden ') + batcher.push('s', 2, 'output') + vi.advanceTimersByTime(TERMINAL_FRAME_FALLBACK_MS) + expect(writes).toEqual(['hidden output']) + expect(acks).toEqual([2]) + batcher.dispose() + }) + + it('scheduleTerminalFrame runs once whichever of frame or timer fires first', () => { + vi.useFakeTimers() + let frame: (() => void) | undefined + vi.stubGlobal('requestAnimationFrame', vi.fn((callback: () => void) => { + frame = callback + return 7 + })) + const cancelAnimationFrame = vi.fn() + vi.stubGlobal('cancelAnimationFrame', cancelAnimationFrame) + const callback = vi.fn() + scheduleTerminalFrame(callback) + frame?.() + vi.advanceTimersByTime(TERMINAL_FRAME_FALLBACK_MS * 2) + expect(callback).toHaveBeenCalledOnce() + expect(vi.getTimerCount()).toBe(0) + }) +}) diff --git a/src/renderer/src/terminal-output-batcher.ts b/src/renderer/src/terminal-output-batcher.ts new file mode 100644 index 00000000..d597a185 --- /dev/null +++ b/src/renderer/src/terminal-output-batcher.ts @@ -0,0 +1,155 @@ +/** + * Renderer-side terminal output batching (P3): output events are merged and + * written to the emulator at most once per animation frame, in sequence order. + * Each batch is acknowledged with its highest sequence after the emulator's + * write callback, which is what Main's ACK-based backpressure expects. + */ + +/** Requests `callback` once; the returned function cancels the request. */ +export type TerminalFrameScheduler = (callback: () => void) => () => void + +/** + * Animation frames stop while the window is minimized or occluded, so a timer + * fallback keeps output (and ACKs) flowing for hidden terminals. Without it + * Main would pause the PTY until the window became visible again. + */ +export const TERMINAL_FRAME_FALLBACK_MS = 100 + +export const scheduleTerminalFrame: TerminalFrameScheduler = (callback) => { + let settled = false + let frame: number | undefined + const run = (): void => { + if (settled) { + return + } + settled = true + if (frame !== undefined) { + cancelAnimationFrame(frame) + } + clearTimeout(timer) + callback() + } + const timer = setTimeout(run, TERMINAL_FRAME_FALLBACK_MS) + if (typeof requestAnimationFrame === 'function') { + frame = requestAnimationFrame(run) + } + return () => { + settled = true + if (frame !== undefined) { + cancelAnimationFrame(frame) + } + clearTimeout(timer) + } +} + +export type TerminalOutputBatcherOptions = { + /** Writes data to the emulator; `done` must run after it is processed. */ + write: (data: string, done: () => void) => void + /** Acknowledges every output event up to and including `sequence`. */ + acknowledge: (sessionId: string, sequence: number) => void + schedule?: TerminalFrameScheduler +} + +export class TerminalOutputBatcher { + private readonly writeData: TerminalOutputBatcherOptions['write'] + private readonly acknowledge: TerminalOutputBatcherOptions['acknowledge'] + private readonly schedule: TerminalFrameScheduler + private pending: string[] = [] + private pendingSessionId: string | undefined + private pendingSequence = 0 + private cancelFrame: (() => void) | undefined + private lastWrite: Promise = Promise.resolve() + private disposed = false + + constructor(options: TerminalOutputBatcherOptions) { + this.writeData = options.write + this.acknowledge = options.acknowledge + this.schedule = options.schedule ?? scheduleTerminalFrame + } + + get hasPending(): boolean { + return this.pending.length > 0 + } + + /** Queues output; events must be pushed in increasing sequence order. */ + push(sessionId: string, sequence: number, data: string): void { + if (this.disposed) { + return + } + if (this.pendingSessionId !== undefined && this.pendingSessionId !== sessionId) { + // Never mix streams: the previous session's output is obsolete. + this.reset() + } + this.pendingSessionId = sessionId + this.pendingSequence = sequence + this.pending.push(data) + this.cancelFrame ??= this.schedule(() => { + this.cancelFrame = undefined + void this.flush() + }) + } + + /** + * Writes queued output now and resolves after every write issued so far has + * been processed by the emulator (so callers can order control events). + */ + flush(): Promise { + if (this.disposed) { + return Promise.resolve() + } + this.cancelFrame?.() + this.cancelFrame = undefined + if (this.pending.length > 0) { + const data = this.pending.length === 1 ? this.pending[0]! : this.pending.join('') + const sessionId = this.pendingSessionId! + const sequence = this.pendingSequence + this.pending = [] + this.pendingSessionId = undefined + this.lastWrite = new Promise((resolve) => { + let finished = false + const done = (): void => { + if (finished) { + return + } + finished = true + if (!this.disposed) { + this.acknowledge(sessionId, sequence) + } + resolve() + } + try { + this.writeData(data, done) + } catch { + done() + } + }) + } + return this.lastWrite + } + + /** + * Drops queued (unwritten) output when the session is replaced. The dropped + * events are still acknowledged so Main does not keep them counted against + * that session's backpressure window. + */ + reset(): void { + const sessionId = this.pendingSessionId + const sequence = this.pendingSequence + this.dropPending() + if (!this.disposed && sessionId !== undefined) { + this.acknowledge(sessionId, sequence) + } + } + + dispose(): void { + this.disposed = true + this.dropPending() + } + + private dropPending(): void { + this.cancelFrame?.() + this.cancelFrame = undefined + this.pending = [] + this.pendingSessionId = undefined + } +}