mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 08:03:39 +00:00
fix(ssh): deliver a host exit that a failed reattach quarantined
A pty.exit that arrived while pty.attach was in flight was held back for the reattach to classify. If that attach then rejected — a timeout on a healthy connection, say — the operation was discarded without ever classifying, and the held exit went with it: no pty:exit reached the renderer, the lease never went 'terminated', and the pane kept showing a dead process as live. Release a held exit when the last operation that could still claim it finishes without a verdict. An operation that did classify keeps ownership, so a reattach that fenced a stale incarnation is not overridden by a sibling's failure.
This commit is contained in:
@@ -288,3 +288,116 @@ it('returns a provisional source activation lease to reconnect authority', async
|
||||
deliveryToken: 'token-reconnect'
|
||||
})
|
||||
})
|
||||
|
||||
function createDeferredAttachMux(): {
|
||||
mux: {
|
||||
request: ReturnType<typeof vi.fn>
|
||||
notify: ReturnType<typeof vi.fn>
|
||||
onNotification: ReturnType<typeof vi.fn>
|
||||
dispose: ReturnType<typeof vi.fn>
|
||||
isDisposed: ReturnType<typeof vi.fn>
|
||||
}
|
||||
attaches: { resolve: (value: unknown) => void; reject: (error: Error) => void }[]
|
||||
} {
|
||||
const attaches: { resolve: (value: unknown) => void; reject: (error: Error) => void }[] = []
|
||||
const mux = {
|
||||
request: vi.fn(),
|
||||
notify: vi.fn(),
|
||||
onNotification: vi.fn(),
|
||||
dispose: vi.fn(),
|
||||
isDisposed: vi.fn().mockReturnValue(false)
|
||||
}
|
||||
mux.request.mockImplementation((method: string, _params, options) => {
|
||||
if (method !== 'pty.attach') {
|
||||
return Promise.resolve(undefined)
|
||||
}
|
||||
return new Promise((resolve, reject) => {
|
||||
attaches.push({
|
||||
resolve: (value) => {
|
||||
options?.beforeResolve?.(value)
|
||||
resolve(value)
|
||||
},
|
||||
reject
|
||||
})
|
||||
})
|
||||
})
|
||||
return { mux, attaches }
|
||||
}
|
||||
|
||||
function notifyExit(
|
||||
mux: { onNotification: ReturnType<typeof vi.fn> },
|
||||
params: Record<string, unknown>
|
||||
): void {
|
||||
const notify = mux.onNotification.mock.calls[0]?.[0] as
|
||||
| ((method: string, params: Record<string, unknown>) => void)
|
||||
| undefined
|
||||
notify?.('pty.exit', params)
|
||||
}
|
||||
|
||||
it('delivers a host exit that a failed reattach quarantined', async () => {
|
||||
const { mux, attaches } = createDeferredAttachMux()
|
||||
const provider = new SshPtyProvider('conn-1', mux as never)
|
||||
const exitListener = vi.fn()
|
||||
provider.onExit(exitListener)
|
||||
|
||||
const spawn = provider.spawn({ cols: 80, rows: 24, sessionId: 'pty-old' })
|
||||
await vi.waitFor(() => expect(attaches).toHaveLength(1))
|
||||
notifyExit(mux, { id: 'pty-old', code: 7, incarnationId: 'incarnation-old' })
|
||||
expect(exitListener).not.toHaveBeenCalled()
|
||||
attaches[0]?.reject(new Error('Request timed out'))
|
||||
|
||||
await expect(spawn).rejects.toThrow(SSH_PTY_LIVENESS_UNVERIFIABLE_ERROR)
|
||||
expect(exitListener).toHaveBeenCalledExactlyOnceWith({
|
||||
id: 'ssh:conn-1@@pty-old',
|
||||
code: 7,
|
||||
incarnationId: 'incarnation-old',
|
||||
providerGeneration: expect.any(Number),
|
||||
ptyIncarnation: 'incarnation-old'
|
||||
})
|
||||
})
|
||||
|
||||
it('releases a quarantined exit once, after every concurrent reattach has failed', async () => {
|
||||
const { mux, attaches } = createDeferredAttachMux()
|
||||
const provider = new SshPtyProvider('conn-1', mux as never)
|
||||
const exitListener = vi.fn()
|
||||
provider.onExit(exitListener)
|
||||
|
||||
const first = provider.spawn({ cols: 80, rows: 24, sessionId: 'pty-old' })
|
||||
const second = provider.spawn({ cols: 80, rows: 24, sessionId: 'pty-old' })
|
||||
await vi.waitFor(() => expect(attaches).toHaveLength(2))
|
||||
notifyExit(mux, { id: 'pty-old', code: 7, incarnationId: 'incarnation-old' })
|
||||
|
||||
attaches[0]?.reject(new Error('Request timed out'))
|
||||
await expect(first).rejects.toThrow(SSH_PTY_LIVENESS_UNVERIFIABLE_ERROR)
|
||||
expect(exitListener).not.toHaveBeenCalled()
|
||||
|
||||
attaches[1]?.reject(new Error('Request timed out'))
|
||||
await expect(second).rejects.toThrow(SSH_PTY_LIVENESS_UNVERIFIABLE_ERROR)
|
||||
expect(exitListener).toHaveBeenCalledOnce()
|
||||
expect(exitListener).toHaveBeenCalledWith(expect.objectContaining({ code: 7 }))
|
||||
})
|
||||
|
||||
it('keeps a concurrent reattach the owner of a quarantined exit it fenced', async () => {
|
||||
const { mux, attaches } = createDeferredAttachMux()
|
||||
const provider = new SshPtyProvider('conn-1', mux as never)
|
||||
const exitListener = vi.fn()
|
||||
provider.onExit(exitListener)
|
||||
|
||||
const first = provider.spawn({ cols: 80, rows: 24, sessionId: 'pty-old' })
|
||||
const second = provider.spawn({ cols: 80, rows: 24, sessionId: 'pty-old' })
|
||||
await vi.waitFor(() => expect(attaches).toHaveLength(2))
|
||||
notifyExit(mux, { id: 'pty-old', code: 7, incarnationId: 'incarnation-old' })
|
||||
|
||||
attaches[0]?.reject(new Error('Request timed out'))
|
||||
await expect(first).rejects.toThrow(SSH_PTY_LIVENESS_UNVERIFIABLE_ERROR)
|
||||
expect(exitListener).not.toHaveBeenCalled()
|
||||
|
||||
attaches[1]?.resolve({ incarnationId: 'incarnation-new' })
|
||||
await expect(second).resolves.toMatchObject({
|
||||
id: 'ssh:conn-1@@pty-old',
|
||||
incarnationId: 'incarnation-new',
|
||||
isReattach: true
|
||||
})
|
||||
expect(exitListener).not.toHaveBeenCalled()
|
||||
await expect(provider.probePtyLiveness('ssh:conn-1@@pty-old')).resolves.toBe(true)
|
||||
})
|
||||
|
||||
@@ -1,11 +1,17 @@
|
||||
import { isPtyIncarnationId, type PtyIncarnationId } from '../../shared/pty-incarnation'
|
||||
|
||||
type QuarantinedSshPtyExit = {
|
||||
holders: number
|
||||
classified: boolean
|
||||
publish: () => void
|
||||
}
|
||||
|
||||
type PendingSshPtySpawn = {
|
||||
relayPtyId?: string
|
||||
exits: {
|
||||
relayPtyId: string
|
||||
incarnationId?: PtyIncarnationId
|
||||
publish?: () => void
|
||||
quarantine?: QuarantinedSshPtyExit
|
||||
}[]
|
||||
}
|
||||
|
||||
@@ -25,24 +31,29 @@ export class SshPtySpawnExitRaceTracker {
|
||||
}
|
||||
|
||||
recordExit(relayPtyId: string, incarnationId: unknown, publish?: () => void): boolean {
|
||||
let quarantined = false
|
||||
let published = false
|
||||
const publishOnce = (): void => {
|
||||
if (!published) {
|
||||
published = true
|
||||
publish?.()
|
||||
const quarantine: QuarantinedSshPtyExit = {
|
||||
holders: 0,
|
||||
classified: false,
|
||||
publish: () => {
|
||||
if (!published) {
|
||||
published = true
|
||||
publish?.()
|
||||
}
|
||||
}
|
||||
}
|
||||
for (const operation of this.pending) {
|
||||
const quarantine = operation.relayPtyId === relayPtyId && publish !== undefined
|
||||
const held = operation.relayPtyId === relayPtyId && publish !== undefined
|
||||
if (held) {
|
||||
quarantine.holders++
|
||||
}
|
||||
operation.exits.push({
|
||||
relayPtyId,
|
||||
...(isPtyIncarnationId(incarnationId) ? { incarnationId } : {}),
|
||||
...(quarantine ? { publish: publishOnce } : {})
|
||||
...(held ? { quarantine } : {})
|
||||
})
|
||||
quarantined ||= quarantine
|
||||
}
|
||||
return quarantined
|
||||
return quarantine.holders > 0
|
||||
}
|
||||
|
||||
classifyPendingExit(
|
||||
@@ -50,12 +61,18 @@ export class SshPtySpawnExitRaceTracker {
|
||||
result: { id: string; incarnationId?: PtyIncarnationId }
|
||||
): SshPtyPendingExitOutcome {
|
||||
const sameIdExits = operation.exits.filter((exit) => exit.relayPtyId === result.id)
|
||||
for (const exit of sameIdExits) {
|
||||
// Why: a reached verdict owns the exit, so finish() must not second-guess it and re-publish.
|
||||
if (exit.quarantine) {
|
||||
exit.quarantine.classified = true
|
||||
}
|
||||
}
|
||||
if (!result.incarnationId) {
|
||||
return sameIdExits.length > 0 ? 'unverifiable' : null
|
||||
}
|
||||
const matchingExit = sameIdExits.find((exit) => exit.incarnationId === result.incarnationId)
|
||||
if (matchingExit) {
|
||||
matchingExit.publish?.()
|
||||
matchingExit.quarantine?.publish()
|
||||
return 'exited'
|
||||
}
|
||||
return sameIdExits.some((exit) => !exit.incarnationId) ? 'unverifiable' : null
|
||||
@@ -63,5 +80,17 @@ export class SshPtySpawnExitRaceTracker {
|
||||
|
||||
finish(operation: PendingSshPtySpawn): void {
|
||||
this.pending.delete(operation)
|
||||
for (const exit of operation.exits) {
|
||||
const quarantine = exit.quarantine
|
||||
if (!quarantine) {
|
||||
continue
|
||||
}
|
||||
quarantine.holders--
|
||||
// Why: a failed attach never classifies, and the host's exit is still positive death
|
||||
// evidence — release it once no other operation can still claim it, or it is lost forever.
|
||||
if (quarantine.holders === 0 && !quarantine.classified) {
|
||||
quarantine.publish()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user