fix(relay): stream large git responses on the bulk lane#8067
Conversation
|
Follow-up (commit 8c47b35) addresses self-review findings:
Confirmed-correct (no change): cross-version marker detection, seq ordering, base64 boundaries, stall/abort/dispose wakeups, resource cleanup, and error propagation (thrown git errors reject before the streaming path). |
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (9)
🚧 Files skipped from review as they are similar to previous changes (8)
📝 WalkthroughWalkthroughAdds opt-in streaming for oversized Git diff, branch diff, commit diff, and exec responses. The relay chunks base64-encoded payloads with acknowledgement-based flow control, while the SSH client reassembles and validates streamed responses, including cancellation and disposal handling. Provider Git operations now use the streaming helper. Tests update request expectations and add congestion, backpressure, timeout, and streamed-result equivalence coverage. 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 9fd32e10-fbe0-458f-86ef-9cf683243625
📒 Files selected for processing (9)
src/main/providers/ssh-git-provider.test.tssrc/main/providers/ssh-git-provider.tssrc/main/ssh/relay-protocol.tssrc/main/ssh/ssh-git-response-stream-reader.tssrc/relay/git-handler.tssrc/relay/git-response-pty-echo-backpressure.integration.test.tssrc/relay/git-response-stream.tssrc/relay/protocol.tssrc/relay/relay.ts
| const handleChunk = (p: Record<string, unknown>): 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 }) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Guard the ack mux.notify call like cancel() does.
cancel() (line ~62-70) wraps its mux.notify call in !mux.isDisposed() + try/catch with a "best-effort" comment, implying mux.notify can throw once the multiplexer is disposed. The per-chunk ack notify here has no such guard, so a disposal race during chunk processing could throw synchronously inside the notification callback.
🛡️ Proposed fix: guard the ack notify like `cancel()`
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 })
+ if (!mux.isDisposed()) {
+ try {
+ mux.notify('git.responseAck', { streamId: streamIdRef.current, seq })
+ } catch {
+ // best-effort, mirrors cancel()
+ }
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| const handleChunk = (p: Record<string, unknown>): 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 handleChunk = (p: Record<string, unknown>): 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. | |
| if (!mux.isDisposed()) { | |
| try { | |
| mux.notify('git.responseAck', { streamId: streamIdRef.current, seq }) | |
| } catch { | |
| // best-effort, mirrors cancel() | |
| } | |
| } | |
| } |
| this.dispatcher.onNotification('git.responseAck', (p) => this.responseAck(p)) | ||
| this.dispatcher.onNotification('git.cancelResponseStream', (p) => this.cancelResponseStream(p)) | ||
| } | ||
|
|
||
| private responseAck(params: Record<string, unknown>): 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<string, unknown>): void { | ||
| const streamId = params.streamId | ||
| if (typeof streamId === 'number') { | ||
| this.responseStreams.abort(streamId) | ||
| } | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
ast-grep outline src/relay/dispatcher.ts --items all --type class,function --match 'onNotification|attachClient|detachClient'
rg -n -B2 -A8 'onNotification' src/relay/dispatcher.tsRepository: stablyai/orca
Length of output: 547
🏁 Script executed:
# Find NotificationHandler type definition
rg -n 'type NotificationHandler|interface NotificationHandler' src/relay/dispatcher.ts
# Find where notification handlers are invoked
rg -n 'notificationHandlers\.' src/relay/dispatcher.ts
# Compare onRequest signature and how it passes context
rg -n -B2 -A10 'onRequest' src/relay/dispatcher.ts | head -50Repository: stablyai/orca
Length of output: 850
🏁 Script executed:
# Check git-handler.ts to see onRequest usage with context
rg -n -B3 -A8 'onRequest' src/relay/git-handler.ts | head -60Repository: stablyai/orca
Length of output: 4102
🏁 Script executed:
# Understand streamId generation - check responseStreams type
rg -n 'class ResponseStreams|interface ResponseStreams|new ResponseStreams' src/relay/ -A 20Repository: stablyai/orca
Length of output: 151
🏁 Script executed:
# Find where notificationHandlers are invoked/dispatched
rg -n -B5 -A15 'notificationHandlers.*handler\|handler\(.*context' src/relay/dispatcher.tsRepository: stablyai/orca
Length of output: 151
🏁 Script executed:
# Look at the full dispatcher dispatch logic to understand how notifications are processed
ast-grep outline src/relay/dispatcher.tsRepository: stablyai/orca
Length of output: 1279
🏁 Script executed:
# Find and read the handleNotification method implementation
rg -n -B2 -A20 'private handleNotification|handleNotification\(' src/relay/dispatcher.ts | head -60Repository: stablyai/orca
Length of output: 1674
🏁 Script executed:
# Also check how handleRequest is implemented for comparison
rg -n -B2 -A20 'private handleRequest|handleRequest\(' src/relay/dispatcher.ts | head -80Repository: stablyai/orca
Length of output: 1955
🏁 Script executed:
# Find ResponseStreams implementation
rg -n 'class ResponseStreams' src/relay/ -A 30Repository: stablyai/orca
Length of output: 151
🏁 Script executed:
# Check how streamId is generated and tracked
rg -n 'nextStreamId\|streamId.*=' src/relay/git-handler.tsRepository: stablyai/orca
Length of output: 151
🏁 Script executed:
# See if responseStreams tracks client ownership
rg -n 'recordAck\|\.abort' src/relay/ -B 2 -A 8Repository: stablyai/orca
Length of output: 151
🏁 Script executed:
# Find files related to response streams
fd -t f 'response' src/relay/Repository: stablyai/orca
Length of output: 249
🏁 Script executed:
# Find where responseStreams is defined in git-handler.ts
rg -n 'responseStreams' src/relay/git-handler.ts -B 3 -A 3Repository: stablyai/orca
Length of output: 1790
🏁 Script executed:
# Look for imports and class definitions containing 'stream' or 'response'
rg -n 'import.*[Ss]tream|class.*[Ss]tream' src/relay/git-handler.tsRepository: stablyai/orca
Length of output: 505
🏁 Script executed:
# Read the GitResponseStreamRegistry implementation
cat -n src/relay/git-response-stream.tsRepository: stablyai/orca
Length of output: 7203
Add client ownership validation to stream ack/cancel handlers to prevent cross-client interference.
The NotificationHandler type signature includes a context parameter with clientId, and handleNotification in the dispatcher does pass this context (line 480–483 in dispatcher.ts). However, the git.responseAck and git.cancelResponseStream handlers in git-handler.ts are registered to ignore this context:
this.dispatcher.onNotification('git.responseAck', (p) => this.responseAck(p))
this.dispatcher.onNotification('git.cancelResponseStream', (p) => this.cancelResponseStream(p))Since GitResponseStreamRegistry stores streamId as a simple global counter (line 34 in git-response-stream.ts) without tracking which client owns it, any connected client can ack or abort any other client's active stream by knowing its streamId. Although the pump has staleness checks and a stall timeout that eventually clean up, a misbehaving client can still interfere with concurrent streams from other clients by sending false acks.
Fix: Update GitResponseStreamEntry to store clientId, validate ownership in the handlers before delegating to recordAck and abort, and update the handlers to accept and use the context parameter.
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 <help@stably.ai>
… for git response streams Address self-review findings on the git response streaming reader: - MEDIUM: mux.request's timeout only bounded the sentinel; the reassembly phase had no deadline, so a relay pump that broke on staleness (sends no responseEnd) while the SSH channel stayed up would hang the client forever. Add a client-side inactivity timeout that resets on each chunk and rejects if no stream frame arrives within the window (default 30s, matching the mux request timeout). New regression test drives a post-sentinel stall. - LOW: widen the options type to declare timeoutMs (sentinel) and a distinct inactivityTimeoutMs (reassembly); forward only mux-request options to the sentinel request. - LOW: cap the pre-sentinel pending-frame buffer (MAX_PENDING_FRAMES) so concurrent streams cannot transiently buffer each other's chunks unbounded; foreign frames are dropped on drain anyway. Co-authored-by: Orca <help@stably.ai>
8c47b35 to
31d21c6
Compare
1 participant
Summary
All SSH relay JSON-RPC traffic shares one ordered SSH channel. Large git responses were sent as a single JSON-RPC frame, so they head-of-line-blocked interactive
pty.dataecho: opening a big diff (e.g. a lockfile) over SSH while typing in another pane stalled the echo for multiple seconds.src/relay/git-handler.ts—git.diff/git.branchDiff/git.commitDiff/git.execresponses (up toMAX_GIT_BUFFER= 10MB) were sent as onesendResponseframe on the regular lane.src/relay/protocol.tsencodeJsonRpcFramedoes a synchronousJSON.stringify+Buffer.fromof the whole payload on the relay event loop (tens of ms at 10MB), during which nopty.datais flushed, then the multi-MB write sits in the outbound pipe ahead of the echo.The 2026-07 fix for this exact problem on file streams (PR #7601) added
RelayDispatcher.notifyBulk()— a per-client serialized bulk lane honoringwrite() === false+waitWriteDrain— plus an ack credit window. But onlyfs.readFileStreamused it; git responses never touched the bulk lane.This PR streams the diff family +
git.execthrough the same bulk lane in chunks, mirroring the fs pattern: the relay returns a small sentinel result and pumpsgit.responseChunkframes with the sameSTREAM_ACK_WINDOW_CHUNKScredit window; the client reassembles andJSON.parses. Small responses stay single-frame.STREAM_CHUNK_SIZEis untouched (it is baked into both sides' fs-stream offset math). Git chunking uses its ownGIT_RESPONSE_CHUNK_SIZE, and the client reassembles by concatenation, so chunk size is not part of any cross-version offset math.git.responseAckacks are wired for backpressure; the pump wakes on client detach / dispose so a vanished client cannot strand it.Cross-version compatibility
Opt-in capability gate, no handshake change (same mechanism fs streaming used). The client adds
__streamResponse: true; the relay decides whether to stream.__streamResponse→ relay always single-frame (today's behavior)__streamResponse, returns plainresult; client'sisGitResponseStreamMarkeris false → returns it directly(In practice the relay's version handshake already refuses mismatched content-hash versions, so the same-version path dominates; the gate makes the streaming layer robust regardless.)
Byte-bound before/after evidence
New deterministic integration test (
src/relay/git-response-pty-echo-backpressure.integration.test.ts) models the SSH channel as a congestible in-memory pipe and asserts bytes queued ahead of apty.dataecho, not wall-clock:queuedBytesAheadOfEcho > 512 KB(the actual staged-diff frame is ~1.4 MB).queuedBytesAheadOfEcho < 2 × framed-chunk ≈ 342 KB, bounded regardless of total response size.Both HOL tests + the equality test pass; verified stable across 6 back-to-back runs (an early client-side subscribe-after-request race that dropped the first chunk was caught by this test and fixed by subscribing before the sentinel resolves and buffering pending frames, mirroring
readFileViaStream).Screenshots
No visual change.
Testing
pnpm typecheckgit-handler,dispatcher,fs-stream-pty-echo-backpressure,protocol-handshake,relay-protocol,ssh-channel-multiplexer,ssh-git-provider, and the new git-response integration test (183 relevant tests pass)oxlinton changed files cleanpnpm check:max-lines-ratchetOKAI Review Report
Reviewed for cross-version compatibility (the four client/relay version combinations above), protocol/framing correctness (seq ordering, ack credit window, abort/stale/dispose handling, base64 chunk boundaries so multi-byte UTF-8 never splits), races (client subscribes before awaiting the sentinel and drains a pending-frame queue), and resource cleanup (stream registry deletion on end/abort/dispose, unsubscribers, no file handles involved). Cross-platform: the relay runs on Linux/macOS/Windows remote hosts; this change is pure JS buffer/JSON handling with no path or shell assumptions, and the test uses
mkdirSync(not shelling out tomkdir). Git operations remain provider-agnostic (GitHub/GitLab). The only behavioral change to existing methods is response transport; thrown git errors still reject the RPC (they never reach the streaming path).Security Audit
Input handling: chunk
seq/dataare type-checked on both sides; out-of-order or malformed chunks fail the stream instead of silently corrupting. No new command execution or path handling (the streamed payload is the already-computed git result). The__streamResponseflag is a boolean gate only; it cannot influence which git command runs. No secrets or IPC surface added. The ack window + detach/dispose wakeups bound memory and prevent a stalled pump from stranding resources.Notes
Known separate gap (out of scope, not expanded here): the
fs.writeFileclient → relay direction has no bulk-lane backpressure. That is a distinct upload path and is left as follow-up.Made with Orca 🐋