diff --git a/src/renderer/src/runtime/remote-terminal-stream-watchdog.test.ts b/src/renderer/src/runtime/remote-terminal-stream-watchdog.test.ts new file mode 100644 index 00000000000..6e25438ec34 --- /dev/null +++ b/src/renderer/src/runtime/remote-terminal-stream-watchdog.test.ts @@ -0,0 +1,127 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS, + createRemoteTerminalStreamWatchdog +} from './remote-terminal-stream-watchdog' + +describe('remote terminal stream watchdog delivery deadline', () => { + beforeEach(() => { + vi.useFakeTimers() + }) + + afterEach(() => { + vi.useRealTimers() + }) + + it('anchors the deadline to the oldest unsettled delivery instead of the last settled sibling', () => { + const onStall = vi.fn() + const watchdog = createRemoteTerminalStreamWatchdog(onStall) + + watchdog.beginOutputDelivery(100) + for (let tick = 0; tick < 3; tick += 1) { + vi.advanceTimersByTime(9_000) + const settle = watchdog.beginOutputDelivery(10) + settle() + } + expect(onStall).not.toHaveBeenCalled() + + vi.advanceTimersByTime(REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS - 27_000) + + expect(onStall).toHaveBeenCalledTimes(1) + expect(onStall.mock.calls[0]?.[0]).toMatchObject({ reason: 'delivery-credit-timeout' }) + }) + + it('stays quiet while continuous output keeps settling and acknowledging', () => { + const onStall = vi.fn() + const watchdog = createRemoteTerminalStreamWatchdog(onStall) + + // Each chunk parses 50ms and is acked 100ms after it lands, so bytes are always in flight. + for ( + let elapsedMs = 0; + elapsedMs < REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS * 4; + elapsedMs += 20 + ) { + const settle = watchdog.beginOutputDelivery(1_000) + setTimeout(settle, 50) + setTimeout(() => watchdog.recordOutputAcknowledged(1_000), 100) + vi.advanceTimersByTime(20) + } + + expect(onStall).not.toHaveBeenCalled() + }) + + it('trips 30s after the oldest unacknowledged chunk once ACKs stop, even while output keeps settling', () => { + const onStall = vi.fn() + const watchdog = createRemoteTerminalStreamWatchdog(onStall) + + // Healthy start: deliveries settle and are ACKed. + for (let tick = 0; tick < 50; tick += 1) { + watchdog.beginOutputDelivery(1_000)() + watchdog.recordOutputAcknowledged(1_000) + vi.advanceTimersByTime(20) + } + const ackStoppedAtMs = Date.now() + // ACKs stop (only part of the oldest chunk is ever acknowledged) while output keeps settling. + let partialAckSent = false + let stalledAfterMs: number | null = null + for (let tick = 0; tick < 2_000 && stalledAfterMs === null; tick += 1) { + watchdog.beginOutputDelivery(1_000)() + if (!partialAckSent) { + watchdog.recordOutputAcknowledged(500) + partialAckSent = true + } + vi.advanceTimersByTime(20) + if (onStall.mock.calls.length > 0) { + stalledAfterMs = Date.now() - ackStoppedAtMs + } + } + + expect(onStall).toHaveBeenCalledTimes(1) + expect(onStall.mock.calls[0]?.[0]).toMatchObject({ reason: 'delivery-credit-timeout' }) + expect(stalledAfterMs).toBeGreaterThanOrEqual(REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS) + expect(stalledAfterMs).toBeLessThanOrEqual(REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS + 20) + }) + + it('drains a single ACK covering many tiny chunks without quadratic cost', () => { + vi.useRealTimers() + const onStall = vi.fn() + const watchdog = createRemoteTerminalStreamWatchdog(onStall) + const chunkCount = 200_000 + for (let index = 0; index < chunkCount; index += 1) { + watchdog.beginOutputDelivery(1)() + } + + const startedAt = performance.now() + watchdog.recordOutputAcknowledged(chunkCount) + const elapsedMs = performance.now() - startedAt + watchdog.dispose() + + // Repeated Array.shift() takes seconds here; a head index takes a few ms. + expect(elapsedMs).toBeLessThan(100) + }) + + it('stays armed while parsed bytes remain unacknowledged to the host', () => { + const onStall = vi.fn() + const watchdog = createRemoteTerminalStreamWatchdog(onStall) + + watchdog.beginOutputDelivery(100)() + vi.advanceTimersByTime(REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS) + + expect(onStall).toHaveBeenCalledTimes(1) + expect(onStall.mock.calls[0]?.[0]).toMatchObject({ + outstandingDeliveryBytes: 0, + reason: 'delivery-credit-timeout' + }) + }) + + it('disarms once the acknowledgement reaches the transport', () => { + const onStall = vi.fn() + const watchdog = createRemoteTerminalStreamWatchdog(onStall) + + watchdog.beginOutputDelivery(100)() + watchdog.recordOutputAcknowledged(100) + vi.advanceTimersByTime(REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS * 2) + + expect(onStall).not.toHaveBeenCalled() + }) +}) diff --git a/src/renderer/src/runtime/remote-terminal-stream-watchdog.ts b/src/renderer/src/runtime/remote-terminal-stream-watchdog.ts index c5b4f2d8725..d7f08bda3d6 100644 --- a/src/renderer/src/runtime/remote-terminal-stream-watchdog.ts +++ b/src/renderer/src/runtime/remote-terminal-stream-watchdog.ts @@ -24,9 +24,12 @@ export function createRemoteTerminalStreamWatchdog( let responseTimer: ReturnType | null = null let deliveryTimer: ReturnType | null = null let outstandingDeliveryBytes = 0 + // Unsettled deliveries in arrival order; the oldest one's start is the stall anchor. + const unsettledDeliveries = new Set<{ startedAtMs: number }>() // Host credit reopens on ACK frames; parsed-but-unacked bytes remain credit whose loss can stall output. - let unacknowledgedBytes = 0 - let deliveryPendingSinceMs: number | null = null + // Why a head index: one ACK can cover thousands of tiny frames, and repeated shift() is O(n) each. + let unacknowledgedChunks: { bytes: number; startedAtMs: number }[] = [] + let unacknowledgedHead = 0 let lastInboundAtMs = Date.now() let commandResponseProbePending = false let disposed = false @@ -60,29 +63,55 @@ export function createRemoteTerminalStreamWatchdog( reason }) } - // Anchor deadline: sibling settles must not postpone it forever, and clearing at zero parse credit would require the very inbound output a stall blocks. - const syncDeliveryTimer = (): void => { - if (disposed || outstandingDeliveryBytes + unacknowledgedBytes <= 0) { - clearDeliveryTimer() - deliveryPendingSinceMs = null - return - } - deliveryPendingSinceMs ??= Date.now() - if (deliveryTimer) { - return - } + const oldestPendingAtMs = (): number => { + const oldestDelivery = unsettledDeliveries.values().next().value + return Math.min( + oldestDelivery?.startedAtMs ?? Infinity, + unacknowledgedChunks[unacknowledgedHead]?.startedAtMs ?? Infinity + ) + } + // Why oldest pending: flowing output always has bytes in flight; only the oldest byte not moving is a stall. + // Why arm once and recheck on fire: the anchor only moves later, so re-arming per frame would churn timers on the output hot path. + const armDeliveryTimer = (anchorMs: number): void => { const remainingMs = Math.max( 0, - REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS - (Date.now() - deliveryPendingSinceMs) + REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS - (Date.now() - anchorMs) ) - deliveryTimer = setTimeout(() => trip('delivery-credit-timeout'), remainingMs) + deliveryTimer = setTimeout(() => { + deliveryTimer = null + const currentAnchorMs = oldestPendingAtMs() + if (disposed || currentAnchorMs === Infinity) { + return + } + if (Date.now() - currentAnchorMs >= REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS) { + trip('delivery-credit-timeout') + return + } + armDeliveryTimer(currentAnchorMs) + }, remainingMs) + } + const ensureDeliveryTimer = (): void => { + if (disposed || deliveryTimer) { + return + } + const anchorMs = oldestPendingAtMs() + if (anchorMs !== Infinity) { + armDeliveryTimer(anchorMs) + } } return { beginOutputDelivery(bytes) { + if (disposed) { + return () => {} + } + const delivery = { startedAtMs: Date.now() } outstandingDeliveryBytes += bytes - unacknowledgedBytes += bytes - syncDeliveryTimer() + unsettledDeliveries.add(delivery) + if (bytes > 0) { + unacknowledgedChunks.push({ bytes, startedAtMs: delivery.startedAtMs }) + } + ensureDeliveryTimer() let settled = false return () => { if (settled || disposed) { @@ -90,15 +119,34 @@ export function createRemoteTerminalStreamWatchdog( } settled = true outstandingDeliveryBytes = Math.max(0, outstandingDeliveryBytes - bytes) - syncDeliveryTimer() + unsettledDeliveries.delete(delivery) } }, recordOutputAcknowledged(bytes) { if (disposed) { return } - unacknowledgedBytes = Math.max(0, unacknowledgedBytes - bytes) - syncDeliveryTimer() + let remaining = bytes + while (remaining > 0 && unacknowledgedHead < unacknowledgedChunks.length) { + const oldest = unacknowledgedChunks[unacknowledgedHead] + const consumed = Math.min(oldest.bytes, remaining) + oldest.bytes -= consumed + remaining -= consumed + if (oldest.bytes === 0) { + unacknowledgedHead += 1 + } + } + // Why compact only past half: keeps slice cost amortized O(1) per consumed chunk. + if (unacknowledgedHead === unacknowledgedChunks.length) { + unacknowledgedChunks = [] + unacknowledgedHead = 0 + } else if ( + unacknowledgedHead > 1024 && + unacknowledgedHead * 2 > unacknowledgedChunks.length + ) { + unacknowledgedChunks = unacknowledgedChunks.slice(unacknowledgedHead) + unacknowledgedHead = 0 + } }, completeCommandResponseProbe() { commandResponseProbePending = false @@ -124,8 +172,9 @@ export function createRemoteTerminalStreamWatchdog( clearResponseTimer() clearDeliveryTimer() outstandingDeliveryBytes = 0 - unacknowledgedBytes = 0 - deliveryPendingSinceMs = null + unsettledDeliveries.clear() + unacknowledgedChunks = [] + unacknowledgedHead = 0 } } }