mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 08:03:39 +00:00
fix(terminal): bound recovery snapshot retries
This commit is contained in:
@@ -49,8 +49,10 @@ class FakeMultiplexServer {
|
||||
holdNextRecoverySnapshot = false
|
||||
emptyNextRecoverySnapshot = false
|
||||
unavailableNextRecoverySnapshot = false
|
||||
unavailableRecoverySnapshots = false
|
||||
pendingEscapeTailAnsiNextRecoverySnapshot: string | undefined
|
||||
snapshotRequests: (number | undefined)[] = []
|
||||
unsubscribeRequests = 0
|
||||
private heldManualRequestId: number | null = null
|
||||
private snapshotData = 'INITIAL'
|
||||
|
||||
@@ -90,7 +92,10 @@ class FakeMultiplexServer {
|
||||
this.emptyNextRecoverySnapshot = false
|
||||
this.snapshotData = ''
|
||||
}
|
||||
if (typeof payload?.requestId !== 'number' && this.unavailableNextRecoverySnapshot) {
|
||||
if (
|
||||
typeof payload?.requestId !== 'number' &&
|
||||
(this.unavailableNextRecoverySnapshot || this.unavailableRecoverySnapshots)
|
||||
) {
|
||||
this.unavailableNextRecoverySnapshot = false
|
||||
this.snapshotData = ''
|
||||
recoveryOptions.unavailable = 'no-serializable-buffer'
|
||||
@@ -118,6 +123,10 @@ class FakeMultiplexServer {
|
||||
return
|
||||
}
|
||||
this.sendSnapshot(payload?.requestId, recoveryOptions)
|
||||
return
|
||||
}
|
||||
if (frame.opcode === TerminalStreamOpcode.Unsubscribe) {
|
||||
this.unsubscribeRequests += 1
|
||||
}
|
||||
}
|
||||
|
||||
@@ -234,7 +243,9 @@ describe('remote terminal frame-drop resync', () => {
|
||||
vi.unstubAllGlobals()
|
||||
})
|
||||
|
||||
async function subscribeClient(): Promise<{
|
||||
async function subscribeClient(
|
||||
onTransportClose?: (event: { recoverable: boolean }) => void
|
||||
): Promise<{
|
||||
data: string[]
|
||||
metas: { seq?: number; rawLength?: number; transformed?: boolean }[]
|
||||
snapshots: string[]
|
||||
@@ -257,7 +268,8 @@ describe('remote terminal frame-drop resync', () => {
|
||||
onSnapshot: (chunk, meta) => {
|
||||
snapshots.push(chunk)
|
||||
snapshotMetas.push(meta ?? {})
|
||||
}
|
||||
},
|
||||
onTransportClose
|
||||
}
|
||||
})
|
||||
// Let the initial snapshot round-trip settle.
|
||||
@@ -347,6 +359,47 @@ describe('remote terminal frame-drop resync', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('replaces the stream after five unavailable recovery attempts', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const onTransportClose = vi.fn()
|
||||
const first = await subscribeClient(onTransportClose)
|
||||
const unavailableServer = server
|
||||
unavailableServer.unavailableRecoverySnapshots = true
|
||||
|
||||
unavailableServer.output('aaa')
|
||||
unavailableServer.dropNextOutput = true
|
||||
unavailableServer.output('bbb')
|
||||
unavailableServer.output('ccc')
|
||||
unavailableServer.output('corrupt-live-tail')
|
||||
|
||||
expect(unavailableServer.snapshotRequests).toEqual([undefined])
|
||||
expect(first.snapshots).toEqual(['INITIAL'])
|
||||
expect(first.data).toEqual(['aaa'])
|
||||
|
||||
await vi.advanceTimersByTimeAsync(500 + 1_000 + 2_000 + 4_000)
|
||||
|
||||
expect(unavailableServer.snapshotRequests).toEqual(Array(5).fill(undefined))
|
||||
expect(unavailableServer.unsubscribeRequests).toBe(1)
|
||||
expect(onTransportClose).toHaveBeenCalledOnce()
|
||||
expect(onTransportClose).toHaveBeenCalledWith({ recoverable: true })
|
||||
expect(first.snapshots).toEqual(['INITIAL'])
|
||||
expect(first.data).toEqual(['aaa'])
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(60_000)
|
||||
expect(unavailableServer.snapshotRequests).toHaveLength(5)
|
||||
|
||||
const replacement = await subscribeClient()
|
||||
server.output('later-live-output')
|
||||
expect(subscribe).toHaveBeenCalledTimes(2)
|
||||
expect(replacement.snapshots).toEqual(['INITIAL'])
|
||||
expect(replacement.data).toEqual(['later-live-output'])
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('replays a pending escape tail after clearing an empty recovered screen', async () => {
|
||||
const { snapshots, snapshotMetas } = await subscribeClient()
|
||||
server.emptyNextRecoverySnapshot = true
|
||||
|
||||
@@ -234,6 +234,7 @@ const REMOTE_TERMINAL_RESYNC_TIMEOUT_MS = 10_000
|
||||
// retrying once per incoming chunk would stampede it, so back off instead.
|
||||
const REMOTE_TERMINAL_RESYNC_RETRY_BASE_MS = 500
|
||||
const REMOTE_TERMINAL_RESYNC_RETRY_MAX_MS = 5_000
|
||||
const REMOTE_TERMINAL_RESYNC_MAX_ATTEMPTS = 5
|
||||
// Why: exported so the transport can classify it as benign — the snapshot was
|
||||
// skipped but live output continues, so it must not surface a fatal red banner.
|
||||
export const REMOTE_TERMINAL_SNAPSHOT_TOO_LARGE =
|
||||
@@ -1021,6 +1022,7 @@ class RemoteRuntimeTerminalMultiplexer {
|
||||
private sendResyncSnapshot(stream: RemoteRuntimeMultiplexedTerminalState): void {
|
||||
stream.resyncPendingSend = false
|
||||
this.startResyncTimer(stream)
|
||||
stream.resyncAttempts += 1
|
||||
const sent = this.sendFrame(
|
||||
stream.streamId,
|
||||
TerminalStreamOpcode.SnapshotRequest,
|
||||
@@ -1030,13 +1032,17 @@ class RemoteRuntimeTerminalMultiplexer {
|
||||
// Transport is down; the reconnect path re-subscribes from scratch.
|
||||
clearResyncTimer(stream)
|
||||
stream.resyncInFlight = false
|
||||
stream.resyncAttempts -= 1
|
||||
}
|
||||
}
|
||||
|
||||
// Why: keep the gate shut across the backoff — the post-gap tail is corrupt
|
||||
// either way — and heal even if the flood ends with no further output.
|
||||
private scheduleResyncRetry(stream: RemoteRuntimeMultiplexedTerminalState): void {
|
||||
stream.resyncAttempts += 1
|
||||
if (stream.resyncAttempts >= REMOTE_TERMINAL_RESYNC_MAX_ATTEMPTS) {
|
||||
this.recoverStalledStream(stream)
|
||||
return
|
||||
}
|
||||
const delay = Math.min(
|
||||
REMOTE_TERMINAL_RESYNC_RETRY_MAX_MS,
|
||||
REMOTE_TERMINAL_RESYNC_RETRY_BASE_MS * 2 ** Math.min(stream.resyncAttempts - 1, 4)
|
||||
|
||||
Reference in New Issue
Block a user