starred/orca
1
0
Fork 0
mirror of https://github.com/stablyai/orca.git synced 2026-10-18 00:03:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cap remote ACK pending terminal output

Browse Source
This commit is contained in:
Neil
2026-06-09 17:06:54 -07:00
parent 274133150e
commit ca5fb1436c
2 changed files with 252 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/runtime/rpc/methods/terminal.ts
+82 -1
View File
@@ -75,6 +75,9 @@ type TerminalMultiplexStream = {
ackInFlightBytes: number
buffering: boolean
ackPendingOutput: TerminalOutputFrameChunk[]
ackPendingOutputBytes: number
ackPendingOutputOverflowed: boolean
ackRecoverySnapshotInFlight: boolean
pendingOutput: TerminalOutputChunk[]
pendingOutputBytes: number
pendingOutputOverflowed: boolean
@@ -249,6 +252,26 @@ function appendPendingMultiplexOutput(
stream.pendingOutputOverflowed ||= trimmed.overflowed
}
function appendAckPendingOutput(
stream: TerminalMultiplexStream,
chunk: TerminalOutputFrameChunk
): void {
stream.ackPendingOutput.push(chunk)
stream.ackPendingOutputBytes += chunk.bytes.byteLength
let omittedChunkCount = 0
while (
stream.ackPendingOutputBytes > TERMINAL_MULTIPLEX_PENDING_MAX_BYTES &&
omittedChunkCount < stream.ackPendingOutput.length
) {
stream.ackPendingOutputBytes -= stream.ackPendingOutput[omittedChunkCount]!.bytes.byteLength
omittedChunkCount += 1
}
if (omittedChunkCount > 0) {
stream.ackPendingOutput.splice(0, omittedChunkCount)
stream.ackPendingOutputOverflowed = true
}
}
function trimPendingOutputToBudget(
pendingOutput: (string | TerminalOutputChunk)[],
pendingOutputBytes: number
@@ -949,15 +972,63 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
return
}
if (
stream.ackPendingOutputOverflowed ||
stream.ackPendingOutput.length > 0 ||
!canSendAckGatedOutput(stream, chunk.bytes.byteLength)
) {
stream.ackPendingOutput.push(chunk)
appendAckPendingOutput(stream, chunk)
return
}
sendAckGatedOutput(stream, chunk)
}
const sendAckRecoverySnapshot = async (stream: TerminalMultiplexStream): Promise<void> => {
if (
closed ||
streams.get(stream.streamId) !== stream ||
stream.ackRecoverySnapshotInFlight
) {
return
}
stream.ackRecoverySnapshotInFlight = true
try {
const serialized = await serializeBudgetedRequestedSnapshot(runtime, stream.ptyId, 0)
if (closed || streams.get(stream.streamId) !== stream) {
return
}
const size = runtime.getTerminalSize(stream.ptyId)
const displayMode = runtime.getMobileDisplayMode(stream.ptyId)
// Why: dropped ACK-pending output means live frames are no longer a
// complete replay. Send a fresh model snapshot before resuming output.
sendSnapshotFrames((opcode, payload) => sendFrame(stream.streamId, opcode, payload), {
kind: 'scrollback',
cols: serialized?.cols ?? size?.cols ?? 80,
rows: serialized?.rows ?? size?.rows ?? 24,
displayMode,
reason: 'ack-pending-overflow',
seq: serialized?.seq,
source: serialized?.source,
truncated: true,
truncatedByByteBudget: serialized?.truncatedByByteBudget,
data: serialized?.data ?? ''
})
stream.ackPendingOutputOverflowed = false
} catch (error) {
sendStreamError(
stream.streamId,
error instanceof Error ? error.message : 'Remote terminal recovery snapshot failed.'
)
} finally {
if (streams.get(stream.streamId) === stream) {
stream.ackRecoverySnapshotInFlight = false
flushAckPendingOutput(stream)
}
}
}
const flushAckPendingOutput = (stream: TerminalMultiplexStream): void => {
if (stream.ackPendingOutputOverflowed) {
void sendAckRecoverySnapshot(stream)
return
}
let flushed = 0
while (
flushed < stream.ackPendingOutput.length &&
@@ -968,6 +1039,10 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
}
if (flushed > 0) {
stream.ackPendingOutput.splice(0, flushed)
stream.ackPendingOutputBytes = stream.ackPendingOutput.reduce(
(total, pending) => total + pending.bytes.byteLength,
0
)
}
}
const acknowledgeOutput = (stream: TerminalMultiplexStream, bytes: number): void => {
@@ -989,6 +1064,9 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
ackTotalInFlightBytes = Math.max(0, ackTotalInFlightBytes - stream.ackInFlightBytes)
stream.ackInFlightBytes = 0
stream.ackPendingOutput = []
stream.ackPendingOutputBytes = 0
stream.ackPendingOutputOverflowed = false
stream.ackRecoverySnapshotInFlight = false
stream.unsubscribeData()
stream.unsubscribeResize()
stream.unsubscribeFit()
@@ -1199,6 +1277,9 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
ackInFlightBytes: 0,
buffering: true,
ackPendingOutput: [],
ackPendingOutputBytes: 0,
ackPendingOutputOverflowed: false,
ackRecoverySnapshotInFlight: false,
pendingOutput: [],
pendingOutputBytes: 0,
pendingOutputOverflowed: false,
src/main/runtime/rpc/terminal-multiplex.test.ts
+170
View File
@@ -489,6 +489,176 @@ describe('terminal multiplex RPC', () => {
await dispatchPromise
})
it('caps stalled ACK output and snapshots before resuming retained tail frames', async () => {
const messages: string[] = []
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
const handlers = new Map<
number,
(frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void
>()
const cleanups = new Map<string, () => void>()
const dataListenerRef: {
current?: (data: string, meta?: { seq?: number; rawLength?: number }) => void
} = {}
const runtime = stubRuntime({
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
serializeTerminalBuffer: vi
.fn()
.mockResolvedValueOnce({ data: 'initial snapshot', cols: 120, rows: 40 })
.mockResolvedValue({ data: 'recovered snapshot', cols: 120, rows: 40, seq: 99 }),
getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }),
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
subscribeToTerminalData: vi.fn(
(
_: string,
listener: (data: string, meta?: { seq?: number; rawLength?: number }) => void
) => {
dataListenerRef.current = listener
return vi.fn()
}
),
subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()),
getTerminalFitOverride: vi.fn().mockReturnValue(null),
getDriver: vi.fn().mockReturnValue({ kind: 'idle' }),
registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
cleanups.set(id, cleanup)
}),
cleanupSubscription: vi.fn((id: string) => {
const cleanup = cleanups.get(id)
cleanups.delete(id)
cleanup?.()
}),
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {})),
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
(msg) => messages.push(msg),
{
connectionId: 'conn-ack-overflow',
sendBinary: (bytes) => binaryFrames.push(bytes),
registerBinaryStreamHandler: (streamId, handler) => {
handlers.set(streamId, handler)
return () => handlers.delete(streamId)
}
}
)
await vi.waitFor(() =>
expect(messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true)
)
handlers.get(0)?.(
decodeTerminalStreamFrame(
encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Subscribe,
streamId: 0,
seq: 1,
payload: encodeTerminalStreamJson({
streamId: 17,
terminal: 'terminal-1',
client: { id: 'desktop-1', type: 'desktop' },
viewport: { cols: 120, rows: 40 },
capabilities: { ackOutput: 1 }
})
})
)!
)
await vi.waitFor(() =>
expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true)
)
binaryFrames.splice(0)
const output = 'x'.repeat(3 * 1024 * 1024)
dataListenerRef.current?.(output, { seq: output.length, rawLength: output.length })
const initialOutputFrames = binaryFrames
.map((frame) => decodeTerminalStreamFrame(frame))
.filter((frame) => frame?.opcode === TerminalStreamOpcode.Output)
const initialBytes = initialOutputFrames.reduce(
(total, frame) => total + (frame?.payload.byteLength ?? 0),
0
)
expect(initialBytes).toBeLessThanOrEqual(512 * 1024)
handlers.get(17)?.(
decodeTerminalStreamFrame(
encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Input,
streamId: 17,
seq: 2,
payload: encodeTerminalStreamText('still interactive\r')
})
)!
)
await vi.waitFor(() =>
expect(runtime.sendTerminal).toHaveBeenCalledWith('terminal-1', {
text: 'still interactive\r',
enter: false,
interrupt: false
})
)
binaryFrames.splice(0)
handlers.get(17)?.(
decodeTerminalStreamFrame(
encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Ack,
streamId: 17,
seq: 3,
payload: encodeTerminalStreamJson({ bytes: initialBytes })
})
)!
)
await vi.waitFor(() =>
expect(
binaryFrames
.map((frame) => decodeTerminalStreamFrame(frame))
.some((frame) => {
if (frame?.opcode !== TerminalStreamOpcode.SnapshotStart) {
return false
}
const payload = decodeTerminalStreamJson<{ reason?: string }>(frame.payload)
return payload?.reason === 'ack-pending-overflow'
})
).toBe(true)
)
const drainFrames = binaryFrames.map((frame) => decodeTerminalStreamFrame(frame))
const recoveryStartIndex = drainFrames.findIndex((frame) => {
if (frame?.opcode !== TerminalStreamOpcode.SnapshotStart) {
return false
}
const payload = decodeTerminalStreamJson<{ reason?: string }>(frame.payload)
return payload?.reason === 'ack-pending-overflow'
})
const firstOutputAfterAckIndex = drainFrames.findIndex(
(frame) => frame?.opcode === TerminalStreamOpcode.Output
)
expect(recoveryStartIndex).toBeGreaterThanOrEqual(0)
expect(firstOutputAfterAckIndex).toBeGreaterThan(recoveryStartIndex)
expect(
drainFrames
.filter((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotChunk)
.map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : ''))
.join('')
).toBe('recovered snapshot')
const outputBytesAfterRecovery = drainFrames
.filter((frame) => frame?.opcode === TerminalStreamOpcode.Output)
.reduce((total, frame) => total + (frame?.payload.byteLength ?? 0), 0)
expect(outputBytesAfterRecovery).toBeLessThanOrEqual(256 * 1024)
runtime.cleanupSubscription('terminal-multiplex:conn-ack-overflow')
await dispatchPromise
})
it('marks multiplex fallback snapshots truncated when the uncursored read is limited', async () => {
const messages: string[] = []
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []