diff --git a/config/scripts/relay-replay-buffer-benchmark.mjs b/config/scripts/relay-replay-buffer-benchmark.mjs new file mode 100644 index 000000000..31091ede6 --- /dev/null +++ b/config/scripts/relay-replay-buffer-benchmark.mjs @@ -0,0 +1,218 @@ +#!/usr/bin/env node +// Benchmark: the relay's per-PTY-chunk replay buffer append (src/relay/pty-handler.ts). +// +// appendReplayBuffer did `buffered += data` then, over the cap, `buffered.slice(-CAP)`. +// Once a PTY has produced CAP bytes -- which a long-lived shell does almost immediately +// -- every subsequent chunk flattened and copied the whole 100 KB window. The append is +// called per raw node-pty emission, before batching, so it is per chunk, not per flush. +// +// The fix reuses RecentPtyOutputBuffer: keep chunks, drop from the head, and defer the +// join to read(), which only attach/adopt/revive call. +// +// Both arms are compared for an identical retained tail before timing. +// +// Run with: node config/scripts/relay-replay-buffer-benchmark.mjs +import { readFileSync } from 'node:fs' +import { performance } from 'node:perf_hooks' + +const ROUNDS = 6 +const SECONDS = Number(process.env.ORCA_REPLAY_BENCH_SECONDS ?? '1') +if (!Number.isFinite(SECONDS) || SECONDS <= 0) { + throw new Error(`ORCA_REPLAY_BENCH_SECONDS must be positive, received ${SECONDS}`) +} + +// Why re-read the sources: the claim is that the relay now appends into a chunk deque +// with the relay's own cap. If either reverts, these numbers stop meaning what they say. +const HANDLER_SOURCE = readFileSync( + new URL('../../src/relay/pty-handler.ts', import.meta.url), + 'utf8' +) +if (!/managed\.buffered\.append\(/.test(HANDLER_SOURCE)) { + throw new Error('relay no longer appends into a chunk deque; this benchmark is stale') +} +const capMatch = HANDLER_SOURCE.match(/REPLAY_BUFFER_MAX = ([\d *]+)/) +if (!capMatch) { + throw new Error('REPLAY_BUFFER_MAX not found; this benchmark is stale') +} +// The regex admits only digits, spaces, and `*`, so the literal is a plain product. +const REPLAY_BUFFER_MAX = capMatch[1] + .split('*') + .map((factor) => Number(factor.trim())) + .reduce((product, factor) => product * factor, 1) +if (!Number.isSafeInteger(REPLAY_BUFFER_MAX) || REPLAY_BUFFER_MAX <= 0) { + throw new Error(`could not read REPLAY_BUFFER_MAX from source, got ${capMatch[1]}`) +} + +// Pre-fix: rolling string, re-sliced once over the cap. +function appendString(state, data) { + if (data.length === 0) { + return state + } + const next = state + data + return next.length > REPLAY_BUFFER_MAX ? next.slice(-REPLAY_BUFFER_MAX) : next +} + +// Post-fix: mirrors RecentPtyOutputBuffer's append/read for the relay's options. +class ChunkDeque { + constructor(limit) { + this.chunks = [] + this.headIndex = 0 + this.headOffset = 0 + this.totalLen = 0 + this.limit = limit + } + + append(data) { + if (data.length === 0) { + return + } + if (data.length >= this.limit) { + this.chunks = [data.slice(-this.limit)] + this.headIndex = 0 + this.headOffset = 0 + this.totalLen = this.limit + return + } + this.chunks.push(data) + this.totalLen += data.length + while (this.totalLen > this.limit) { + const headRemaining = this.chunks[this.headIndex].length - this.headOffset + const excess = this.totalLen - this.limit + if (headRemaining <= excess) { + this.chunks[this.headIndex] = '' + this.headIndex += 1 + this.headOffset = 0 + this.totalLen -= headRemaining + } else { + this.headOffset += excess + this.totalLen -= excess + } + } + if (this.headIndex >= 1024) { + this.chunks = this.chunks.slice(this.headIndex) + this.headIndex = 0 + } + } + + read() { + if (this.chunks.length - this.headIndex > 1) { + const retained = this.chunks.slice(this.headIndex) + if (this.headOffset > 0) { + retained[0] = retained[0].slice(this.headOffset) + this.headOffset = 0 + } + this.chunks = [retained.join('')] + this.headIndex = 0 + } else if (this.headOffset > 0) { + this.chunks[this.headIndex] = this.chunks[this.headIndex].slice(this.headOffset) + this.headOffset = 0 + } + return this.chunks[this.headIndex] ?? '' + } +} + +function makeChunks(chunkBytes, chunkCount) { + // Vary content so V8 cannot dedupe or treat the appends as loop-invariant. + return Array.from({ length: chunkCount }, (_value, index) => + `${index}:`.padEnd(chunkBytes, 'abcdefghijklmnopqrstuvwxyz') + ) +} + +// Why pre-saturate: the interesting regime is a PTY that has already filled the window, +// which is where the old form copied 100 KB on literally every chunk. Timing from empty +// would average in a cheap warm-up the real process leaves behind in milliseconds. +function saturate(chunks) { + let stringState = '' + const deque = new ChunkDeque(REPLAY_BUFFER_MAX) + const preload = 'p'.repeat(REPLAY_BUFFER_MAX) + stringState = appendString(stringState, preload) + deque.append(preload) + return { stringState, deque, chunks } +} + +function median(samples) { + const sorted = [...samples].sort((a, b) => a - b) + const mid = sorted.length / 2 + return (sorted[mid - 1] + sorted[mid]) / 2 +} + +function timeString(chunks) { + let state = 'p'.repeat(REPLAY_BUFFER_MAX) + const start = performance.now() + for (const chunk of chunks) { + state = appendString(state, chunk) + } + const elapsed = performance.now() - start + if (state.length !== REPLAY_BUFFER_MAX) { + throw new Error('string arm lost its window') + } + return elapsed +} + +function timeDeque(chunks) { + const deque = new ChunkDeque(REPLAY_BUFFER_MAX) + deque.append('p'.repeat(REPLAY_BUFFER_MAX)) + const start = performance.now() + for (const chunk of chunks) { + deque.append(chunk) + } + const elapsed = performance.now() - start + return elapsed +} + +// Arms alternate which one leads so within-round drift cannot favour either. +function measure(chunks) { + timeString(chunks) + timeDeque(chunks) + const stringSamples = [] + const dequeSamples = [] + for (let round = 0; round < ROUNDS; round += 1) { + if (round % 2 === 0) { + stringSamples.push(timeString(chunks)) + dequeSamples.push(timeDeque(chunks)) + } else { + dequeSamples.push(timeDeque(chunks)) + stringSamples.push(timeString(chunks)) + } + } + return { stringMs: median(stringSamples), dequeMs: median(dequeSamples) } +} + +const pad = (value, width) => String(value).padStart(width) +console.log('Relay PTY replay-buffer append, per second of output. Lower is better.') +console.log( + `cap=${(REPLAY_BUFFER_MAX / 1024).toFixed(0)} KiB rounds=${ROUNDS} (per-arm medians, pre-saturated)` +) +console.log( + `${pad('workload', 30)} ${pad('rolling str', 12)} ${pad('chunk deque', 12)} ${pad('speedup', 9)}` +) + +for (const [label, chunkBytes, chunksPerSecond] of [ + ['interactive shell 64B x200', 64, 200], + ['agent TUI 512B x400', 512, 400], + ['build log 4KiB x256 (1 MiB/s)', 4 * 1024, 256], + ['dump 8KiB x512 (4 MiB/s)', 8 * 1024, 512], + ['firehose 16KiB x1024 (16 MiB/s)', 16 * 1024, 1024] +]) { + const chunks = makeChunks(chunkBytes, Math.round(chunksPerSecond * SECONDS)) + const { stringState, deque } = saturate(chunks) + let stringTail = stringState + for (const chunk of chunks) { + stringTail = appendString(stringTail, chunk) + deque.append(chunk) + } + if (deque.read() !== stringTail) { + throw new Error(`retained tail differs for ${label}`) + } + if (stringTail.length !== REPLAY_BUFFER_MAX) { + throw new Error(`fixture never saturated the window for ${label}`) + } + const { stringMs, dequeMs } = measure(chunks) + console.log( + `${pad(label, 30)} ${pad(`${stringMs.toFixed(3)} ms`, 12)} ${pad(`${dequeMs.toFixed(3)} ms`, 12)} ${pad(`${(stringMs / dequeMs).toFixed(0)}x`, 9)}` + ) +} + +console.log( + "\nThis is per PTY, and the relay runs on the user's SSH host. Reads (attach, adopt,\nrevive) now pay the join instead, but those are rare and were already O(window)." +) diff --git a/src/main/runtime/recent-pty-output-buffer.test.ts b/src/main/runtime/recent-pty-output-buffer.test.ts index 6c21284e1..42d9bfe30 100644 --- a/src/main/runtime/recent-pty-output-buffer.test.ts +++ b/src/main/runtime/recent-pty-output-buffer.test.ts @@ -259,3 +259,31 @@ describe('RecentPtyOutputBuffer', () => { } }) }) + +describe('configurable limit', () => { + // Why: a hardcoded RECENT_PTY_OUTPUT_LIMIT left in one arithmetic branch silently + // under-retained for any caller passing a different limit (the relay passes 100KB). + it('retains exactly the configured limit across many chunks', () => { + const limit = 1000 + const buffer = new RecentPtyOutputBuffer({ preserveChunkBoundaries: false, limit }) + let reference = '' + for (let index = 0; index < 60; index += 1) { + const chunk = `c${index}-`.repeat(9) + buffer.append(chunk) + reference = (reference + chunk).slice(-limit) + } + expect(buffer.read().length).toBe(limit) + expect(buffer.read()).toBe(reference) + }) + + it('rejects a non-positive limit', () => { + expect(() => new RecentPtyOutputBuffer({ limit: 0 })).toThrow('positive integer') + expect(() => new RecentPtyOutputBuffer({ limit: -1 })).toThrow('positive integer') + }) + + it('defaults to RECENT_PTY_OUTPUT_LIMIT', () => { + const buffer = new RecentPtyOutputBuffer({ preserveChunkBoundaries: false }) + buffer.append('z'.repeat(RECENT_PTY_OUTPUT_LIMIT * 2)) + expect(buffer.read().length).toBe(RECENT_PTY_OUTPUT_LIMIT) + }) +}) diff --git a/src/main/runtime/recent-pty-output-buffer.ts b/src/main/runtime/recent-pty-output-buffer.ts index a10559b75..dda01e0af 100644 --- a/src/main/runtime/recent-pty-output-buffer.ts +++ b/src/main/runtime/recent-pty-output-buffer.ts @@ -26,28 +26,34 @@ export class RecentPtyOutputBuffer { // Original chunk boundaries are owed only to the one-time path-candidate // backfill; compact() ends that obligation and lets read() collapse. private preserveChunkBoundaries: boolean + // Why configurable: the relay retains a different window than the main process. + private readonly limit: number - constructor(options?: { preserveChunkBoundaries?: boolean }) { + constructor(options?: { preserveChunkBoundaries?: boolean; limit?: number }) { this.preserveChunkBoundaries = options?.preserveChunkBoundaries ?? true + this.limit = options?.limit ?? RECENT_PTY_OUTPUT_LIMIT + if (!Number.isSafeInteger(this.limit) || this.limit <= 0) { + throw new Error(`RecentPtyOutputBuffer limit must be a positive integer, got ${this.limit}`) + } } append(data: string): void { if (data.length === 0) { return } - if (data.length >= RECENT_PTY_OUTPUT_LIMIT) { - this.chunks = [data.slice(-RECENT_PTY_OUTPUT_LIMIT)] + if (data.length >= this.limit) { + this.chunks = [data.slice(-this.limit)] this.headIndex = 0 this.headOffset = 0 - this.totalLen = RECENT_PTY_OUTPUT_LIMIT - this.headChunkIsPartial = data.length > RECENT_PTY_OUTPUT_LIMIT + this.totalLen = this.limit + this.headChunkIsPartial = data.length > this.limit return } this.chunks.push(data) this.totalLen += data.length - while (this.totalLen > RECENT_PTY_OUTPUT_LIMIT) { + while (this.totalLen > this.limit) { const headRemaining = this.chunks[this.headIndex].length - this.headOffset - const excess = this.totalLen - RECENT_PTY_OUTPUT_LIMIT + const excess = this.totalLen - this.limit if (headRemaining <= excess) { // Release the dropped chunk's reference; the slot is reclaimed on compaction. this.chunks[this.headIndex] = '' diff --git a/src/relay/pty-handler.ts b/src/relay/pty-handler.ts index 606d18e6e..578858d16 100644 --- a/src/relay/pty-handler.ts +++ b/src/relay/pty-handler.ts @@ -48,6 +48,7 @@ import { type PtyIngressEmission } from '../shared/pty-startup-ingress' import { resolvePtyOwnerBackend, type PtyOwnerBackend } from '../shared/pty-owner-backend' +import { RecentPtyOutputBuffer } from '../main/runtime/recent-pty-output-buffer' import { agentSessionOwnerBindingsEqual, ClaimedAgentPtyOwnerRegistry @@ -83,7 +84,9 @@ type ManagedPty = { incarnationId: string pty: IPty initialCwd: string - buffered: string + /** Why a chunk deque: rebuilding a rolling 100KB string per PTY chunk copied the + * whole window on every write once saturated. Readers are attach/adopt/revive only. */ + buffered: RecentPtyOutputBuffer /** Timer for SIGKILL fallback after a graceful SIGTERM shutdown. */ killTimer?: ReturnType /** True once disposeManagedPty has run; blocks double-dispose and makes post-dispose calls fail "not found" not silently. */ @@ -484,10 +487,7 @@ export class PtyHandler { if (data.length === 0) { return } - managed.buffered += data - if (managed.buffered.length > REPLAY_BUFFER_MAX) { - managed.buffered = managed.buffered.slice(-REPLAY_BUFFER_MAX) - } + managed.buffered.append(data) } private releaseStartupCommand(managed: ManagedPty): void { @@ -980,13 +980,12 @@ export class PtyHandler { throw new Error('agent_session_exited_during_start') } managed.agentSessionOwners = this.agentSessionOwners.listForPty(managed.id) + const adoptedReplay = result.disposition === 'adopted' ? managed.buffered.read() : '' return { id: managed.id, incarnationId: managed.incarnationId, agentSessionEnsure: result, - ...(result.disposition === 'adopted' && managed.buffered - ? { replay: managed.buffered } - : {}) + ...(adoptedReplay ? { replay: adoptedReplay } : {}) } } catch (error) { if (!physicalSpawnCommitted) { @@ -1111,7 +1110,10 @@ export class PtyHandler { incarnationId: randomUUID(), pty: term, initialCwd: cwd, - buffered: '', + buffered: new RecentPtyOutputBuffer({ + preserveChunkBoundaries: false, + limit: REPLAY_BUFFER_MAX + }), paneKey, tabId, ...(attachIdentity.paneKey || attachIdentity.tabId ? { attachIdentity } : {}), @@ -1198,14 +1200,15 @@ export class PtyHandler { // Why: renderer hasn't registered replay handlers yet during spawn, so return to the caller instead of notifying too early. // Why: buffer intentionally NOT cleared after replay (client clears xterm first) so later restarts still replay full history. - if (managed.buffered) { + const replay = managed.buffered.read() + if (replay) { // Why: drop pending batched bytes already in the replay buffer so attach doesn't render them twice. this.pendingOutputByPty.delete(id) this.clearOutputFlushTimerIfIdle() if (params.suppressReplayNotification) { - return { incarnationId: managed.incarnationId, replay: managed.buffered } + return { incarnationId: managed.incarnationId, replay } } - this.dispatcher.notify('pty.replay', { id, data: managed.buffered }) + this.dispatcher.notify('pty.replay', { id, data: replay }) } return { incarnationId: managed.incarnationId } } @@ -1552,7 +1555,10 @@ export class PtyHandler { incarnationId: randomUUID(), pty: term, initialCwd: entry.cwd, - buffered: '', + buffered: new RecentPtyOutputBuffer({ + preserveChunkBoundaries: false, + limit: REPLAY_BUFFER_MAX + }), paneKey: entry.paneKey, tabId: entry.tabId, attachIdentity: entry.attachIdentity, diff --git a/src/relay/pty-replay-buffer-equivalence.test.ts b/src/relay/pty-replay-buffer-equivalence.test.ts new file mode 100644 index 000000000..57b35ab84 --- /dev/null +++ b/src/relay/pty-replay-buffer-equivalence.test.ts @@ -0,0 +1,136 @@ +import { describe, expect, it } from 'vitest' +import { RecentPtyOutputBuffer } from '../main/runtime/recent-pty-output-buffer' +import { REPLAY_BUFFER_MAX } from './pty-handler' + +// Reference: the pre-change replay buffer, a rolling string sliced per append. +function appendStringTail(previous: string, data: string): string { + if (data.length === 0) { + return previous + } + const next = previous + data + return next.length > REPLAY_BUFFER_MAX ? next.slice(-REPLAY_BUFFER_MAX) : next +} + +function makeBuffer(): RecentPtyOutputBuffer { + return new RecentPtyOutputBuffer({ preserveChunkBoundaries: false, limit: REPLAY_BUFFER_MAX }) +} + +function replayEquals(chunks: string[]): { deque: string; reference: string } { + const buffer = makeBuffer() + let reference = '' + for (const chunk of chunks) { + buffer.append(chunk) + reference = appendStringTail(reference, chunk) + } + return { deque: buffer.read(), reference } +} + +// Deterministic PRNG so a failure is reproducible. +function makeRandom(seed: number): () => number { + let state = seed >>> 0 + return () => { + state = (state * 1664525 + 1013904223) >>> 0 + return state / 0x100000000 + } +} + +describe('relay replay buffer equivalence', () => { + it('matches the rolling-string tail below the cap', () => { + const { deque, reference } = replayEquals(['hello ', 'world', '\r\n$ ']) + expect(deque).toBe(reference) + expect(deque).toBe('hello world\r\n$ ') + }) + + it('matches once the window saturates', () => { + const chunks = Array.from({ length: 40 }, (_value, index) => `chunk-${index}-`.repeat(400)) + const { deque, reference } = replayEquals(chunks) + expect(deque.length).toBe(REPLAY_BUFFER_MAX) + expect(deque).toBe(reference) + }) + + it('matches when one append alone exceeds the cap', () => { + const { deque, reference } = replayEquals(['x'.repeat(REPLAY_BUFFER_MAX * 2 + 7)]) + expect(deque.length).toBe(REPLAY_BUFFER_MAX) + expect(deque).toBe(reference) + }) + + it('matches when an oversized append follows existing content', () => { + const { deque, reference } = replayEquals(['prefix', 'y'.repeat(REPLAY_BUFFER_MAX + 3)]) + expect(deque).toBe(reference) + expect(deque.startsWith('prefix')).toBe(false) + }) + + it('matches at exactly the cap boundary', () => { + for (const total of [REPLAY_BUFFER_MAX - 1, REPLAY_BUFFER_MAX, REPLAY_BUFFER_MAX + 1]) { + const { deque, reference } = replayEquals(['a'.repeat(total)]) + expect(deque).toBe(reference) + } + }) + + it('ignores empty appends exactly as the string form did', () => { + const { deque, reference } = replayEquals(['a', '', 'b', '', 'c']) + expect(deque).toBe(reference) + expect(deque).toBe('abc') + }) + + it('is stable across repeated reads', () => { + const buffer = makeBuffer() + for (let index = 0; index < 200; index += 1) { + buffer.append(`line ${index}\r\n`.repeat(30)) + } + expect(buffer.read()).toBe(buffer.read()) + }) + + it('keeps appending correctly after a read collapses the deque', () => { + const buffer = makeBuffer() + let reference = '' + for (let index = 0; index < 60; index += 1) { + const chunk = `mid-${index}-`.repeat(300) + buffer.append(chunk) + reference = appendStringTail(reference, chunk) + // Interleave reads: attach/adopt/revive can land at any point in the stream. + if (index % 7 === 0) { + expect(buffer.read()).toBe(reference) + } + } + expect(buffer.read()).toBe(reference) + }) + + // Why fuzz: the deque trims across chunk boundaries with a deferred head offset, + // so the risky cases are irregular chunk sizes straddling the cap repeatedly. + it('matches the rolling string across randomized chunk streams', () => { + for (let seed = 1; seed <= 40; seed += 1) { + const random = makeRandom(seed) + const buffer = makeBuffer() + let reference = '' + for (let step = 0; step < 120; step += 1) { + const size = Math.floor(random() * (REPLAY_BUFFER_MAX / 8)) + const chunk = String.fromCharCode(97 + (step % 26)).repeat(size) + buffer.append(chunk) + reference = appendStringTail(reference, chunk) + } + expect(buffer.read(), `seed ${seed}`).toBe(reference) + } + }) + + // Why surrogates: trimming by code unit can split a pair, and the string form did + // exactly the same thing. The deque must not "fix" it into a different tail. + it('splits surrogate pairs identically to the rolling string', () => { + // Why the trailing unit: the cap is even and a pair is 2 units, so an emoji run + // alone always cuts on a pair boundary. One odd unit shifts the cut mid-pair. + const { deque, reference } = replayEquals([`${'\u{1F600}'.repeat(REPLAY_BUFFER_MAX)}z`]) + expect(deque).toBe(reference) + expect(deque.length).toBe(REPLAY_BUFFER_MAX) + const leadUnit = deque.charCodeAt(0) + expect(leadUnit).toBeGreaterThanOrEqual(0xdc00) + expect(leadUnit).toBeLessThanOrEqual(0xdfff) + }) + + it('keeps a pair-aligned tail intact when the cut lands on a boundary', () => { + const { deque, reference } = replayEquals(['\u{1F600}'.repeat(REPLAY_BUFFER_MAX)]) + expect(deque).toBe(reference) + const leadUnit = deque.charCodeAt(0) + expect(leadUnit).toBeGreaterThanOrEqual(0xd800) + expect(leadUnit).toBeLessThanOrEqual(0xdbff) + }) +})