From 4546181b10ca643241fcfe22fa80396870d4b13c Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Fri, 10 Jul 2026 04:46:03 -0400 Subject: [PATCH] fix(relay): stream large git responses on the bulk lane MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Large git diff/exec responses (up to MAX_GIT_BUFFER=10MB) were sent as one JSON-RPC frame on the shared SSH channel. encodeJsonRpcFrame's synchronous JSON.stringify + Buffer.from of the whole payload, plus the multi-MB write, head-of-line-blocked interactive pty.data echo: opening a big diff (e.g. a lockfile) over SSH while typing in another pane stalled the echo for seconds. The 2026-07 fs.readFileStream fix added a per-client serialized bulk lane (RelayDispatcher.notifyBulk) honoring write()===false + waitWriteDrain, plus an ack credit window — but only fs streams used it; git responses never did. Stream the diff family (git.diff/branchDiff/commitDiff) and git.exec through the bulk lane in chunks with the same ack-window backpressure. The relay returns a small sentinel result and pumps git.responseChunk frames; the client reassembles and JSON.parses. Small responses stay single-frame. Cross-version safe via an opt-in capability gate (no handshake change): - client sends __streamResponse:true; old relays ignore it → single-frame. - new relays only stream when the flag is present → old clients get single-frame. - new client checks for the stream sentinel; a plain result (old relay) is returned directly. STREAM_CHUNK_SIZE is untouched; git chunking uses its own GIT_RESPONSE_CHUNK_SIZE and the client reassembles by concatenation, so chunk size is not baked into any cross-version offset math. Co-authored-by: Orca --- src/main/providers/ssh-git-provider.test.ts | 20 +- src/main/providers/ssh-git-provider.ts | 11 +- src/main/ssh/relay-protocol.ts | 29 ++ .../ssh/ssh-git-response-stream-reader.ts | 240 +++++++++++++++ src/relay/git-handler.ts | 72 ++++- ...-pty-echo-backpressure.integration.test.ts | 289 ++++++++++++++++++ src/relay/git-response-stream.ts | 190 ++++++++++++ src/relay/protocol.ts | 21 ++ src/relay/relay.ts | 6 +- 9 files changed, 855 insertions(+), 23 deletions(-) create mode 100644 src/main/ssh/ssh-git-response-stream-reader.ts create mode 100644 src/relay/git-response-pty-echo-backpressure.integration.test.ts create mode 100644 src/relay/git-response-stream.ts diff --git a/src/main/providers/ssh-git-provider.test.ts b/src/main/providers/ssh-git-provider.test.ts index 0974367ef81..91c781bf452 100644 --- a/src/main/providers/ssh-git-provider.test.ts +++ b/src/main/providers/ssh-git-provider.test.ts @@ -7,6 +7,7 @@ type MockMultiplexer = { notify: ReturnType onNotification: ReturnType onNotificationByMethod: ReturnType + onDispose: ReturnType dispose: ReturnType isDisposed: ReturnType } @@ -17,6 +18,9 @@ function createMockMux(): MockMultiplexer { notify: vi.fn(), onNotification: vi.fn(), onNotificationByMethod: vi.fn().mockReturnValue(vi.fn()), + // Why: requestGitStreamable subscribes to onDispose before awaiting the + // response so it can reject in-flight reassembly if the link drops. + onDispose: vi.fn().mockReturnValue(vi.fn()), dispose: vi.fn(), isDisposed: vi.fn().mockReturnValue(false) } @@ -320,7 +324,10 @@ describe('SshGitProvider', () => { 'git.exec', { args: ['clone', '--progress', '--', 'git@example.com:repo.git', 'repo'], - cwd: '/home/user' + cwd: '/home/user', + // Why: exec opts into response streaming so a large stdout is chunked + // onto the bulk lane; old relays ignore the flag. + __streamResponse: true }, { signal: controller.signal, @@ -353,7 +360,8 @@ describe('SshGitProvider', () => { }) expect(mux.request).toHaveBeenCalledWith('git.exec', { args: ['diff', '--cached', '--patch', '--minimal', '--no-color', '--no-ext-diff'], - cwd: '/home/user/repo' + cwd: '/home/user/repo', + __streamResponse: true }) }) @@ -636,7 +644,10 @@ describe('SshGitProvider', () => { expect(mux.request).toHaveBeenCalledWith('git.diff', { worktreePath: '/home/user/repo', filePath: 'src/index.ts', - staged: true + staged: true, + // Why: opts into response streaming; a small result still comes back as a + // single frame (relay decides), and old relays ignore the flag. + __streamResponse: true }) expect(result).toEqual(diffResult) }) @@ -880,7 +891,8 @@ describe('SshGitProvider', () => { const result = await provider.getBranchDiff('/home/user/repo', 'main') expect(mux.request).toHaveBeenCalledWith('git.branchDiff', { worktreePath: '/home/user/repo', - baseRef: 'main' + baseRef: 'main', + __streamResponse: true }) expect(result).toEqual(diffs) }) diff --git a/src/main/providers/ssh-git-provider.ts b/src/main/providers/ssh-git-provider.ts index 0cde62d534f..76fb163909b 100644 --- a/src/main/providers/ssh-git-provider.ts +++ b/src/main/providers/ssh-git-provider.ts @@ -21,6 +21,7 @@ import type { import type { GitHistoryOptions, GitHistoryResult } from '../../shared/git-history' import { buildHostedRemoteCommitUrl, buildHostedRemoteFileUrl } from '../git/hosted-remote-url' import { JsonRpcErrorCode } from '../ssh/relay-protocol' +import { requestGitStreamable } from '../ssh/ssh-git-response-stream-reader' import type { CommitMessageDraftContext } from '../../shared/commit-message-generation' import type { CommitMessagePlan } from '../../shared/commit-message-plan' import type { RemoteCommitMessageExecResult } from '../text-generation/commit-message-text-generation' @@ -373,7 +374,7 @@ export class SshGitProvider implements IGitProvider { return this.gitDiffReadDedupe.run( stableInFlightKey(['diff', worktreePath, filePath, staged, compareAgainstHead]), async () => - (await this.mux.request('git.diff', { + (await requestGitStreamable(this.mux, 'git.diff', { worktreePath, filePath, staged, @@ -597,7 +598,7 @@ export class SshGitProvider implements IGitProvider { keyOptions.oldPath ?? null ]), async () => - (await this.mux.request('git.branchDiff', { + (await requestGitStreamable(this.mux, 'git.branchDiff', { worktreePath, baseRef, ...options @@ -619,7 +620,7 @@ export class SshGitProvider implements IGitProvider { args.oldPath ?? null ]), async () => - (await this.mux.request('git.commitDiff', { + (await requestGitStreamable(this.mux, 'git.commitDiff', { worktreePath, ...args })) as GitDiffResult @@ -761,8 +762,8 @@ export class SshGitProvider implements IGitProvider { options?: { signal?: AbortSignal; timeoutMs?: number } ): Promise<{ stdout: string; stderr: string }> { const result = options - ? await this.mux.request('git.exec', { args, cwd }, options) - : await this.mux.request('git.exec', { args, cwd }) + ? await requestGitStreamable(this.mux, 'git.exec', { args, cwd }, options) + : await requestGitStreamable(this.mux, 'git.exec', { args, cwd }) return result as { stdout: string stderr: string diff --git a/src/main/ssh/relay-protocol.ts b/src/main/ssh/relay-protocol.ts index db57c4d19aa..6c61c5a4ba6 100644 --- a/src/main/ssh/relay-protocol.ts +++ b/src/main/ssh/relay-protocol.ts @@ -55,6 +55,35 @@ export const STREAM_CHUNK_SIZE = 256 * 1024 * client. */ export const MAX_CONCURRENT_STREAMS = 16 +// ── Git response streaming (see docs/relay-git-response-stream-design.md) ── + +/** Serialized-JSON size above which the relay chunks a streamable git response + * (diff family + exec) onto the bulk lane instead of one JSON-RPC frame. Mirror + * of the relay-side constant; the client only opts in — the relay owns the + * decision — so this is documentation of the shared contract. */ +export const GIT_RESPONSE_STREAM_THRESHOLD = 256 * 1024 + +/** Per-chunk size (serialized-result UTF-8 bytes) for git response streaming. + * The client reassembles by concatenation and does not depend on this value, + * so it stays cross-version safe. */ +export const GIT_RESPONSE_CHUNK_SIZE = 128 * 1024 + +/** Sentinel the relay returns as the RPC result when the real payload streams + * as git.responseChunk frames. Absent from old relays, so a new client falls + * back to the plain result they return. */ +export type GitResponseStreamMarker = { + __orcaGitResponseStream: { streamId: number; totalBytes: number; chunkCount: number } +} + +export function isGitResponseStreamMarker(value: unknown): value is GitResponseStreamMarker { + return ( + typeof value === 'object' && + value !== null && + typeof (value as GitResponseStreamMarker).__orcaGitResponseStream === 'object' && + (value as GitResponseStreamMarker).__orcaGitResponseStream !== null + ) +} + // ── JSON-RPC types ────────────────────────────────────────────────── export type JsonRpcRequest = { diff --git a/src/main/ssh/ssh-git-response-stream-reader.ts b/src/main/ssh/ssh-git-response-stream-reader.ts new file mode 100644 index 00000000000..02041666049 --- /dev/null +++ b/src/main/ssh/ssh-git-response-stream-reader.ts @@ -0,0 +1,240 @@ +import type { SshChannelMultiplexer } from './ssh-channel-multiplexer' +import { RelayErrorCode, isGitResponseStreamMarker } from './relay-protocol' + +const SENTINEL_STREAM_ID = -1 + +export class GitResponseStreamError extends Error { + readonly code = RelayErrorCode.StreamProtocolError + constructor(message: string) { + super(message) + } +} + +type PendingFrame = + | { kind: 'chunk'; params: Record } + | { kind: 'end'; params: Record } + | { kind: 'error'; params: Record } + +/** + * Request a git method that may return a large payload, opting into response + * streaming so a big diff/exec response is chunked onto the relay's bulk lane + * instead of one JSON-RPC frame (which would head-of-line-block pty.data echo + * on the shared SSH channel). + * + * Cross-version behavior: + * - New relay + big result → returns the stream sentinel; we reassemble chunks. + * - New relay + small result, or old client → plain single-frame result. + * - Old relay (ignores `__streamResponse`) → returns the plain result; the + * marker check fails and we return it directly, i.e. today's behavior. + */ +export function requestGitStreamable( + mux: SshChannelMultiplexer, + method: string, + params: Record, + options?: { signal?: AbortSignal } +): Promise { + // Why: subscribe to chunk/end/error BEFORE awaiting the sentinel response so a + // chunk that lands in the same dispatch tick as the response is not dropped + // (mirrors readFileViaStream). streamIdRef stays SENTINEL until the sentinel + // resolves; frames are queued until then and drained. + const streamIdRef = { current: SENTINEL_STREAM_ID } + const unsubscribers: (() => void)[] = [] + const cleanup = (): void => { + while (unsubscribers.length > 0) { + try { + unsubscribers.pop()?.() + } catch { + // best-effort + } + } + } + + return new Promise((resolve, reject) => { + const parts: Buffer[] = [] + let expectedSeq = 0 + let receivedBytes = 0 + let totalBytes = 0 + let chunkCount = 0 + let settled = false + let metadataReady = false + const pending: PendingFrame[] = [] + + const cancel = (): void => { + if (streamIdRef.current !== SENTINEL_STREAM_ID && !mux.isDisposed()) { + try { + mux.notify('git.cancelResponseStream', { streamId: streamIdRef.current }) + } catch { + // best-effort + } + } + } + const fail = (err: Error): void => { + if (settled) { + return + } + settled = true + cancel() + cleanup() + reject(err) + } + const succeed = (value: unknown): void => { + if (settled) { + return + } + settled = true + cleanup() + resolve(value) + } + + const handleChunk = (p: Record): void => { + if (settled || p.streamId !== streamIdRef.current) { + return + } + const seq = p.seq as number + const data = p.data as string + if (typeof seq !== 'number' || typeof data !== 'string') { + fail(new GitResponseStreamError(`Malformed chunk for git stream ${streamIdRef.current}`)) + return + } + if (seq !== expectedSeq) { + fail( + new GitResponseStreamError( + `Out-of-order chunk for git stream ${streamIdRef.current}: expected ${expectedSeq}, got ${seq}` + ) + ) + return + } + const decoded = Buffer.from(data, 'base64') + parts.push(decoded) + receivedBytes += decoded.length + expectedSeq += 1 + // Why: credit-based flow control — the relay caps unacked chunks so a big + // response cannot queue unbounded ahead of interactive pty.data frames. + mux.notify('git.responseAck', { streamId: streamIdRef.current, seq }) + } + + const handleEnd = (p: Record): void => { + if (settled || p.streamId !== streamIdRef.current) { + return + } + if (expectedSeq !== chunkCount || receivedBytes !== totalBytes) { + fail( + new GitResponseStreamError( + `Git stream ${streamIdRef.current} incomplete: chunks ${expectedSeq}/${chunkCount}, bytes ${receivedBytes}/${totalBytes}` + ) + ) + return + } + try { + succeed(JSON.parse(Buffer.concat(parts).toString('utf-8'))) + } catch (err) { + fail( + new GitResponseStreamError( + `Git stream ${streamIdRef.current} JSON parse failed: ${String(err)}` + ) + ) + } + } + + const handleStreamError = (p: Record): void => { + if (settled || p.streamId !== streamIdRef.current) { + return + } + fail(new Error((p.message as string | undefined) ?? 'git response stream error')) + } + + const drainPending = (): void => { + while (!settled && pending.length > 0) { + const frame = pending.shift()! + if (frame.kind === 'chunk') { + handleChunk(frame.params) + } else if (frame.kind === 'end') { + handleEnd(frame.params) + } else { + handleStreamError(frame.params) + } + } + } + + unsubscribers.push( + mux.onNotificationByMethod('git.responseChunk', (p) => { + if (!metadataReady) { + pending.push({ kind: 'chunk', params: p }) + return + } + handleChunk(p) + }) + ) + unsubscribers.push( + mux.onNotificationByMethod('git.responseEnd', (p) => { + if (!metadataReady) { + pending.push({ kind: 'end', params: p }) + return + } + handleEnd(p) + }) + ) + unsubscribers.push( + mux.onNotificationByMethod('git.responseError', (p) => { + if (!metadataReady) { + pending.push({ kind: 'error', params: p }) + return + } + handleStreamError(p) + }) + ) + unsubscribers.push( + mux.onDispose((reason) => { + const err = new Error( + reason === 'connection_lost' + ? 'SSH connection lost, reconnecting...' + : 'Multiplexer disposed' + ) as Error & { code: string } + err.code = reason === 'connection_lost' ? 'CONNECTION_LOST' : 'DISPOSED' + fail(err) + }) + ) + + if (options?.signal) { + const signal = options.signal + if (signal.aborted) { + const err = new Error('Request was cancelled') as Error & { name: string } + err.name = 'AbortError' + fail(err) + return + } + const onAbort = (): void => { + const err = new Error('Request was cancelled') as Error & { name: string } + err.name = 'AbortError' + fail(err) + } + signal.addEventListener('abort', onAbort, { once: true }) + unsubscribers.push(() => signal.removeEventListener('abort', onAbort)) + } + + // Why: forward options only when present so callers that previously issued a + // 2-arg mux.request keep the same call shape (and their tests). + const streamParams = { ...params, __streamResponse: true } + const requestPromise = options + ? mux.request(method, streamParams, options) + : mux.request(method, streamParams) + void requestPromise + .then((result) => { + if (settled) { + return + } + // Old relay / small result: plain single-frame value, no stream follows. + if (!isGitResponseStreamMarker(result)) { + succeed(result) + return + } + const marker = result.__orcaGitResponseStream + totalBytes = marker.totalBytes + chunkCount = marker.chunkCount + streamIdRef.current = marker.streamId + metadataReady = true + drainPending() + }) + .catch((err) => fail(err as Error)) + }) +} diff --git a/src/relay/git-handler.ts b/src/relay/git-handler.ts index c878dfd563e..2c816115c35 100644 --- a/src/relay/git-handler.ts +++ b/src/relay/git-handler.ts @@ -61,6 +61,8 @@ import { getGitCloneFailureMessage } from '../shared/git-clone-failure-message' import { syncForkDefaultBranch, validateGitForkSyncExpectedUpstream } from '../shared/git-fork-sync' import { InFlightPromiseDedupe, stableInFlightKey } from '../shared/in-flight-promise-dedupe' import { GIT_FETCH_SKIP_AUTO_MAINTENANCE_CONFIG_ARGS } from '../shared/git-fetch-auto-maintenance' +import { GitResponseStreamRegistry } from './git-response-stream' +import { GIT_RESPONSE_STREAM_THRESHOLD } from './protocol' const execFileAsync = promisify(execFile) const MAX_GIT_BUFFER = 10 * 1024 * 1024 @@ -171,6 +173,9 @@ function execFileWithStdin( export class GitHandler { private dispatcher: RelayDispatcher private readonly gitDiffReadDedupe = new InFlightPromiseDedupe() + // Why: large diff/exec responses are chunked onto the bulk lane so they do + // not head-of-line-block interactive pty.data echo on the shared SSH channel. + private readonly responseStreams = new GitResponseStreamRegistry() // Why: configured submodule paths change rarely; an instance-level TTL cache // avoids re-reading `.gitmodules` on every diff click over SSH, and being @@ -182,6 +187,13 @@ export class GitHandler { constructor(dispatcher: RelayDispatcher, _context: RelayContext) { this.dispatcher = dispatcher this.registerHandlers() + // Why: a detached client's git.responseAck frames will never arrive; wake + // any pump parked on the ack window so it re-checks staleness and exits. + this.dispatcher.onClientDetached?.(() => this.responseStreams.wakeAll()) + } + + dispose(): void { + this.responseStreams.disposeAll() } private registerHandlers(): void { @@ -190,7 +202,7 @@ export class GitHandler { this.dispatcher.onRequest('git.checkIgnored', (p) => this.checkIgnored(p)) this.dispatcher.onRequest('git.history', (p) => this.history(p)) this.dispatcher.onRequest('git.commit', (p) => this.commit(p)) - this.dispatcher.onRequest('git.diff', (p) => this.getDiff(p)) + this.dispatcher.onRequest('git.diff', (p, context) => this.getDiff(p, context)) this.dispatcher.onRequest('git.stage', (p) => this.stage(p)) this.dispatcher.onRequest('git.unstage', (p) => this.unstage(p)) this.dispatcher.onRequest('git.bulkStage', (p) => this.bulkStage(p)) @@ -215,8 +227,8 @@ export class GitHandler { this.dispatcher.onRequest('git.pull', (p) => this.pull(p)) this.dispatcher.onRequest('git.fastForward', (p) => this.fastForward(p)) this.dispatcher.onRequest('git.rebaseFromBase', (p) => this.rebaseFromBase(p)) - this.dispatcher.onRequest('git.branchDiff', (p) => this.branchDiff(p)) - this.dispatcher.onRequest('git.commitDiff', (p) => this.commitDiff(p)) + this.dispatcher.onRequest('git.branchDiff', (p, context) => this.branchDiff(p, context)) + this.dispatcher.onRequest('git.commitDiff', (p, context) => this.commitDiff(p, context)) this.dispatcher.onRequest('git.listWorktrees', (p, context) => this.listWorktrees(p, context)) this.dispatcher.onRequest('git.addWorktree', (p) => this.addWorktree(p)) this.dispatcher.onRequest('git.removeWorktree', (p) => this.removeWorktree(p)) @@ -231,6 +243,43 @@ export class GitHandler { this.dispatcher.onRequest('git.exec', (p, context) => this.exec(p, context)) this.dispatcher.onRequest('git.clone', (p, context) => this.clone(p, context)) this.dispatcher.onRequest('git.isGitRepo', (p) => this.isGitRepo(p)) + this.dispatcher.onNotification('git.responseAck', (p) => this.responseAck(p)) + this.dispatcher.onNotification('git.cancelResponseStream', (p) => this.cancelResponseStream(p)) + } + + private responseAck(params: Record): void { + const streamId = params.streamId + const seq = params.seq + if (typeof streamId === 'number' && typeof seq === 'number') { + this.responseStreams.recordAck(streamId, seq) + } + } + + private cancelResponseStream(params: Record): void { + const streamId = params.streamId + if (typeof streamId === 'number') { + this.responseStreams.abort(streamId) + } + } + + // Why: when the client opted into response streaming and the serialized result + // exceeds the threshold, chunk it onto the bulk lane and return a small + // sentinel as the RPC result. Old clients omit the flag (single-frame, as + // today); old relays never call this, so a new client falls back to the plain + // result they return. + private maybeStreamResponse( + result: unknown, + params: Record, + context: RequestContext | undefined + ): unknown { + if (params.__streamResponse !== true || !context) { + return result + } + const payload = Buffer.from(JSON.stringify(result ?? null), 'utf-8') + if (payload.length <= GIT_RESPONSE_STREAM_THRESHOLD) { + return result + } + return this.responseStreams.startStream(payload, this.dispatcher, context) } private async runWithDiffDedupeClear(run: () => Promise): Promise { @@ -352,7 +401,7 @@ export class GitHandler { }) } - private async getDiff(params: Record) { + private async getDiff(params: Record, context?: RequestContext) { const worktreePath = params.worktreePath as string const filePath = params.filePath as string // Why: filePath is relative to worktreePath and used in readWorkingFile via @@ -366,7 +415,7 @@ export class GitHandler { const compareAgainstHead = params.compareAgainstHead as boolean | undefined // Why: register the in-flight dedupe synchronously (before any await) so // concurrent identical reads coalesce; submodule routing happens inside. - return this.gitDiffReadDedupe.run( + const result = await this.gitDiffReadDedupe.run( stableInFlightKey(['diff', worktreePath, filePath, staged, compareAgainstHead]), async () => { // Why: gitlink paths can't be read as blobs and submodule working dirs @@ -431,6 +480,7 @@ export class GitHandler { ) } ) + return this.maybeStreamResponse(result, params, context) } private async stage(params: Record) { @@ -1027,7 +1077,7 @@ export class GitHandler { } } - private async branchDiff(params: Record) { + private async branchDiff(params: Record, context?: RequestContext) { const worktreePath = params.worktreePath as string const baseRef = params.baseRef as string if (baseRef.startsWith('-')) { @@ -1038,7 +1088,7 @@ export class GitHandler { filePath: params.filePath as string | undefined, oldPath: params.oldPath as string | undefined } - return this.gitDiffReadDedupe.run( + const result = await this.gitDiffReadDedupe.run( stableInFlightKey([ 'branchDiff', worktreePath, @@ -1056,9 +1106,10 @@ export class GitHandler { options ) ) + return this.maybeStreamResponse(result, params, context) } - private async commitDiff(params: Record) { + private async commitDiff(params: Record, context?: RequestContext) { const worktreePath = params.worktreePath as string const args = { commitOid: params.commitOid as string, @@ -1066,7 +1117,7 @@ export class GitHandler { filePath: params.filePath as string, oldPath: params.oldPath as string | undefined } - return this.gitDiffReadDedupe.run( + const result = await this.gitDiffReadDedupe.run( stableInFlightKey([ 'commitDiff', worktreePath, @@ -1077,6 +1128,7 @@ export class GitHandler { ]), () => commitDiffEntry(this.gitBuffer.bind(this), worktreePath, args) ) + return this.maybeStreamResponse(result, params, context) } private async exec(params: Record, context?: RequestContext) { @@ -1085,7 +1137,7 @@ export class GitHandler { validateGitExecArgs(args) const { stdout, stderr } = await this.git(args, cwd, { signal: context?.signal }) - return { stdout, stderr } + return this.maybeStreamResponse({ stdout, stderr }, params, context) } private async clone(params: Record, context?: RequestContext) { diff --git a/src/relay/git-response-pty-echo-backpressure.integration.test.ts b/src/relay/git-response-pty-echo-backpressure.integration.test.ts new file mode 100644 index 00000000000..28dcf219a90 --- /dev/null +++ b/src/relay/git-response-pty-echo-backpressure.integration.test.ts @@ -0,0 +1,289 @@ +/** + * Regression: SSH typing latency under large git responses. + * + * The relay and the client share ONE ordered SSH channel. A large git diff/exec + * response sent as a single JSON-RPC frame queues megabytes into the outbound + * pipe at once; an interactive pty.data echo emitted mid-response then sits + * behind all of it and typing feels seconds-slow. + * + * These tests model the SSH channel as a congestible in-memory pipe and assert + * deterministic byte bounds instead of wall-clock latency: + * - WITHOUT the fix (client does not opt in), the whole response queues ahead + * of a pty echo; + * - WITH the fix (client opts in), the response streams on the bulk lane and + * at most ~1 chunk frame sits ahead of the echo; + * - both paths reassemble the identical git result. + */ +import { describe, expect, it, beforeEach, afterEach } from 'vitest' +import { mkdtempSync, mkdirSync, writeFileSync } from 'node:fs' +import { rm } from 'node:fs/promises' +import { execFileSync } from 'node:child_process' +import * as path from 'node:path' +import { tmpdir } from 'node:os' + +import { + SshChannelMultiplexer, + type MultiplexerTransport +} from '../main/ssh/ssh-channel-multiplexer' +import { requestGitStreamable } from '../main/ssh/ssh-git-response-stream-reader' + +import { RelayDispatcher } from './dispatcher' +import { RelayContext } from './context' +import { GitHandler } from './git-handler' +import { GIT_RESPONSE_CHUNK_SIZE } from './protocol' + +// One framed git.responseChunk: base64 (4/3) + JSON envelope + header slack. +const FRAMED_CHUNK_BYTES = Math.ceil((GIT_RESPONSE_CHUNK_SIZE * 4) / 3) + 512 +const SINK_HIGH_WATER_MARK = 64 * 1024 + +async function waitUntil( + predicate: () => boolean, + what: string, + timeoutMs = 10_000 +): Promise { + const deadline = Date.now() + timeoutMs + while (!predicate()) { + if (Date.now() > deadline) { + throw new Error(`waitUntil timed out: ${what}`) + } + await new Promise((r) => setImmediate(r)) + } +} + +async function waitUntilSettled(read: () => number, stableTurns = 25): Promise { + let last = read() + let stable = 0 + while (stable < stableTurns) { + await new Promise((r) => setImmediate(r)) + const current = read() + if (current === last) { + stable += 1 + } else { + stable = 0 + last = current + } + } +} + +type Harness = { + mux: SshChannelMultiplexer + dispatcher: RelayDispatcher + gitHandler: GitHandler + queuedBytes: () => number + deliverAll: () => void + startAutoDeliver: () => void + dispose: () => void +} + +function createHarness(opts: { congested: boolean }): Harness { + let relayFeed: ((data: Buffer) => void) | null = null + const clientDataCallbacks: ((data: Buffer) => void)[] = [] + + const clientTransport: MultiplexerTransport = { + write: (data: Buffer) => { + // Client → relay (keystrokes, acks) is the opposite duplex direction and + // is not blocked by relay→client congestion. + setImmediate(() => relayFeed?.(data)) + }, + onData: (cb) => { + clientDataCallbacks.push(cb) + }, + onClose: () => {} + } + + const outQueue: Buffer[] = [] + let queuedBytes = 0 + const drainWaiters = new Set<() => void>() + const fireDrainIfIdle = (): void => { + if (queuedBytes > 0) { + return + } + for (const cb of Array.from(drainWaiters)) { + drainWaiters.delete(cb) + cb() + } + } + + const dispatcher = new RelayDispatcher( + (data: Buffer) => { + outQueue.push(data) + queuedBytes += data.length + if (!opts.congested) { + return true + } + return queuedBytes < SINK_HIGH_WATER_MARK + }, + { + waitWriteDrain: (cb: () => void) => { + drainWaiters.add(cb) + fireDrainIfIdle() + } + } + ) + relayFeed = (data: Buffer) => dispatcher.feed(data) + + const deliverAll = (): void => { + while (outQueue.length > 0) { + const buf = outQueue.shift()! + queuedBytes -= buf.length + for (const cb of clientDataCallbacks) { + cb(buf) + } + } + fireDrainIfIdle() + } + + let autoDeliverTimer: ReturnType | null = null + const startAutoDeliver = (): void => { + if (autoDeliverTimer) { + return + } + autoDeliverTimer = setInterval(deliverAll, 1) + } + + const context = new RelayContext() + const gitHandler = new GitHandler(dispatcher, context) + const mux = new SshChannelMultiplexer(clientTransport) + + return { + mux, + dispatcher, + gitHandler, + queuedBytes: () => queuedBytes, + deliverAll, + startAutoDeliver, + dispose: () => { + if (autoDeliverTimer) { + clearInterval(autoDeliverTimer) + } + mux.dispose() + dispatcher.dispose() + gitHandler.dispose() + } + } +} + +// Build a repo whose staged diff for `big.txt` is several MB so git.diff +// returns a payload well over the stream threshold. +function makeRepoWithLargeStagedDiff(dir: string): void { + const env = { ...process.env } + const run = (args: string[]): void => { + execFileSync('git', args, { cwd: dir, stdio: 'pipe', env }) + } + run(['init']) + run(['config', 'user.email', 'test@test.com']) + run(['config', 'user.name', 'Test']) + // Distinct non-repeating lines keep the diff from compressing to nothing. + // Stay under the render limits (120k lines / 6M chars) so the diff result + // carries the full ~4MB content and exceeds the stream threshold. + const lines: string[] = [] + for (let i = 0; i < 12_000; i += 1) { + lines.push(`line ${i} ${'x'.repeat(100)}`) + } + writeFileSync(path.join(dir, 'big.txt'), lines.join('\n')) + run(['add', 'big.txt']) +} + +describe('large git response vs pty.data echo head-of-line blocking', () => { + let tmpDir: string + let repoDir: string + + beforeEach(() => { + tmpDir = mkdtempSync(path.join(tmpdir(), 'relay-git-hol-')) + repoDir = path.join(tmpDir, 'repo') + mkdirSync(repoDir, { recursive: true }) + makeRepoWithLargeStagedDiff(repoDir) + }) + + afterEach(async () => { + await rm(tmpDir, { recursive: true, force: true }) + }) + + it('WITHOUT opt-in: the whole diff queues ahead of a pty echo (single-frame HOL)', async () => { + const harness = createHarness({ congested: true }) + try { + let queuedBytesAheadOfEcho = -1 + harness.dispatcher.onNotification('pty.data', (params) => { + queuedBytesAheadOfEcho = harness.queuedBytes() + harness.dispatcher.notify('pty.data', { id: params.id, data: params.data }) + }) + + // Plain request WITHOUT __streamResponse: single-frame response, as today. + const diffPromise = harness.mux.request('git.diff', { + worktreePath: repoDir, + filePath: 'big.txt', + staged: true + }) + // Let the whole single JSON-RPC frame land in the congested pipe. + await waitUntil(() => harness.queuedBytes() > SINK_HIGH_WATER_MARK, 'diff frame queued') + await waitUntilSettled(() => harness.queuedBytes()) + + harness.mux.notify('pty.data', { id: 'pty-1', data: 'x' }) + await waitUntil(() => queuedBytesAheadOfEcho >= 0, 'echo emitted by relay') + + // The echo sits behind the entire (multi-hundred-KB) diff frame — far + // more than a single bulk chunk would be. + expect(queuedBytesAheadOfEcho).toBeGreaterThan(512 * 1024) + + harness.startAutoDeliver() + const result = (await diffPromise) as { diff?: string } + expect(typeof result).toBe('object') + } finally { + harness.dispose() + } + }, 30_000) + + it('WITH opt-in: the diff streams on the bulk lane and the echo stays bounded', async () => { + const harness = createHarness({ congested: true }) + try { + let queuedBytesAheadOfEcho = -1 + harness.dispatcher.onNotification('pty.data', (params) => { + queuedBytesAheadOfEcho = harness.queuedBytes() + harness.dispatcher.notify('pty.data', { id: params.id, data: params.data }) + }) + + const diffPromise = requestGitStreamable(harness.mux, 'git.diff', { + worktreePath: repoDir, + filePath: 'big.txt', + staged: true + }) + // Deliver the sentinel response, then let the pump run into congestion. + await waitUntil(() => harness.queuedBytes() > 0, 'sentinel queued') + harness.deliverAll() + await waitUntil(() => harness.queuedBytes() > 0, 'first chunk queued') + await waitUntilSettled(() => harness.queuedBytes()) + + harness.mux.notify('pty.data', { id: 'pty-1', data: 'x' }) + await waitUntil(() => queuedBytesAheadOfEcho >= 0, 'echo emitted by relay') + + // At most one in-flight bulk frame (the write that saturated the sink) + // plus slack sits ahead of the echo. + expect(queuedBytesAheadOfEcho).toBeLessThan(2 * FRAMED_CHUNK_BYTES) + + harness.startAutoDeliver() + const streamed = (await diffPromise) as Record + expect(typeof streamed).toBe('object') + expect(streamed).not.toHaveProperty('__orcaGitResponseStream') + } finally { + harness.dispose() + } + }, 30_000) + + it('streamed result equals the single-frame result', async () => { + const harness = createHarness({ congested: false }) + try { + harness.startAutoDeliver() + const params = { worktreePath: repoDir, filePath: 'big.txt', staged: true } + // Why: request both concurrently so the shared 1ms delivery pump drives + // the plain frame and the streamed chunk/ack round-trips without either + // starving the other on a slow CI event loop. + const [plain, streamed] = await Promise.all([ + harness.mux.request('git.diff', params), + requestGitStreamable(harness.mux, 'git.diff', params) + ]) + expect(streamed).toEqual(plain) + } finally { + harness.dispose() + } + }, 30_000) +}) diff --git a/src/relay/git-response-stream.ts b/src/relay/git-response-stream.ts new file mode 100644 index 00000000000..d114bf23a02 --- /dev/null +++ b/src/relay/git-response-stream.ts @@ -0,0 +1,190 @@ +// Streams large git RPC responses (diff family + exec) onto the bulk lane in +// chunks instead of one JSON-RPC frame, so a big diff cannot head-of-line-block +// interactive pty.data echo on the shared SSH channel. Mirrors the fs +// read-stream credit-window pattern (see fs-handler-file-read.ts) but the +// payload is an in-memory serialized string rather than a file handle. +import type { RelayDispatcher, RequestContext } from './dispatcher' +import { + GIT_RESPONSE_CHUNK_SIZE, + STREAM_ACK_WINDOW_CHUNKS, + STREAM_ACK_STALL_RECHECK_MS, + type GitResponseStreamMarker +} from './protocol' + +type GitResponseStreamEntry = { + aborted: boolean + /** Highest chunk seq the client acknowledged (in-order; -1 = none yet). */ + ackedThroughSeq: number + ackWaiters: Set<() => void> +} + +/** Serialized git responses are chunked as base64 so multi-byte UTF-8 + * sequences never split across a chunk boundary (the client concatenates the + * decoded bytes and parses once). */ +function encodeChunks(payload: Buffer): string[] { + const chunks: string[] = [] + for (let offset = 0; offset < payload.length; offset += GIT_RESPONSE_CHUNK_SIZE) { + chunks.push(payload.subarray(offset, offset + GIT_RESPONSE_CHUNK_SIZE).toString('base64')) + } + return chunks +} + +export class GitResponseStreamRegistry { + private streams = new Map() + private nextId = 1 + + private register(): number { + const streamId = this.nextId++ + this.streams.set(streamId, { aborted: false, ackedThroughSeq: -1, ackWaiters: new Set() }) + return streamId + } + + recordAck(streamId: number, seq: number): void { + const entry = this.streams.get(streamId) + if (!entry || typeof seq !== 'number' || !Number.isFinite(seq)) { + return + } + if (seq > entry.ackedThroughSeq) { + entry.ackedThroughSeq = seq + } + this.wake(entry) + } + + abort(streamId: number): void { + const entry = this.streams.get(streamId) + if (entry) { + entry.aborted = true + this.wake(entry) + } + } + + /** Wake every parked pump so it re-checks staleness — used when a client + * detaches and its acks will never arrive. */ + wakeAll(): void { + for (const entry of this.streams.values()) { + this.wake(entry) + } + } + + private wake(entry: GitResponseStreamEntry): void { + for (const waiter of Array.from(entry.ackWaiters)) { + waiter() + } + } + + private waitForAck(streamId: number): Promise { + const entry = this.streams.get(streamId) + if (!entry || entry.aborted) { + return Promise.resolve() + } + return new Promise((resolve) => { + let settled = false + const finish = (): void => { + if (settled) { + return + } + settled = true + clearTimeout(timer) + entry.ackWaiters.delete(finish) + resolve() + } + const timer = setTimeout(finish, STREAM_ACK_STALL_RECHECK_MS) + timer.unref?.() + entry.ackWaiters.add(finish) + }) + } + + /** + * Register a stream for `payload`, kick off the bulk-lane pump on a later + * task (so the sentinel response reaches the client first), and return the + * sentinel marker to send as the RPC result. + */ + startStream( + payload: Buffer, + dispatcher: RelayDispatcher, + context: RequestContext + ): GitResponseStreamMarker { + const streamId = this.register() + const chunks = encodeChunks(payload) + // Why: kick the pump off the response task so the client sees the sentinel + // (and can subscribe/reassemble) before the first chunk frame arrives. + setImmediate(() => { + void this.pump(streamId, chunks, dispatcher, context) + }) + return { + __orcaGitResponseStream: { streamId, totalBytes: payload.length, chunkCount: chunks.length } + } + } + + private async pump( + streamId: number, + chunks: string[], + dispatcher: RelayDispatcher, + context: RequestContext + ): Promise { + const entry = this.streams.get(streamId) + if (!entry) { + return + } + const clientId = context.clientId + let seq = 0 + let endReason: 'end' | 'aborted' | 'stale' = 'end' + try { + for (seq = 0; seq < chunks.length; seq += 1) { + if (context.isStale()) { + endReason = 'stale' + break + } + if (entry.aborted) { + endReason = 'aborted' + break + } + // Why: credit window — the client acks each chunk, bounding how many + // bulk bytes a keystroke echo can queue behind on the shared channel. + while ( + seq - entry.ackedThroughSeq > STREAM_ACK_WINDOW_CHUNKS && + !context.isStale() && + !entry.aborted + ) { + await this.waitForAck(streamId) + } + if (context.isStale()) { + endReason = 'stale' + break + } + if (entry.aborted) { + endReason = 'aborted' + break + } + // Why: notifyBulk waits out sink saturation so chunk frames never pile + // up in the outbound pipe ahead of interactive pty.data frames. + await dispatcher.notifyBulk( + 'git.responseChunk', + { streamId, seq, data: chunks[seq] }, + { + clientId + } + ) + } + if (endReason === 'end') { + dispatcher.notify('git.responseEnd', { streamId }) + } + } catch (err) { + if (!context.isStale() && !entry.aborted) { + dispatcher.notify('git.responseError', { + streamId, + message: err instanceof Error ? err.message : String(err) + }) + } + } finally { + this.streams.delete(streamId) + } + } + + disposeAll(): void { + for (const id of Array.from(this.streams.keys())) { + this.abort(id) + } + this.streams.clear() + } +} diff --git a/src/relay/protocol.ts b/src/relay/protocol.ts index 267f6ef480f..4df0f7bd9b5 100644 --- a/src/relay/protocol.ts +++ b/src/relay/protocol.ts @@ -59,6 +59,27 @@ export const STREAM_ACK_WINDOW_CHUNKS = 4 * (and its open file handle) forever. */ export const STREAM_ACK_STALL_RECHECK_MS = 1_000 +// ── Git response streaming (see docs/relay-git-response-stream-design.md) ── + +/** Serialized-JSON size above which a streamable git response (diff family + + * exec) is chunked onto the bulk lane instead of one JSON-RPC frame, so a large + * diff cannot head-of-line-block interactive pty.data echo on the shared SSH + * channel. Below this, single-frame is cheaper and avoids stream overhead. */ +export const GIT_RESPONSE_STREAM_THRESHOLD = 256 * 1024 + +/** Per-chunk size (UTF-8 bytes of the serialized result) for git response + * streaming. Independent from STREAM_CHUNK_SIZE — this offset math is not + * shared with fs streams, so tuning it here is cross-version safe as long as + * the client reassembles by concatenation (it does not depend on chunk size). */ +export const GIT_RESPONSE_CHUNK_SIZE = 128 * 1024 + +/** Sentinel result returned in place of a large git response: the real payload + * follows as git.responseChunk frames on the bulk lane. Old relays never emit + * this, so a new client falls back to the plain result they return. */ +export type GitResponseStreamMarker = { + __orcaGitResponseStream: { streamId: number; totalBytes: number; chunkCount: number } +} + export const RelayErrorCode = { TooManyStreams: -33006, StreamProtocolError: -33007 diff --git a/src/relay/relay.ts b/src/relay/relay.ts index 65f90e88628..059f83034af 100644 --- a/src/relay/relay.ts +++ b/src/relay/relay.ts @@ -454,10 +454,7 @@ async function main(): Promise { const ptyHandler = new PtyHandler(dispatcher, graceTimeMs) const fsHandler = new FsHandler(dispatcher, context) - // Why: GitHandler registers its own request handlers on construction, - // so we hold the reference only for potential future disposal. - const _gitHandler = new GitHandler(dispatcher, context) - void _gitHandler + const gitHandler = new GitHandler(dispatcher, context) const _preflightHandler = new PreflightHandler(dispatcher) const _externalAutomationsHandler = new ExternalAutomationsHandler(dispatcher) @@ -1013,6 +1010,7 @@ async function main(): Promise { dispatcher.dispose() ptyHandler.dispose() fsHandler.dispose() + gitHandler.dispose() hookServer.stop() // Why: Node's Unix server.close() can unlink the listen path. If the path // was externally removed and rebound by a newer relay, closing this older