mirror of
https://github.com/stablyai/orca.git
synced 2026-10-20 08:03:28 +00:00
Replace ack-stall watchdog with cumulative ACKs + solicited delivery resync
Design review: the 10s blind-reset watchdog decided correctness from a wall-clock threshold. Rework piece 1 into a deterministic two-part design (pieces 2 and 3 — lifecycle-reset counter zeroing and powerMonitor wake IPC — are unchanged): - Cumulative ACKs (TCP-style): the renderer dispatcher now tracks a monotonic per-pty total of processed chars (terminal-pty-ack-gate) and sends it on every ACK alongside the legacy per-chunk delta. Main keeps per-pty sentChars/ackedChars and max-merges received totals — idempotent and reorder-tolerant, so a lost ACK self-heals when any later ACK arrives instead of becoming permanent in-flight debt. Provider (SSH/daemon) backpressure is credited only the derived delta, clamped, never negative. Main tolerates both payload shapes keyed by field presence (dev hot-reload can mix renderer/main versions); totals reset on pty exit and renderer lifecycle reset on both sides. - Solicited resync (replaces the blind reset): when new pty data arrives while that pty's delivery is fully gated and no probe is outstanding, main sends pty:requestDeliveryResync; the renderer replies with its cumulative totals and main reconciles via max-merge, then flushes held pendingData. Event-triggered, verified-state recovery — no wall-clock threshold decides correctness. The only timer is a 5s request/response hygiene timeout that clears the outstanding flag and logs one diagnostic warn per silent streak; it never mutates counters (a renderer that cannot answer has dead IPC — reload is the only cure). The 10s corrective watchdog is deleted. Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
+279
-96
@@ -534,13 +534,16 @@ describe('registerPtyHandlers', () => {
|
||||
|
||||
function getPtyAckDataListener(): (
|
||||
event: unknown,
|
||||
args: { id: string; charCount: number }
|
||||
args: { id: string; charCount?: number; processedChars?: number }
|
||||
) => void {
|
||||
const ackCall = onMock.mock.calls.find((call: unknown[]) => call[0] === 'pty:ackData')
|
||||
if (!ackCall) {
|
||||
throw new Error('missing pty:ackData listener')
|
||||
}
|
||||
return ackCall[1] as (event: unknown, args: { id: string; charCount: number }) => void
|
||||
return ackCall[1] as (
|
||||
event: unknown,
|
||||
args: { id: string; charCount?: number; processedChars?: number }
|
||||
) => void
|
||||
}
|
||||
|
||||
function getPtySetActiveRendererPtyListener(): (
|
||||
@@ -6936,9 +6939,7 @@ describe('registerPtyHandlers', () => {
|
||||
expect(mainWindow.webContents.send).toHaveBeenCalledTimes(32)
|
||||
vi.advanceTimersByTime(1)
|
||||
expect(mainWindow.webContents.send).toHaveBeenCalledTimes(32)
|
||||
// Why: the only remaining timer is the ack-stall watchdog armed by the
|
||||
// gate-blocked flush; the flush loop itself stopped rescheduling.
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
|
||||
pendingPtyCount: 1,
|
||||
pendingChars: 88 * 1024,
|
||||
@@ -6994,15 +6995,44 @@ describe('registerPtyHandlers', () => {
|
||||
}
|
||||
})
|
||||
|
||||
const ACK_STALL_WATCHDOG_WARNING =
|
||||
'[pty] ack stall watchdog: resetting in-flight delivery counters'
|
||||
const DELIVERY_RESYNC_UNANSWERED_WARNING =
|
||||
'[pty] delivery resync probe unanswered — renderer IPC unresponsive'
|
||||
|
||||
function countAckStallWarnings(warnSpy: { mock: { calls: unknown[][] } }): number {
|
||||
return warnSpy.mock.calls.filter((call) => call[0] === ACK_STALL_WATCHDOG_WARNING).length
|
||||
function countResyncUnansweredWarnings(warnSpy: { mock: { calls: unknown[][] } }): number {
|
||||
return warnSpy.mock.calls.filter((call) => call[0] === DELIVERY_RESYNC_UNANSWERED_WARNING)
|
||||
.length
|
||||
}
|
||||
|
||||
/** Saturates one PTY to its 512 KiB in-flight cap so the next flush attempt
|
||||
* is gate-blocked and arms the ack-stall watchdog. Leaves 88 KiB pending. */
|
||||
function getPtyDataSendCalls(): unknown[][] {
|
||||
return mainWindow.webContents.send.mock.calls.filter(
|
||||
(call: unknown[]) => call[0] === 'pty:data'
|
||||
)
|
||||
}
|
||||
|
||||
function getDeliveryResyncProbeCalls(): unknown[][] {
|
||||
return mainWindow.webContents.send.mock.calls.filter(
|
||||
(call: unknown[]) => call[0] === 'pty:requestDeliveryResync'
|
||||
)
|
||||
}
|
||||
|
||||
function getDeliveryResyncResponseListener(): (
|
||||
event: unknown,
|
||||
args: { requestId: number; processedCharsByPty: Record<string, number> }
|
||||
) => void {
|
||||
const responseCall = onMock.mock.calls.find(
|
||||
(call: unknown[]) => call[0] === 'pty:deliveryResyncResponse'
|
||||
)
|
||||
if (!responseCall) {
|
||||
throw new Error('missing pty:deliveryResyncResponse listener')
|
||||
}
|
||||
return responseCall[1] as (
|
||||
event: unknown,
|
||||
args: { requestId: number; processedCharsByPty: Record<string, number> }
|
||||
) => void
|
||||
}
|
||||
|
||||
/** Saturates one PTY to its 512 KiB in-flight cap so delivery is fully
|
||||
* gated for that PTY. Leaves 88 KiB pending and no timers scheduled. */
|
||||
async function spawnAndSaturateRendererDeliveryGate(
|
||||
mockProc: ReturnType<typeof createMockProc>
|
||||
): Promise<{ id: string }> {
|
||||
@@ -7021,54 +7051,38 @@ describe('registerPtyHandlers', () => {
|
||||
return spawnResult
|
||||
}
|
||||
|
||||
it('resets the renderer delivery gate after gate-blocked sends see no ACKs for the stall window', async () => {
|
||||
it('self-heals lost ACKs when a later cumulative ACK arrives', async () => {
|
||||
vi.useFakeTimers()
|
||||
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
const mockProc = createMockProc()
|
||||
spawnMock.mockReturnValue(mockProc.proc)
|
||||
|
||||
try {
|
||||
await spawnAndSaturateRendererDeliveryGate(mockProc)
|
||||
expect(mainWindow.webContents.send).toHaveBeenCalledTimes(32)
|
||||
expect(countAckStallWarnings(warnSpy)).toBe(0)
|
||||
|
||||
vi.advanceTimersByTime(9_999)
|
||||
expect(countAckStallWarnings(warnSpy)).toBe(0)
|
||||
expect(mainWindow.webContents.send).toHaveBeenCalledTimes(32)
|
||||
|
||||
vi.advanceTimersByTime(1)
|
||||
expect(countAckStallWarnings(warnSpy)).toBe(1)
|
||||
expect(warnSpy).toHaveBeenCalledWith(
|
||||
ACK_STALL_WATCHDOG_WARNING,
|
||||
expect.objectContaining({
|
||||
rendererInFlightChars: 512 * 1024,
|
||||
rendererInFlightPtyCount: 1,
|
||||
pendingPtyCount: 1
|
||||
})
|
||||
)
|
||||
// Why: the reset must reopen the gate so the held pendingData delivers
|
||||
// without any renderer ACK ever arriving.
|
||||
vi.runOnlyPendingTimers()
|
||||
expect(mainWindow.webContents.send).toHaveBeenCalledTimes(33)
|
||||
|
||||
for (let index = 0; index < 5; index++) {
|
||||
vi.advanceTimersByTime(1)
|
||||
}
|
||||
const spawnResult = await spawnAndSaturateRendererDeliveryGate(mockProc)
|
||||
expect(getPtyDataSendCalls()).toHaveLength(32)
|
||||
expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
|
||||
pendingPtyCount: 0,
|
||||
pendingChars: 0,
|
||||
rendererInFlightChars: 88 * 1024
|
||||
rendererInFlightChars: 512 * 1024,
|
||||
rendererInFlightPtyCount: 1
|
||||
})
|
||||
expect(countAckStallWarnings(warnSpy)).toBe(1)
|
||||
|
||||
// Every per-chunk ACK was lost, but the next ACK carries the renderer's
|
||||
// full cumulative total — the debt clears without any timer or reset.
|
||||
const ackData = getPtyAckDataListener()
|
||||
ackData(null, { id: spawnResult.id, processedChars: 512 * 1024 })
|
||||
|
||||
expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
|
||||
rendererInFlightChars: 0,
|
||||
rendererInFlightPtyCount: 0
|
||||
})
|
||||
|
||||
vi.runOnlyPendingTimers()
|
||||
expect(getPtyDataSendCalls()).toHaveLength(33)
|
||||
} finally {
|
||||
warnSpy.mockRestore()
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('does not fire the ack-stall watchdog while renderer ACKs keep arriving', async () => {
|
||||
it('applies cumulative ACKs idempotently and ignores stale reordered totals', async () => {
|
||||
vi.useFakeTimers()
|
||||
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
const mockProc = createMockProc()
|
||||
spawnMock.mockReturnValue(mockProc.proc)
|
||||
|
||||
@@ -7076,16 +7090,206 @@ describe('registerPtyHandlers', () => {
|
||||
const spawnResult = await spawnAndSaturateRendererDeliveryGate(mockProc)
|
||||
const ackData = getPtyAckDataListener()
|
||||
|
||||
// Healthy heavy load: delivery stays gate-blocked for far longer than
|
||||
// the stall window, but ACKs keep trickling in and restart it each time.
|
||||
for (let round = 0; round < 3; round++) {
|
||||
vi.advanceTimersByTime(5_000)
|
||||
ackData(null, { id: spawnResult.id, charCount: 16 * 1024 })
|
||||
vi.advanceTimersByTime(2)
|
||||
}
|
||||
vi.advanceTimersByTime(5_000)
|
||||
ackData(null, { id: spawnResult.id, processedChars: 256 * 1024 })
|
||||
expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
|
||||
rendererInFlightChars: 256 * 1024,
|
||||
maxRendererInFlightCharsByPty: 256 * 1024
|
||||
})
|
||||
|
||||
expect(countAckStallWarnings(warnSpy)).toBe(0)
|
||||
// Replayed duplicate credits nothing further.
|
||||
ackData(null, { id: spawnResult.id, processedChars: 256 * 1024 })
|
||||
expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
|
||||
rendererInFlightChars: 256 * 1024
|
||||
})
|
||||
|
||||
// A stale reordered total can never move accounting backwards.
|
||||
ackData(null, { id: spawnResult.id, processedChars: 128 * 1024 })
|
||||
expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
|
||||
rendererInFlightChars: 256 * 1024
|
||||
})
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('tolerates mixed legacy delta and cumulative ACK payloads', async () => {
|
||||
vi.useFakeTimers()
|
||||
const mockProc = createMockProc()
|
||||
spawnMock.mockReturnValue(mockProc.proc)
|
||||
|
||||
try {
|
||||
const spawnResult = await spawnAndSaturateRendererDeliveryGate(mockProc)
|
||||
const ackData = getPtyAckDataListener()
|
||||
|
||||
// Legacy delta shape (no processedChars) still credits per chunk.
|
||||
ackData(null, { id: spawnResult.id, charCount: 16 * 1024 })
|
||||
expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
|
||||
rendererInFlightChars: 496 * 1024
|
||||
})
|
||||
|
||||
// A cumulative total then supersedes without double-crediting the delta.
|
||||
ackData(null, { id: spawnResult.id, processedChars: 512 * 1024 })
|
||||
expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
|
||||
rendererInFlightChars: 0,
|
||||
rendererInFlightPtyCount: 0
|
||||
})
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('forwards only newly acknowledged cumulative bytes to provider ACK backpressure', async () => {
|
||||
vi.useFakeTimers()
|
||||
const acknowledgeDataEvent = vi.fn()
|
||||
const mockProc = createMockProc()
|
||||
spawnMock.mockReturnValue(mockProc.proc)
|
||||
|
||||
try {
|
||||
setLocalPtyProvider({
|
||||
spawn: vi.fn(async () => ({ id: 'cumulative-pty' })),
|
||||
write: vi.fn(),
|
||||
resize: vi.fn(),
|
||||
shutdown: vi.fn(),
|
||||
sendSignal: vi.fn(),
|
||||
getCwd: vi.fn(),
|
||||
getInitialCwd: vi.fn(),
|
||||
clearBuffer: vi.fn(),
|
||||
acknowledgeDataEvent,
|
||||
hasChildProcesses: vi.fn(),
|
||||
getForegroundProcess: vi.fn(),
|
||||
serialize: vi.fn(),
|
||||
revive: vi.fn(),
|
||||
onData: vi.fn((callback) => {
|
||||
mockProc.proc.onData((data: string) => callback({ id: 'cumulative-pty', data }))
|
||||
return () => {}
|
||||
}),
|
||||
onReplay: vi.fn(() => () => {}),
|
||||
onExit: vi.fn(() => () => {}),
|
||||
listProcesses: vi.fn(async () => []),
|
||||
attach: vi.fn(),
|
||||
getDefaultShell: vi.fn(),
|
||||
getProfiles: vi.fn()
|
||||
} as never)
|
||||
registerPtyHandlers(mainWindow as never)
|
||||
const ackData = getPtyAckDataListener()
|
||||
mainWindow.webContents.send.mockClear()
|
||||
|
||||
mockProc.emitData('remote-output')
|
||||
vi.advanceTimersByTime(8)
|
||||
|
||||
// Why: cumulative totals are clamped to what main actually sent, and a
|
||||
// replayed total must credit SSH/relay flow control with zero, not
|
||||
// duplicate bytes.
|
||||
ackData(null, { id: 'cumulative-pty', processedChars: 1024 })
|
||||
ackData(null, { id: 'cumulative-pty', processedChars: 1024 })
|
||||
|
||||
expect(acknowledgeDataEvent).toHaveBeenNthCalledWith(
|
||||
1,
|
||||
'cumulative-pty',
|
||||
'remote-output'.length
|
||||
)
|
||||
expect(acknowledgeDataEvent).toHaveBeenNthCalledWith(2, 'cumulative-pty', 0)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('probes for a delivery resync when data arrives for a fully gated PTY and reconciles on reply', async () => {
|
||||
vi.useFakeTimers()
|
||||
const mockProc = createMockProc()
|
||||
spawnMock.mockReturnValue(mockProc.proc)
|
||||
|
||||
try {
|
||||
const spawnResult = await spawnAndSaturateRendererDeliveryGate(mockProc)
|
||||
|
||||
mockProc.emitData('stuck-output')
|
||||
expect(getDeliveryResyncProbeCalls()).toHaveLength(1)
|
||||
const probePayload = getDeliveryResyncProbeCalls()[0]![1] as { requestId: number }
|
||||
|
||||
// Only one probe may be outstanding at a time.
|
||||
mockProc.emitData('still-stuck')
|
||||
expect(getDeliveryResyncProbeCalls()).toHaveLength(1)
|
||||
|
||||
const respondDeliveryResync = getDeliveryResyncResponseListener()
|
||||
respondDeliveryResync(null, {
|
||||
requestId: probePayload.requestId,
|
||||
processedCharsByPty: { [spawnResult.id]: 512 * 1024 }
|
||||
})
|
||||
|
||||
expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
|
||||
rendererInFlightChars: 0,
|
||||
rendererInFlightPtyCount: 0
|
||||
})
|
||||
|
||||
// The reconciled gate lets the held pendingData flush again.
|
||||
vi.advanceTimersByTime(8)
|
||||
expect(getPtyDataSendCalls()).toHaveLength(33)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('ignores resync replies with stale request ids', async () => {
|
||||
vi.useFakeTimers()
|
||||
const mockProc = createMockProc()
|
||||
spawnMock.mockReturnValue(mockProc.proc)
|
||||
|
||||
try {
|
||||
const spawnResult = await spawnAndSaturateRendererDeliveryGate(mockProc)
|
||||
mockProc.emitData('stuck-output')
|
||||
const probePayload = getDeliveryResyncProbeCalls()[0]![1] as { requestId: number }
|
||||
|
||||
const respondDeliveryResync = getDeliveryResyncResponseListener()
|
||||
respondDeliveryResync(null, {
|
||||
requestId: probePayload.requestId + 41,
|
||||
processedCharsByPty: { [spawnResult.id]: 512 * 1024 }
|
||||
})
|
||||
|
||||
expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
|
||||
rendererInFlightChars: 512 * 1024
|
||||
})
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('clears an unanswered resync probe, warns once per silent streak, and never mutates counters', async () => {
|
||||
vi.useFakeTimers()
|
||||
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
const mockProc = createMockProc()
|
||||
spawnMock.mockReturnValue(mockProc.proc)
|
||||
|
||||
try {
|
||||
await spawnAndSaturateRendererDeliveryGate(mockProc)
|
||||
|
||||
mockProc.emitData('stuck-output')
|
||||
expect(getDeliveryResyncProbeCalls()).toHaveLength(1)
|
||||
|
||||
vi.advanceTimersByTime(4_999)
|
||||
expect(countResyncUnansweredWarnings(warnSpy)).toBe(0)
|
||||
|
||||
vi.advanceTimersByTime(1)
|
||||
expect(countResyncUnansweredWarnings(warnSpy)).toBe(1)
|
||||
expect(warnSpy).toHaveBeenCalledWith(
|
||||
DELIVERY_RESYNC_UNANSWERED_WARNING,
|
||||
expect.objectContaining({
|
||||
rendererInFlightChars: 512 * 1024,
|
||||
pendingPtyCount: 1
|
||||
})
|
||||
)
|
||||
// No blind reset: counters and pending output are untouched.
|
||||
expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
|
||||
rendererInFlightChars: 512 * 1024,
|
||||
pendingChars: 88 * 1024 + 'stuck-output'.length
|
||||
})
|
||||
expect(getPtyDataSendCalls()).toHaveLength(32)
|
||||
|
||||
// The cleared flag lets the next gated arrival probe again, but a
|
||||
// still-silent renderer does not spam a second warn.
|
||||
mockProc.emitData('still-stuck')
|
||||
expect(getDeliveryResyncProbeCalls()).toHaveLength(2)
|
||||
vi.advanceTimersByTime(5_000)
|
||||
expect(countResyncUnansweredWarnings(warnSpy)).toBe(1)
|
||||
expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
|
||||
rendererInFlightChars: 512 * 1024
|
||||
})
|
||||
@@ -7095,34 +7299,7 @@ describe('registerPtyHandlers', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('disarms the ack-stall watchdog when an ACK arrives and restarts the stall window', async () => {
|
||||
vi.useFakeTimers()
|
||||
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
const mockProc = createMockProc()
|
||||
spawnMock.mockReturnValue(mockProc.proc)
|
||||
|
||||
try {
|
||||
const spawnResult = await spawnAndSaturateRendererDeliveryGate(mockProc)
|
||||
const ackData = getPtyAckDataListener()
|
||||
|
||||
vi.advanceTimersByTime(9_000)
|
||||
ackData(null, { id: spawnResult.id, charCount: 16 * 1024 })
|
||||
|
||||
// Past the original 10s deadline: the ACK disarmed the first timer.
|
||||
vi.advanceTimersByTime(2_000)
|
||||
expect(countAckStallWarnings(warnSpy)).toBe(0)
|
||||
|
||||
// No further ACKs while still gate-blocked: the re-armed watchdog fires
|
||||
// one full window after the next gate-blocked send.
|
||||
vi.advanceTimersByTime(9_000)
|
||||
expect(countAckStallWarnings(warnSpy)).toBe(1)
|
||||
} finally {
|
||||
warnSpy.mockRestore()
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('cleans up the ack-stall watchdog when the window is destroyed', async () => {
|
||||
it('clears resync probe state when the window is destroyed', async () => {
|
||||
vi.useFakeTimers()
|
||||
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
const mockProc = createMockProc()
|
||||
@@ -7142,16 +7319,16 @@ describe('registerPtyHandlers', () => {
|
||||
for (let index = 0; index < 32; index++) {
|
||||
vi.advanceTimersByTime(1)
|
||||
}
|
||||
expect(destroyableWindow.webContents.send).toHaveBeenCalledTimes(32)
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
mockProc.emitData('stuck-output')
|
||||
// Pending flush timer plus the outstanding probe's hygiene timeout.
|
||||
expect(vi.getTimerCount()).toBe(2)
|
||||
|
||||
destroyed = true
|
||||
mockProc.emitData('post-destroy output')
|
||||
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
vi.advanceTimersByTime(60_000)
|
||||
expect(countAckStallWarnings(warnSpy)).toBe(0)
|
||||
expect(destroyableWindow.webContents.send).toHaveBeenCalledTimes(32)
|
||||
expect(countResyncUnansweredWarnings(warnSpy)).toBe(0)
|
||||
} finally {
|
||||
warnSpy.mockRestore()
|
||||
vi.useRealTimers()
|
||||
@@ -7282,9 +7459,7 @@ describe('registerPtyHandlers', () => {
|
||||
vi.advanceTimersByTime(1)
|
||||
}
|
||||
expect(mainWindow.webContents.send).toHaveBeenCalledTimes(512)
|
||||
// Why: the ack-stall watchdog stays armed while bulk delivery is
|
||||
// gate-blocked; no flush timer remains scheduled.
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
|
||||
writeListener(mainWindowIpcEvent, {
|
||||
id: interactiveSpawn.id,
|
||||
@@ -7314,7 +7489,10 @@ describe('registerPtyHandlers', () => {
|
||||
data: 'a'
|
||||
})
|
||||
interactiveProc.emitData(reserveChunk)
|
||||
expect(mainWindow.webContents.send).toHaveBeenCalledTimes(529)
|
||||
// Why: the reserve-exhausted send stays gated, and the fully gated
|
||||
// arrival now also emits one delivery resync probe (not pty:data).
|
||||
expect(getPtyDataSendCalls()).toHaveLength(529)
|
||||
expect(getDeliveryResyncProbeCalls()).toHaveLength(1)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
@@ -7397,11 +7575,16 @@ describe('registerPtyHandlers', () => {
|
||||
setActiveRendererPty(null, { id: spawns[activeIndex]!.id, active: true })
|
||||
vi.advanceTimersByTime(8)
|
||||
|
||||
expect(mainWindow.webContents.send).toHaveBeenCalledTimes(513)
|
||||
expect(mainWindow.webContents.send).toHaveBeenNthCalledWith(513, 'pty:data', {
|
||||
id: spawns[activeIndex]!.id,
|
||||
data: 'active-output'
|
||||
})
|
||||
// Why: the fully gated arrival also emits one delivery resync probe, so
|
||||
// count pty:data sends rather than raw webContents.send calls.
|
||||
expect(getPtyDataSendCalls()).toHaveLength(513)
|
||||
expect(getPtyDataSendCalls()[512]).toEqual([
|
||||
'pty:data',
|
||||
{
|
||||
id: spawns[activeIndex]!.id,
|
||||
data: 'active-output'
|
||||
}
|
||||
])
|
||||
expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
|
||||
activeRendererPtyCount: 1,
|
||||
pendingPtyCount: procs.length - 1,
|
||||
|
||||
+160
-90
@@ -1289,6 +1289,7 @@ export function registerPtyHandlers(
|
||||
ipcMain.removeAllListeners('pty:write')
|
||||
ipcMain.removeAllListeners('pty:ackColdRestore')
|
||||
ipcMain.removeAllListeners('pty:ackData')
|
||||
ipcMain.removeAllListeners('pty:deliveryResyncResponse')
|
||||
ipcMain.removeAllListeners('pty:serializeBuffer:response')
|
||||
|
||||
// Configure the local provider with app-specific hooks.
|
||||
@@ -1374,12 +1375,19 @@ export function registerPtyHandlers(
|
||||
}
|
||||
|
||||
const pendingData = new Map<string, PendingPtyData>()
|
||||
const rendererInFlightCharsByPty = new Map<string, number>()
|
||||
// Why: TCP-style cumulative delivery accounting. Relative in-flight counters
|
||||
// make every lost ACK a permanent debt; monotonic sent/acked totals self-heal
|
||||
// as soon as any later ACK (or resync reply) reports the renderer's full
|
||||
// processed count.
|
||||
type RendererPtyDeliveryAccounting = { sentChars: number; ackedChars: number }
|
||||
const rendererDeliveryAccountingByPty = new Map<string, RendererPtyDeliveryAccounting>()
|
||||
const trustedTerminalHandleEnv = new Set<string>()
|
||||
let flushTimer: ReturnType<typeof setTimeout> | null = null
|
||||
let rendererInFlightTotalChars = 0
|
||||
let ackStallWatchdogTimer: ReturnType<typeof setTimeout> | null = null
|
||||
let ackGateBlockedSinceMs: number | null = null
|
||||
let deliveryResyncRequestSerial = 0
|
||||
let deliveryResyncOutstandingRequestId: number | null = null
|
||||
let deliveryResyncTimer: ReturnType<typeof setTimeout> | null = null
|
||||
let deliveryResyncUnansweredWarnLogged = false
|
||||
let lastAckReceivedAtMs: number | null = null
|
||||
const PTY_BATCH_INTERVAL_MS = 8
|
||||
const PTY_BATCH_DRAIN_CONTINUE_MS = 1
|
||||
@@ -1391,13 +1399,10 @@ export function registerPtyHandlers(
|
||||
// Why: active panes need a bounded lane through old hidden bulk output so a
|
||||
// keystroke redraw can reach the renderer before every background ACK lands.
|
||||
const PTY_RENDERER_ACTIVE_PTY_IN_FLIGHT_RESERVE_CHARS = 512 * 1024
|
||||
// Why: renderer ACKs can be lost across a system suspend, pinning the
|
||||
// in-flight counters at the cap and silently gating every PTY (old and new)
|
||||
// forever. Healthy renderers ACK continuously even under heavy load, so a
|
||||
// full stall window with gate-blocked sends and zero ACKs means the
|
||||
// outstanding ACKs are gone. A false fire only over-delivers transiently —
|
||||
// the very next gating cycle bounds delivery again.
|
||||
const PTY_ACK_STALL_WATCHDOG_MS = 10_000
|
||||
// Why: request/response hygiene only — this timeout never mutates delivery
|
||||
// state. It clears the outstanding-probe flag so a later gated arrival can
|
||||
// probe again, and logs once per silent streak for field diagnosis.
|
||||
const PTY_DELIVERY_RESYNC_TIMEOUT_MS = 5_000
|
||||
// Why: keep the immediate path bounded to keystroke-sized TUI redraws;
|
||||
// large output and non-interactive output must still use the batcher.
|
||||
const INTERACTIVE_OUTPUT_WINDOW_MS = 100
|
||||
@@ -1410,12 +1415,9 @@ export function registerPtyHandlers(
|
||||
let peakMaxRendererInFlightCharsByPty = 0
|
||||
let ackGatedFlushSkipCount = 0
|
||||
|
||||
function getMaxMapValue(values: Iterable<number>): number {
|
||||
let max = 0
|
||||
for (const value of values) {
|
||||
max = Math.max(max, value)
|
||||
}
|
||||
return max
|
||||
function getRendererInFlightCharsForPty(id: string): number {
|
||||
const accounting = rendererDeliveryAccountingByPty.get(id)
|
||||
return accounting ? accounting.sentChars - accounting.ackedChars : 0
|
||||
}
|
||||
|
||||
function readCurrentPtyRendererDeliveryDebugSnapshot(): PtyRendererDeliveryDebugSnapshot {
|
||||
@@ -1426,13 +1428,22 @@ export function registerPtyHandlers(
|
||||
pendingChars += chars
|
||||
maxPendingCharsByPty = Math.max(maxPendingCharsByPty, chars)
|
||||
}
|
||||
let rendererInFlightPtyCount = 0
|
||||
let maxRendererInFlightCharsByPty = 0
|
||||
for (const accounting of rendererDeliveryAccountingByPty.values()) {
|
||||
const inFlight = accounting.sentChars - accounting.ackedChars
|
||||
if (inFlight > 0) {
|
||||
rendererInFlightPtyCount++
|
||||
}
|
||||
maxRendererInFlightCharsByPty = Math.max(maxRendererInFlightCharsByPty, inFlight)
|
||||
}
|
||||
return {
|
||||
pendingPtyCount: pendingData.size,
|
||||
pendingChars,
|
||||
maxPendingCharsByPty,
|
||||
rendererInFlightPtyCount: rendererInFlightCharsByPty.size,
|
||||
rendererInFlightPtyCount,
|
||||
rendererInFlightChars: rendererInFlightTotalChars,
|
||||
maxRendererInFlightCharsByPty: getMaxMapValue(rendererInFlightCharsByPty.values()),
|
||||
maxRendererInFlightCharsByPty,
|
||||
activeRendererPtyCount: activeRendererPtys.size,
|
||||
flushScheduled: flushTimer !== null,
|
||||
peakPendingChars,
|
||||
@@ -1527,68 +1538,86 @@ export function registerPtyHandlers(
|
||||
const ptyLimit =
|
||||
PTY_RENDERER_IN_FLIGHT_HIGH_WATER_CHARS +
|
||||
(options.interactive === true ? PTY_RENDERER_ACTIVE_PTY_IN_FLIGHT_RESERVE_CHARS : 0)
|
||||
return (
|
||||
(rendererInFlightCharsByPty.get(id) ?? 0) < ptyLimit &&
|
||||
rendererInFlightTotalChars < totalLimit
|
||||
return getRendererInFlightCharsForPty(id) < ptyLimit && rendererInFlightTotalChars < totalLimit
|
||||
}
|
||||
|
||||
// Why: max-merge on cumulative totals is idempotent and reorder-tolerant —
|
||||
// a replayed or out-of-order ACK can never double-credit, and a lost ACK
|
||||
// self-heals when any later ACK reports the full processed count. Returns
|
||||
// the newly acknowledged delta so provider (SSH/daemon) backpressure is only
|
||||
// credited for bytes main actually tracked in flight, never negative.
|
||||
function applyCumulativeAck(id: string, processedChars: number): number {
|
||||
const accounting = rendererDeliveryAccountingByPty.get(id)
|
||||
if (!accounting) {
|
||||
return 0
|
||||
}
|
||||
// Clamped to sentChars so a corrupt payload cannot drive in-flight negative.
|
||||
const nextAckedChars = Math.min(
|
||||
accounting.sentChars,
|
||||
Math.max(accounting.ackedChars, processedChars)
|
||||
)
|
||||
const acknowledged = nextAckedChars - accounting.ackedChars
|
||||
accounting.ackedChars = nextAckedChars
|
||||
rendererInFlightTotalChars = Math.max(0, rendererInFlightTotalChars - acknowledged)
|
||||
return acknowledged
|
||||
}
|
||||
|
||||
function disarmAckStallWatchdog(): void {
|
||||
ackGateBlockedSinceMs = null
|
||||
if (ackStallWatchdogTimer) {
|
||||
clearTimeout(ackStallWatchdogTimer)
|
||||
ackStallWatchdogTimer = null
|
||||
function clearDeliveryResyncProbe(): void {
|
||||
deliveryResyncOutstandingRequestId = null
|
||||
if (deliveryResyncTimer) {
|
||||
clearTimeout(deliveryResyncTimer)
|
||||
deliveryResyncTimer = null
|
||||
}
|
||||
}
|
||||
|
||||
// Why: armed lazily on the first gate-blocked send and disarmed by every
|
||||
// renderer ACK, so the watchdog never fires under healthy heavy load and
|
||||
// costs nothing while delivery is unblocked.
|
||||
function noteAckGateBlockedSend(): void {
|
||||
if (ackStallWatchdogTimer) {
|
||||
// Why: event-triggered verified-state recovery. Data arriving for a fully
|
||||
// gated PTY is the deterministic signal that delivery may be stuck on lost
|
||||
// ACKs (e.g. dropped across a system suspend); ask the renderer for its
|
||||
// authoritative processed totals instead of resetting on a wall-clock guess.
|
||||
function requestDeliveryResyncForGatedPty(): void {
|
||||
if (deliveryResyncOutstandingRequestId !== null || mainWindow.isDestroyed()) {
|
||||
return
|
||||
}
|
||||
ackGateBlockedSinceMs = Date.now()
|
||||
ackStallWatchdogTimer = setTimeout(resetDeliveryGateAfterAckStall, PTY_ACK_STALL_WATCHDOG_MS)
|
||||
ackStallWatchdogTimer.unref?.()
|
||||
}
|
||||
|
||||
function resetDeliveryGateAfterAckStall(): void {
|
||||
ackStallWatchdogTimer = null
|
||||
const gateBlockedSinceMs = ackGateBlockedSinceMs
|
||||
ackGateBlockedSinceMs = null
|
||||
if (mainWindow.isDestroyed() || mainWindow.webContents.isDestroyed?.() === true) {
|
||||
return
|
||||
}
|
||||
// Why: reaching here means zero ACKs arrived for the full stall window
|
||||
// (any ACK disarms the timer). Only reset while delivery is still blocked
|
||||
// on unacked bytes, so an idle-but-healthy renderer is never touched.
|
||||
if (rendererInFlightTotalChars === 0 || pendingData.size === 0) {
|
||||
return
|
||||
}
|
||||
console.warn('[pty] ack stall watchdog: resetting in-flight delivery counters', {
|
||||
gateBlockedForMs: gateBlockedSinceMs === null ? null : Date.now() - gateBlockedSinceMs,
|
||||
msSinceLastAck: lastAckReceivedAtMs === null ? null : Date.now() - lastAckReceivedAtMs,
|
||||
rendererInFlightChars: rendererInFlightTotalChars,
|
||||
rendererInFlightPtyCount: rendererInFlightCharsByPty.size,
|
||||
pendingPtyCount: pendingData.size
|
||||
})
|
||||
rendererInFlightCharsByPty.clear()
|
||||
rendererInFlightTotalChars = 0
|
||||
recordPtyRendererDeliveryPressure()
|
||||
schedulePendingDataFlush(0)
|
||||
deliveryResyncRequestSerial += 1
|
||||
const requestId = deliveryResyncRequestSerial
|
||||
deliveryResyncOutstandingRequestId = requestId
|
||||
deliveryResyncTimer = setTimeout(() => {
|
||||
if (deliveryResyncOutstandingRequestId !== requestId) {
|
||||
return
|
||||
}
|
||||
clearDeliveryResyncProbe()
|
||||
// Why: no state mutation on timeout — a renderer that cannot answer has
|
||||
// dead IPC, and only a reload cures that. Log once per silent streak so
|
||||
// field diagnosis is captured without spamming every probe cycle.
|
||||
if (deliveryResyncUnansweredWarnLogged) {
|
||||
return
|
||||
}
|
||||
deliveryResyncUnansweredWarnLogged = true
|
||||
console.warn('[pty] delivery resync probe unanswered — renderer IPC unresponsive', {
|
||||
msSinceLastAck: lastAckReceivedAtMs === null ? null : Date.now() - lastAckReceivedAtMs,
|
||||
...readCurrentPtyRendererDeliveryDebugSnapshot()
|
||||
})
|
||||
}, PTY_DELIVERY_RESYNC_TIMEOUT_MS)
|
||||
deliveryResyncTimer.unref?.()
|
||||
mainWindow.webContents.send('pty:requestDeliveryResync', { requestId })
|
||||
}
|
||||
|
||||
resetRendererInFlightDeliveryGate = () => {
|
||||
disarmAckStallWatchdog()
|
||||
rendererInFlightCharsByPty.clear()
|
||||
clearDeliveryResyncProbe()
|
||||
deliveryResyncUnansweredWarnLogged = false
|
||||
rendererDeliveryAccountingByPty.clear()
|
||||
rendererInFlightTotalChars = 0
|
||||
recordPtyRendererDeliveryPressure()
|
||||
}
|
||||
|
||||
function sendPtyDataToRenderer(id: string, payload: PtyDataPayload): void {
|
||||
const charCount = getPtyPayloadCharCount(payload)
|
||||
rendererInFlightCharsByPty.set(id, (rendererInFlightCharsByPty.get(id) ?? 0) + charCount)
|
||||
const accounting = rendererDeliveryAccountingByPty.get(id)
|
||||
if (accounting) {
|
||||
accounting.sentChars += charCount
|
||||
} else {
|
||||
rendererDeliveryAccountingByPty.set(id, { sentChars: charCount, ackedChars: 0 })
|
||||
}
|
||||
rendererInFlightTotalChars += charCount
|
||||
recordPtyRendererDeliveryPressure()
|
||||
mainWindow.webContents.send('pty:data', payload)
|
||||
@@ -1677,9 +1706,9 @@ export function registerPtyHandlers(
|
||||
function flushPendingData(): void {
|
||||
flushTimer = null
|
||||
if (mainWindow.isDestroyed()) {
|
||||
disarmAckStallWatchdog()
|
||||
clearDeliveryResyncProbe()
|
||||
pendingData.clear()
|
||||
rendererInFlightCharsByPty.clear()
|
||||
rendererDeliveryAccountingByPty.clear()
|
||||
rendererInFlightTotalChars = 0
|
||||
recordPtyRendererDeliveryPressure()
|
||||
return
|
||||
@@ -1690,7 +1719,6 @@ export function registerPtyHandlers(
|
||||
break
|
||||
}
|
||||
if (!canSendPtyDataToRenderer(id, { interactive: activeRendererPtys.has(id) })) {
|
||||
noteAckGateBlockedSend()
|
||||
continue
|
||||
}
|
||||
pendingData.delete(id)
|
||||
@@ -1782,9 +1810,11 @@ export function registerPtyHandlers(
|
||||
interactiveOutputCharsByPty.delete(payload.id)
|
||||
rendererInFlightTotalChars = Math.max(
|
||||
0,
|
||||
rendererInFlightTotalChars - (rendererInFlightCharsByPty.get(payload.id) ?? 0)
|
||||
rendererInFlightTotalChars - getRendererInFlightCharsForPty(payload.id)
|
||||
)
|
||||
rendererInFlightCharsByPty.delete(payload.id)
|
||||
// Why: the renderer also drops its cumulative total on pty:exit, so a
|
||||
// reused id restarts aligned at zero on both sides.
|
||||
rendererDeliveryAccountingByPty.delete(payload.id)
|
||||
recordPtyRendererDeliveryPressure()
|
||||
mainWindow.webContents.send('pty:exit', payload)
|
||||
}
|
||||
@@ -1839,9 +1869,9 @@ export function registerPtyHandlers(
|
||||
clearTimeout(flushTimer)
|
||||
flushTimer = null
|
||||
}
|
||||
disarmAckStallWatchdog()
|
||||
clearDeliveryResyncProbe()
|
||||
pendingData.clear()
|
||||
rendererInFlightCharsByPty.clear()
|
||||
rendererDeliveryAccountingByPty.clear()
|
||||
rendererInFlightTotalChars = 0
|
||||
recordPtyRendererDeliveryPressure()
|
||||
return
|
||||
@@ -1873,7 +1903,7 @@ export function registerPtyHandlers(
|
||||
// terminal output already handed to the renderer. The reserve is
|
||||
// bounded, and the per-PTY cap still prevents an active TUI runaway.
|
||||
if (!canSendPtyDataToRenderer(payload.id, { interactive: true })) {
|
||||
noteAckGateBlockedSend()
|
||||
requestDeliveryResyncForGatedPty()
|
||||
pendingData.set(payload.id, pending)
|
||||
recordPtyRendererDeliveryPressure()
|
||||
return
|
||||
@@ -1894,6 +1924,13 @@ export function registerPtyHandlers(
|
||||
}
|
||||
pendingData.set(payload.id, pending)
|
||||
recordPtyRendererDeliveryPressure()
|
||||
// Why: probe on data arrival, not on flush skips — new output for a
|
||||
// fully gated PTY is the moment stuck delivery becomes observable.
|
||||
if (
|
||||
!canSendPtyDataToRenderer(payload.id, { interactive: activeRendererPtys.has(payload.id) })
|
||||
) {
|
||||
requestDeliveryResyncForGatedPty()
|
||||
}
|
||||
if (!flushTimer) {
|
||||
schedulePendingDataFlush(PTY_BATCH_INTERVAL_MS)
|
||||
}
|
||||
@@ -3509,27 +3546,60 @@ export function registerPtyHandlers(
|
||||
// Why: renderer ACKs bound main→renderer terminal delivery without stopping
|
||||
// PTY ingestion. Agent/status consumers still see every chunk through the
|
||||
// provider/runtime path while background renderer writes wait their turn.
|
||||
ipcMain.on('pty:ackData', (_event, args: { id: string; charCount: number }) => {
|
||||
lastAckReceivedAtMs = Date.now()
|
||||
// Why: any renderer ACK proves the ACK channel is alive; the stall window
|
||||
// restarts from the next gate-blocked send.
|
||||
disarmAckStallWatchdog()
|
||||
const charCount = Number.isFinite(args.charCount) ? Math.max(0, args.charCount) : 0
|
||||
const current = rendererInFlightCharsByPty.get(args.id) ?? 0
|
||||
const acknowledged = Math.min(current, charCount)
|
||||
const next = Math.max(0, current - charCount)
|
||||
rendererInFlightTotalChars = Math.max(0, rendererInFlightTotalChars - acknowledged)
|
||||
if (next === 0) {
|
||||
rendererInFlightCharsByPty.delete(args.id)
|
||||
} else {
|
||||
rendererInFlightCharsByPty.set(args.id, next)
|
||||
ipcMain.on(
|
||||
'pty:ackData',
|
||||
(_event, args: { id: string; charCount?: number; processedChars?: number }) => {
|
||||
lastAckReceivedAtMs = Date.now()
|
||||
// Why: a live ACK channel means a future unanswered probe is a fresh
|
||||
// diagnostic event, not a continuation of the last silent streak.
|
||||
deliveryResyncUnansweredWarnLogged = false
|
||||
let acknowledged = 0
|
||||
if (typeof args.processedChars === 'number' && Number.isFinite(args.processedChars)) {
|
||||
acknowledged = applyCumulativeAck(args.id, Math.max(0, args.processedChars))
|
||||
} else {
|
||||
// Why: tolerate legacy per-chunk delta payloads — dev hot-reload can
|
||||
// pair an old renderer with a new main. Keyed by field presence.
|
||||
const accounting = rendererDeliveryAccountingByPty.get(args.id)
|
||||
const delta = Number.isFinite(args.charCount) ? Math.max(0, args.charCount ?? 0) : 0
|
||||
acknowledged = accounting ? applyCumulativeAck(args.id, accounting.ackedChars + delta) : 0
|
||||
}
|
||||
tryGetProviderForPty(args.id)?.acknowledgeDataEvent(args.id, acknowledged)
|
||||
recordPtyRendererDeliveryPressure()
|
||||
if (pendingData.size > 0 && !flushTimer) {
|
||||
schedulePendingDataFlush(0)
|
||||
}
|
||||
}
|
||||
tryGetProviderForPty(args.id)?.acknowledgeDataEvent(args.id, acknowledged)
|
||||
recordPtyRendererDeliveryPressure()
|
||||
if (pendingData.size > 0 && !flushTimer) {
|
||||
schedulePendingDataFlush(0)
|
||||
)
|
||||
|
||||
ipcMain.on(
|
||||
'pty:deliveryResyncResponse',
|
||||
(_event, args: { requestId: number; processedCharsByPty: Record<string, number> }) => {
|
||||
if (
|
||||
deliveryResyncOutstandingRequestId === null ||
|
||||
args?.requestId !== deliveryResyncOutstandingRequestId
|
||||
) {
|
||||
return
|
||||
}
|
||||
clearDeliveryResyncProbe()
|
||||
deliveryResyncUnansweredWarnLogged = false
|
||||
// Why: max-merge — the renderer's cumulative totals are authoritative
|
||||
// for what it processed; reconciling them drains exactly the in-flight
|
||||
// debt left by lost ACKs, nothing more.
|
||||
for (const [id, processedChars] of Object.entries(args.processedCharsByPty ?? {})) {
|
||||
if (typeof processedChars !== 'number' || !Number.isFinite(processedChars)) {
|
||||
continue
|
||||
}
|
||||
const acknowledged = applyCumulativeAck(id, Math.max(0, processedChars))
|
||||
if (acknowledged > 0) {
|
||||
tryGetProviderForPty(id)?.acknowledgeDataEvent(id, acknowledged)
|
||||
}
|
||||
}
|
||||
recordPtyRendererDeliveryPressure()
|
||||
if (pendingData.size > 0 && !flushTimer) {
|
||||
schedulePendingDataFlush(0)
|
||||
}
|
||||
}
|
||||
})
|
||||
)
|
||||
|
||||
ipcMain.removeAllListeners('pty:setActiveRendererPty')
|
||||
ipcMain.on('pty:setActiveRendererPty', (_event, args: { id: string; active: boolean }) => {
|
||||
|
||||
@@ -1152,7 +1152,12 @@ export type PreloadApi = {
|
||||
signal: (id: string, signal: string) => void
|
||||
kill: (id: string, opts?: { keepHistory?: boolean }) => Promise<void>
|
||||
ackColdRestore: (id: string) => void
|
||||
ackData: (id: string, charCount: number) => void
|
||||
ackData: (id: string, charCount: number, processedChars?: number) => void
|
||||
onDeliveryResyncRequest: (callback: (payload: { requestId: number }) => void) => () => void
|
||||
respondDeliveryResync: (payload: {
|
||||
requestId: number
|
||||
processedCharsByPty: Record<string, number>
|
||||
}) => void
|
||||
setActiveRendererPty: (id: string, active: boolean) => void
|
||||
setRendererPtyVisible: (id: string, visible: boolean) => void
|
||||
hasChildProcesses: (id: string) => Promise<boolean>
|
||||
|
||||
+22
-2
@@ -815,8 +815,28 @@ const api = {
|
||||
ackColdRestore: (id: string): void => {
|
||||
ipcRenderer.send('pty:ackColdRestore', { id })
|
||||
},
|
||||
ackData: (id: string, charCount: number): void => {
|
||||
ipcRenderer.send('pty:ackData', { id, charCount })
|
||||
/** charCount is the legacy per-chunk delta; processedChars is the
|
||||
* cumulative per-pty total (self-healing under lost ACK messages). */
|
||||
ackData: (id: string, charCount: number, processedChars?: number): void => {
|
||||
ipcRenderer.send('pty:ackData', {
|
||||
id,
|
||||
charCount,
|
||||
...(typeof processedChars === 'number' ? { processedChars } : {})
|
||||
})
|
||||
},
|
||||
/** Main asks for the renderer's cumulative processed totals when terminal
|
||||
* delivery looks stuck on lost ACKs. */
|
||||
onDeliveryResyncRequest: (callback: (payload: { requestId: number }) => void): (() => void) => {
|
||||
const listener = (_event: Electron.IpcRendererEvent, payload: { requestId: number }) =>
|
||||
callback(payload)
|
||||
ipcRenderer.on('pty:requestDeliveryResync', listener)
|
||||
return () => ipcRenderer.removeListener('pty:requestDeliveryResync', listener)
|
||||
},
|
||||
respondDeliveryResync: (payload: {
|
||||
requestId: number
|
||||
processedCharsByPty: Record<string, number>
|
||||
}): void => {
|
||||
ipcRenderer.send('pty:deliveryResyncResponse', payload)
|
||||
},
|
||||
setActiveRendererPty: (id: string, active: boolean): void => {
|
||||
ipcRenderer.send('pty:setActiveRendererPty', { id, active })
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
// Why: pins the renderer half of the delivery-resync protocol — the singleton
|
||||
// dispatcher must answer main's probe with the cumulative processed totals
|
||||
// that back its ACKs, and must drop a PTY's total once it exits.
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
describe('pty dispatcher delivery resync', () => {
|
||||
const originalWindow = (globalThis as { window?: typeof window }).window
|
||||
|
||||
let dataCallback:
|
||||
| ((payload: { id: string; data: string; rawLength?: number; background?: boolean }) => void)
|
||||
| null = null
|
||||
let exitCallback: ((payload: { id: string; code: number }) => void) | null = null
|
||||
let resyncRequestCallback: ((payload: { requestId: number }) => void) | null = null
|
||||
const ackDataMock = vi.fn()
|
||||
const respondDeliveryResyncMock = vi.fn()
|
||||
|
||||
beforeEach(() => {
|
||||
vi.resetModules()
|
||||
dataCallback = null
|
||||
exitCallback = null
|
||||
resyncRequestCallback = null
|
||||
ackDataMock.mockClear()
|
||||
respondDeliveryResyncMock.mockClear()
|
||||
;(globalThis as { window: typeof window }).window = {
|
||||
...originalWindow,
|
||||
api: {
|
||||
...originalWindow?.api,
|
||||
pty: {
|
||||
...originalWindow?.api?.pty,
|
||||
ackData: ackDataMock,
|
||||
onData: vi.fn(
|
||||
(
|
||||
cb: (payload: {
|
||||
id: string
|
||||
data: string
|
||||
rawLength?: number
|
||||
background?: boolean
|
||||
}) => void
|
||||
) => {
|
||||
dataCallback ??= cb
|
||||
return () => {}
|
||||
}
|
||||
),
|
||||
onReplay: vi.fn(() => () => {}),
|
||||
onExit: vi.fn((cb: (payload: { id: string; code: number }) => void) => {
|
||||
exitCallback ??= cb
|
||||
return () => {}
|
||||
}),
|
||||
onDeliveryResyncRequest: vi.fn((cb: (payload: { requestId: number }) => void) => {
|
||||
resyncRequestCallback ??= cb
|
||||
return () => {}
|
||||
}),
|
||||
respondDeliveryResync: respondDeliveryResyncMock
|
||||
}
|
||||
}
|
||||
} as unknown as typeof window
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
if (originalWindow) {
|
||||
;(globalThis as { window: typeof window }).window = originalWindow
|
||||
} else {
|
||||
delete (globalThis as { window?: typeof window }).window
|
||||
}
|
||||
})
|
||||
|
||||
it('answers delivery resync probes with cumulative totals and drops exited PTYs', async () => {
|
||||
const { ensurePtyDispatcher } = await import('./pty-dispatcher')
|
||||
ensurePtyDispatcher()
|
||||
|
||||
dataCallback?.({ id: 'pty-1', data: 'hello' })
|
||||
dataCallback?.({ id: 'pty-1', data: 'world!!', rawLength: 7 })
|
||||
dataCallback?.({ id: 'pty-2', data: 'abc' })
|
||||
|
||||
expect(ackDataMock).toHaveBeenNthCalledWith(1, 'pty-1', 5, 5)
|
||||
expect(ackDataMock).toHaveBeenNthCalledWith(2, 'pty-1', 7, 12)
|
||||
expect(ackDataMock).toHaveBeenNthCalledWith(3, 'pty-2', 3, 3)
|
||||
|
||||
resyncRequestCallback?.({ requestId: 7 })
|
||||
expect(respondDeliveryResyncMock).toHaveBeenCalledWith({
|
||||
requestId: 7,
|
||||
processedCharsByPty: { 'pty-1': 12, 'pty-2': 3 }
|
||||
})
|
||||
|
||||
exitCallback?.({ id: 'pty-1', code: 0 })
|
||||
resyncRequestCallback?.({ requestId: 8 })
|
||||
expect(respondDeliveryResyncMock).toHaveBeenLastCalledWith({
|
||||
requestId: 8,
|
||||
processedCharsByPty: { 'pty-2': 3 }
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -91,7 +91,7 @@ describe('dispatcher → transport → onTitleChange for Pi spinner', () => {
|
||||
dispatcherCallback?.({ id: 'pty-pi', data: 'chunk', rawLength: 10, background: true } as never)
|
||||
|
||||
expect(handler).toHaveBeenCalledWith('chunk', { rawLength: 10, background: true })
|
||||
expect(window.api.pty.ackData).toHaveBeenCalledWith('pty-pi', 10)
|
||||
expect(window.api.pty.ackData).toHaveBeenCalledWith('pty-pi', 10, 10)
|
||||
ptyDataHandlers.delete('pty-pi')
|
||||
})
|
||||
|
||||
@@ -285,7 +285,7 @@ describe('dispatcher → transport → onTitleChange for Pi spinner', () => {
|
||||
|
||||
dispatcherCallback?.({ id: 'pty-early', data: 'setup starts here\r\n', rawLength: 19 })
|
||||
expect(onData).not.toHaveBeenCalled()
|
||||
expect(window.api.pty.ackData).toHaveBeenCalledWith('pty-early', 19)
|
||||
expect(window.api.pty.ackData).toHaveBeenCalledWith('pty-early', 19, 19)
|
||||
|
||||
resolveSpawn({ id: 'pty-early' })
|
||||
await connectPromise
|
||||
|
||||
@@ -6,7 +6,12 @@
|
||||
* and the eager-buffer reconnection logic share.
|
||||
*/
|
||||
import { TERMINAL_SCROLLBACK_SESSION_BUFFER_BYTE_LIMIT } from '../../../../shared/terminal-scrollback-limits'
|
||||
import { ackPtyData, exposeE2eTerminalPtyAckGate } from './terminal-pty-ack-gate'
|
||||
import {
|
||||
ackPtyData,
|
||||
clearProcessedPtyCharTotal,
|
||||
exposeE2eTerminalPtyAckGate,
|
||||
getProcessedPtyCharTotals
|
||||
} from './terminal-pty-ack-gate'
|
||||
import { clampUtf8Tail, type EagerBufferChunk } from './pty-eager-buffer-clamp'
|
||||
import {
|
||||
bufferPreHandlerPtyData,
|
||||
@@ -151,6 +156,9 @@ export function ensurePtyDispatcher(): void {
|
||||
ptyReplayHandlers.get(payload.id)?.(payload.data)
|
||||
})
|
||||
window.api.pty.onExit((payload) => {
|
||||
// Why: main drops its delivery accounting for this pty on exit; drop the
|
||||
// cumulative total too so a reused id restarts at zero on both sides.
|
||||
clearProcessedPtyCharTotal(payload.id)
|
||||
const handler = ptyExitHandlers.get(payload.id)
|
||||
if (handler) {
|
||||
clearPreHandlerPtyState(payload.id)
|
||||
@@ -167,6 +175,15 @@ export function ensurePtyDispatcher(): void {
|
||||
}
|
||||
}
|
||||
})
|
||||
// Why: main probes when delivery looks stuck on lost ACKs (data arriving
|
||||
// for a fully gated PTY). Replying with the cumulative processed totals
|
||||
// lets main reconcile verified state instead of resetting blindly.
|
||||
window.api.pty.onDeliveryResyncRequest?.((payload) => {
|
||||
window.api.pty.respondDeliveryResync?.({
|
||||
requestId: payload.requestId,
|
||||
processedCharsByPty: getProcessedPtyCharTotals()
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
export function subscribeToPtyExit(ptyId: string, watcher: (code: number) => void): () => void {
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
// @vitest-environment happy-dom
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
vi.mock('@/lib/e2e-config', () => ({ e2eConfig: { exposeStore: false } }))
|
||||
|
||||
describe('terminal-pty-ack-gate cumulative totals', () => {
|
||||
const ackDataMock = vi.fn()
|
||||
|
||||
beforeEach(() => {
|
||||
// Why: the cumulative totals are module state; a fresh module per test
|
||||
// mirrors a fresh renderer page (renderer lifecycle reset).
|
||||
vi.resetModules()
|
||||
ackDataMock.mockClear()
|
||||
;(window as unknown as { api: unknown }).api = { pty: { ackData: ackDataMock } }
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
delete (window as unknown as { api?: unknown }).api
|
||||
})
|
||||
|
||||
async function loadAckGate() {
|
||||
return await import('./terminal-pty-ack-gate')
|
||||
}
|
||||
|
||||
it('sends monotonic cumulative totals alongside per-chunk deltas', async () => {
|
||||
const { ackPtyData, getProcessedPtyCharTotals } = await loadAckGate()
|
||||
|
||||
ackPtyData('pty-a', 5)
|
||||
ackPtyData('pty-a', 7)
|
||||
ackPtyData('pty-b', 3)
|
||||
|
||||
expect(ackDataMock).toHaveBeenNthCalledWith(1, 'pty-a', 5, 5)
|
||||
expect(ackDataMock).toHaveBeenNthCalledWith(2, 'pty-a', 7, 12)
|
||||
expect(ackDataMock).toHaveBeenNthCalledWith(3, 'pty-b', 3, 3)
|
||||
expect(getProcessedPtyCharTotals()).toEqual({ 'pty-a': 12, 'pty-b': 3 })
|
||||
})
|
||||
|
||||
it('clears a PTY total so a reused id restarts from zero on both sides', async () => {
|
||||
const { ackPtyData, clearProcessedPtyCharTotal, getProcessedPtyCharTotals } =
|
||||
await loadAckGate()
|
||||
|
||||
ackPtyData('pty-a', 9)
|
||||
clearProcessedPtyCharTotal('pty-a')
|
||||
|
||||
expect(getProcessedPtyCharTotals()).toEqual({})
|
||||
|
||||
ackPtyData('pty-a', 4)
|
||||
expect(ackDataMock).toHaveBeenLastCalledWith('pty-a', 4, 4)
|
||||
})
|
||||
})
|
||||
@@ -18,13 +18,26 @@ type E2eTerminalPtyAckGateWindow = Window & {
|
||||
|
||||
const e2eTerminalAckGatePtyIds = new Set<string>()
|
||||
const e2eTerminalAckGateHeldChars = new Map<string, number>()
|
||||
// Why: monotonic per-PTY totals of processed chars, mirrored to main as
|
||||
// TCP-style cumulative ACKs so a lost ACK message never becomes permanent
|
||||
// in-flight debt. Cleared on pty:exit so a reused id restarts aligned with
|
||||
// main's fresh accounting; a renderer reload resets it with the page.
|
||||
const processedPtyCharTotals = new Map<string, number>()
|
||||
|
||||
function sendPtyAck(ptyId: string, chars: number): void {
|
||||
const processedChars = (processedPtyCharTotals.get(ptyId) ?? 0) + chars
|
||||
processedPtyCharTotals.set(ptyId, processedChars)
|
||||
// Why: keep the legacy per-chunk delta alongside the cumulative total so an
|
||||
// older main (dev hot-reload mix) still credits deltas.
|
||||
window.api.pty.ackData?.(ptyId, chars, processedChars)
|
||||
}
|
||||
|
||||
function releaseE2eTerminalAckGate(): void {
|
||||
const held = Array.from(e2eTerminalAckGateHeldChars.entries())
|
||||
e2eTerminalAckGatePtyIds.clear()
|
||||
e2eTerminalAckGateHeldChars.clear()
|
||||
for (const [ptyId, chars] of held) {
|
||||
window.api.pty.ackData?.(ptyId, chars)
|
||||
sendPtyAck(ptyId, chars)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -58,9 +71,19 @@ export function exposeE2eTerminalPtyAckGate(): void {
|
||||
}
|
||||
|
||||
export function ackPtyData(ptyId: string, chars: number): void {
|
||||
// Why: held e2e-gate chars stay out of the cumulative total too, so a
|
||||
// delivery-resync probe cannot leak them past the simulated backpressure.
|
||||
if (e2eTerminalAckGatePtyIds.has(ptyId)) {
|
||||
e2eTerminalAckGateHeldChars.set(ptyId, (e2eTerminalAckGateHeldChars.get(ptyId) ?? 0) + chars)
|
||||
return
|
||||
}
|
||||
window.api.pty.ackData?.(ptyId, chars)
|
||||
sendPtyAck(ptyId, chars)
|
||||
}
|
||||
|
||||
export function getProcessedPtyCharTotals(): Record<string, number> {
|
||||
return Object.fromEntries(processedPtyCharTotals)
|
||||
}
|
||||
|
||||
export function clearProcessedPtyCharTotal(ptyId: string): void {
|
||||
processedPtyCharTotals.delete(ptyId)
|
||||
}
|
||||
|
||||
@@ -2544,6 +2544,8 @@ function createPtyApi(): NonNullable<Partial<PreloadApi>['pty']> {
|
||||
kill: () => Promise.resolve(),
|
||||
ackColdRestore: () => {},
|
||||
ackData: () => {},
|
||||
onDeliveryResyncRequest: () => noopUnsubscribe,
|
||||
respondDeliveryResync: () => {},
|
||||
setActiveRendererPty: () => {},
|
||||
setRendererPtyVisible: () => {},
|
||||
hasChildProcesses: () => Promise.resolve(false),
|
||||
|
||||
Reference in New Issue
Block a user