diff --git a/config/scripts/benchmark-cli-response-framing.mjs b/config/scripts/benchmark-cli-response-framing.mjs new file mode 100644 index 00000000000..40aab8d08f7 --- /dev/null +++ b/config/scripts/benchmark-cli-response-framing.mjs @@ -0,0 +1,128 @@ +import assert from 'node:assert/strict' +import { execFileSync } from 'node:child_process' +import { EventEmitter } from 'node:events' +import { readFileSync } from 'node:fs' +import Module from 'node:module' +import { dirname, resolve } from 'node:path' +import { performance } from 'node:perf_hooks' +import { build } from 'esbuild' + +// Run from the worktree root: node config/scripts/benchmark-cli-response-framing.mjs +const sourcePath = 'src/cli/runtime/transport.ts' +const baselineRef = process.argv[2] +assert.ok(baselineRef, 'Pass the pre-change transport revision as base-ref.') +const beforeSource = execFileSync('git', ['show', `${baselineRef}:${sourcePath}`], { + encoding: 'utf8' +}) +let chunks = [] + +async function loadTransport(source) { + const built = await build({ + stdin: { contents: source, loader: 'ts', resolveDir: dirname(resolve(sourcePath)) }, + bundle: true, + platform: 'node', + format: 'cjs', + write: false, + logLevel: 'silent' + }) + const module = new Module(resolve(sourcePath)) + const originalRequire = module.require.bind(module) + module.require = (name) => { + if (name === 'node:crypto') { + return { randomUUID: () => 'benchmark-request' } + } + if (name !== 'node:net') { + return originalRequire(name) + } + return { + createConnection() { + const socket = new EventEmitter() + socket.setEncoding = () => {} + socket.end = () => {} + socket.destroy = () => {} + socket.write = () => { + for (const chunk of chunks) { + socket.emit('data', chunk) + } + } + queueMicrotask(() => socket.emit('connect')) + return socket + } + } + } + module._compile(built.outputFiles[0].text, resolve(sourcePath)) + return module.exports.sendRequest +} + +const before = await loadTransport(beforeSource) +const after = await loadTransport(readFileSync(sourcePath, 'utf8')) +const metadata = { + runtimeId: 'benchmark-runtime', + authToken: 'benchmark-token', + transports: [{ kind: 'unix', endpoint: 'injected-socket' }] +} +const run = (sendRequest) => sendRequest(metadata, 'terminal.read', {}, 30000) + +async function measure(sendRequest, payloadBytes, repetitions) { + const warmup = await run(sendRequest) + assert.equal(warmup.result.data.length, payloadBytes) + const samples = [] + for (let sample = 0; sample < 5; sample++) { + const start = performance.now() + for (let iteration = 0; iteration < repetitions; iteration++) { + await run(sendRequest) + } + samples.push((performance.now() - start) / repetitions) + } + return samples.sort((a, b) => a - b)[2] +} + +async function searchedCharacters(sendRequest) { + const original = String.prototype.indexOf + let searched = 0 + String.prototype.indexOf = function (needle, position) { + if (needle === '\n') { + searched += this.length - (position ?? 0) + } + return original.call(this, needle, position) + } + try { + await run(sendRequest) + } finally { + String.prototype.indexOf = original + } + return searched +} + +const rows = [] +for (const [payloadBytes, chunkChars, repetitions] of [ + [32, 65536, 1000], + [1024 * 1024, 2 * 1024 * 1024, 20], + [1024 * 1024, 65536, 10], + [1024 * 1024, 4096, 5], + [4 * 1024 * 1024, 4096, 2], + [4 * 1024 * 1024, 256, 1] +]) { + const line = `${JSON.stringify({ + id: 'benchmark-request', + ok: true, + result: { data: 'x'.repeat(payloadBytes) }, + _meta: { runtimeId: 'benchmark-runtime' } + })}\n` + chunks = [] + for (let offset = 0; offset < line.length; offset += chunkChars) { + chunks.push(line.slice(offset, offset + chunkChars)) + } + const beforeMs = await measure(before, payloadBytes, repetitions) + const afterMs = await measure(after, payloadBytes, repetitions) + rows.push({ + payloadBytes, + chunkChars, + beforeMs: +beforeMs.toFixed(6), + afterMs: +afterMs.toFixed(6), + speedup: +(beforeMs / afterMs).toFixed(2), + beforeSearchedCharacters: await searchedCharacters(before), + afterSearchedCharacters: await searchedCharacters(after) + }) +} +console.log(JSON.stringify({ node: process.version, baselineRef, rows }, null, 2)) diff --git a/src/cli/runtime/transport-framing.test.ts b/src/cli/runtime/transport-framing.test.ts new file mode 100644 index 00000000000..cf3150f60c9 --- /dev/null +++ b/src/cli/runtime/transport-framing.test.ts @@ -0,0 +1,150 @@ +import { EventEmitter } from 'node:events' +import { StringDecoder } from 'node:string_decoder' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { RuntimeMetadata } from '../../shared/runtime-bootstrap' +import { sendRequest } from './transport' + +const { createConnection } = vi.hoisted(() => ({ createConnection: vi.fn() })) +vi.mock('node:net', () => ({ createConnection })) +vi.mock('node:crypto', () => ({ randomUUID: () => 'request-1' })) + +const metadata: RuntimeMetadata = { + runtimeId: 'runtime-1', + pid: 123, + transports: [{ kind: 'unix', endpoint: 'test-only' }], + authToken: 'token', + startedAt: 1 +} +const reply = (result: unknown) => + `${JSON.stringify({ id: 'request-1', ok: true, result, _meta: { runtimeId: 'runtime-1' } })}\n` + +class TestSocket extends EventEmitter { + setEncoding = vi.fn() + write = vi.fn() + end = vi.fn() + destroy = vi.fn() +} +let socket: TestSocket + +beforeEach(() => { + socket = new TestSocket() + createConnection.mockReturnValue(socket) +}) +afterEach(() => { + vi.restoreAllMocks() + vi.useRealTimers() +}) + +describe('CLI runtime response framing', () => { + it.each([1, 7, 256, 4096])( + 'reads a fragmented response with %i-character chunks', + async (size) => { + const result = { data: '็•Œ๐Ÿ˜€'.repeat(10000) } + const encoded = reply(result) + const pending = sendRequest(metadata, 'terminal.read', {}, 30000) + for (let offset = 0; offset < encoded.length; offset += size) { + socket.emit('data', encoded.slice(offset, offset + size)) + } + await expect(pending).resolves.toMatchObject({ result }) + expect(socket.setEncoding).toHaveBeenCalledExactlyOnceWith('utf8') + expect(socket.end).toHaveBeenCalledOnce() + } + ) + + it('accepts Unicode split across socket bytes using the existing UTF-8 decoder', async () => { + const pending = sendRequest(metadata, 'terminal.read', {}, 30000) + const decoder = new StringDecoder('utf8') + for (const byte of Buffer.from(reply({ data: '็•Œ๐Ÿ˜€รฉ' }))) { + socket.emit('data', decoder.write(Buffer.from([byte]))) + } + socket.emit('data', decoder.end()) + await expect(pending).resolves.toMatchObject({ result: { data: '็•Œ๐Ÿ˜€รฉ' } }) + }) + + it('searches each fragment once without rescanning the accumulated reply', async () => { + const encoded = reply({ data: 'x'.repeat(1024 * 1024) }) + const pending = sendRequest(metadata, 'terminal.read', {}, 30000) + const originalIndexOf = String.prototype.indexOf + let searchedCharacters = 0 + const search = vi + .spyOn(String.prototype, 'indexOf') + .mockImplementation(function (this: string, value, position) { + if (value === '\n') { + searchedCharacters += this.length - (position ?? 0) + } + return originalIndexOf.call(this, value, position) + }) + try { + for (let offset = 0; offset < encoded.length; offset += 256) { + socket.emit('data', encoded.slice(offset, offset + 256)) + } + } finally { + search.mockRestore() + } + await expect(pending).resolves.toMatchObject({ ok: true }) + expect(searchedCharacters).toBe(encoded.length) + }) + + it('refreshes keepalives across chunks and ignores blanks and data after the final frame', async () => { + vi.useFakeTimers() + const pending = sendRequest(metadata, 'terminal.read', {}, 100) + await vi.advanceTimersByTimeAsync(90) + socket.emit('data', ' \r\n{"_keep') + socket.emit('data', 'alive":true}\n\t\n') + await vi.advanceTimersByTimeAsync(90) + socket.emit('data', `${reply({ data: 'done' })}invalid JSON\n`) + socket.emit('data', 'more ignored data') + await expect(pending).resolves.toMatchObject({ result: { data: 'done' } }) + expect(socket.end).toHaveBeenCalledOnce() + expect(vi.getTimerCount()).toBe(0) + }) + + it.each([ + ['broken JSON\n', 'invalid_runtime_response'], + ['{}\n', 'invalid_runtime_response'], + ['{"id":"other","ok":true,"result":{}}\n', 'invalid_runtime_response'], + [ + '{"id":"request-1","ok":true,"result":{},"_meta":{"runtimeId":"other"}}\n', + 'runtime_unavailable' + ] + ])( + 'rejects a fragmented invalid first frame before subsequent valid frames', + async (line, code) => { + const pending = sendRequest(metadata, 'terminal.read', {}, 30000) + socket.emit('data', line.slice(0, 2)) + socket.emit('data', line.slice(2) + reply({ data: 'ignored' })) + await expect(pending).rejects.toMatchObject({ code }) + expect(socket.end).toHaveBeenCalledOnce() + } + ) + + it('preserves terminal failure envelopes', async () => { + const pending = sendRequest(metadata, 'terminal.read', {}, 30000) + socket.emit('data', '{"id":"request-1","ok":false,"error":{"code":"bad","message":"no"}}\n') + await expect(pending).resolves.toMatchObject({ + ok: false, + error: { code: 'bad', message: 'no' } + }) + }) + + it('rejects close with an incomplete frame and does not parse later data', async () => { + const pending = sendRequest(metadata, 'terminal.read', {}, 30000) + socket.emit('data', '{"id":') + socket.emit('close') + socket.emit('data', reply({ data: 'ignored' })) + await expect(pending).rejects.toMatchObject({ code: 'runtime_unavailable' }) + expect(socket.end).toHaveBeenCalledOnce() + }) + + it('destroys a timed out socket holding an incomplete frame', async () => { + vi.useFakeTimers() + const pending = sendRequest(metadata, 'terminal.read', {}, 100) + const rejected = expect(pending).rejects.toMatchObject({ code: 'runtime_timeout' }) + socket.emit('data', '{"id":') + await vi.advanceTimersByTimeAsync(100) + socket.emit('data', reply({ data: 'ignored' })) + await rejected + expect(socket.destroy).toHaveBeenCalledOnce() + expect(vi.getTimerCount()).toBe(0) + }) +}) diff --git a/src/cli/runtime/transport.ts b/src/cli/runtime/transport.ts index 091ca9da0e0..4e0d0a15ec7 100644 --- a/src/cli/runtime/transport.ts +++ b/src/cli/runtime/transport.ts @@ -31,7 +31,7 @@ export async function sendRequest( return } const socket = createConnection(transport.endpoint) - let buffer = '' + let lineSegments: string[] = [] let settled = false const requestId = randomUUID() @@ -40,6 +40,7 @@ export async function sendRequest( return } settled = true + lineSegments = [] socket.destroy() reject( new RuntimeClientError( @@ -56,6 +57,7 @@ export async function sendRequest( return } settled = true + lineSegments = [] clearTimeout(timeout) socket.end() if (result.ok === false) { @@ -89,18 +91,27 @@ export async function sendRequest( }) }) socket.on('data', (chunk: string) => { - buffer += chunk // Why: the server may interleave `{"_keepalive":true}\n` frames with the // final success/failure frame to keep both idle timers alive during a // long-poll (see design doc ยง3.1). Read frames in a loop until we see a // terminal frame. Each keepalive refreshes the client-side timer so a // 10 min wait doesn't trip the 60 s default ceiling. - let newlineIndex = buffer.indexOf('\n') - while (newlineIndex !== -1 && !settled) { - const line = buffer.slice(0, newlineIndex) - buffer = buffer.slice(newlineIndex + 1) + let cursor = 0 + while (cursor < chunk.length && !settled) { + const newlineIndex = chunk.indexOf('\n', cursor) + if (newlineIndex === -1) { + lineSegments.push(chunk.slice(cursor)) + return + } + const segment = chunk.slice(cursor, newlineIndex) + let line = segment + if (lineSegments.length > 0) { + lineSegments.push(segment) + line = lineSegments.join('') + lineSegments = [] + } + cursor = newlineIndex + 1 if (line.trim().length === 0) { - newlineIndex = buffer.indexOf('\n') continue } @@ -124,7 +135,6 @@ export async function sendRequest( // major). See ยง7 risk #9. if (isKeepaliveFrame(raw)) { timeout.refresh() - newlineIndex = buffer.indexOf('\n') continue } @@ -150,7 +160,6 @@ export async function sendRequest( const frame = parsed.data if ('_keepalive' in frame) { timeout.refresh() - newlineIndex = buffer.indexOf('\n') continue }