mirror of
https://github.com/stablyai/orca.git
synced 2026-10-24 08:03:58 +00:00
fix(ssh): fence in-flight liveness responses
This commit is contained in:
@@ -15,7 +15,8 @@ export async function attachSshPtyWithLiveEvidence(args: {
|
||||
rememberPtyIncarnation: (relayPtyId: string, incarnationId: unknown) => void
|
||||
livenessState: SshPtyLivenessState
|
||||
}): Promise<void> {
|
||||
let liveEvidence: ReturnType<SshPtyLivenessState['beginLiveEvidence']> | undefined
|
||||
let liveEvidence: ReturnType<SshPtyLivenessState['beginLiveEvidence']> | undefined =
|
||||
args.livenessState.beginLiveEvidence(args.appPtyId)
|
||||
try {
|
||||
await requestSshPtyAttach({
|
||||
mux: args.mux,
|
||||
@@ -23,12 +24,8 @@ export async function attachSshPtyWithLiveEvidence(args: {
|
||||
params: { id: args.relayPtyId },
|
||||
commitSourceActivation: true,
|
||||
installSourceActivation: args.installSourceActivation,
|
||||
rememberPtyIncarnation: args.rememberPtyIncarnation,
|
||||
observeResult: () => {
|
||||
liveEvidence = args.livenessState.beginLiveEvidence(args.appPtyId)
|
||||
}
|
||||
rememberPtyIncarnation: args.rememberPtyIncarnation
|
||||
})
|
||||
liveEvidence ??= args.livenessState.beginLiveEvidence(args.appPtyId)
|
||||
args.livenessState.settleLiveEvidence(args.appPtyId, liveEvidence, true)
|
||||
liveEvidence = undefined
|
||||
} finally {
|
||||
|
||||
@@ -1,11 +1,16 @@
|
||||
export const MAX_SSH_PTY_EXIT_TOMBSTONES = 1000
|
||||
|
||||
export type SshPtyLiveEvidence = { valid: boolean }
|
||||
export type SshPtyLiveEvidenceWindow = {
|
||||
valid: boolean
|
||||
readonly invalidatedPtyIds: Set<string>
|
||||
}
|
||||
|
||||
export class SshPtyLivenessState {
|
||||
readonly livePtyIds = new Set<string>()
|
||||
private readonly exitedPtyIds = new Set<string>()
|
||||
private readonly pendingLiveEvidenceByPtyId = new Map<string, Set<SshPtyLiveEvidence>>()
|
||||
private readonly liveEvidenceWindows = new Set<SshPtyLiveEvidenceWindow>()
|
||||
|
||||
constructor(private readonly toAppPtyId: (id: string) => string) {}
|
||||
|
||||
@@ -16,6 +21,11 @@ export class SshPtyLivenessState {
|
||||
}
|
||||
}
|
||||
this.pendingLiveEvidenceByPtyId.clear()
|
||||
for (const window of this.liveEvidenceWindows) {
|
||||
window.valid = false
|
||||
window.invalidatedPtyIds.clear()
|
||||
}
|
||||
this.liveEvidenceWindows.clear()
|
||||
this.livePtyIds.clear()
|
||||
this.exitedPtyIds.clear()
|
||||
}
|
||||
@@ -30,15 +40,36 @@ export class SshPtyLivenessState {
|
||||
this.livePtyIds.add(appPtyId)
|
||||
}
|
||||
|
||||
beginLiveEvidence(id: string): SshPtyLiveEvidence {
|
||||
beginLiveEvidence(id: string, window?: SshPtyLiveEvidenceWindow): SshPtyLiveEvidence {
|
||||
const appPtyId = this.toAppPtyId(id)
|
||||
const evidence = { valid: true }
|
||||
const evidence = {
|
||||
valid:
|
||||
window === undefined ||
|
||||
(window.valid &&
|
||||
this.liveEvidenceWindows.has(window) &&
|
||||
!window.invalidatedPtyIds.has(appPtyId))
|
||||
}
|
||||
if (!evidence.valid) {
|
||||
return evidence
|
||||
}
|
||||
const pending = this.pendingLiveEvidenceByPtyId.get(appPtyId) ?? new Set()
|
||||
pending.add(evidence)
|
||||
this.pendingLiveEvidenceByPtyId.set(appPtyId, pending)
|
||||
return evidence
|
||||
}
|
||||
|
||||
beginLiveEvidenceWindow(): SshPtyLiveEvidenceWindow {
|
||||
const window = { valid: true, invalidatedPtyIds: new Set<string>() }
|
||||
this.liveEvidenceWindows.add(window)
|
||||
return window
|
||||
}
|
||||
|
||||
closeLiveEvidenceWindow(window: SshPtyLiveEvidenceWindow): void {
|
||||
window.valid = false
|
||||
window.invalidatedPtyIds.clear()
|
||||
this.liveEvidenceWindows.delete(window)
|
||||
}
|
||||
|
||||
settleLiveEvidence(id: string, evidence: SshPtyLiveEvidence, acceptLive: boolean): void {
|
||||
const appPtyId = this.toAppPtyId(id)
|
||||
const pending = this.pendingLiveEvidenceByPtyId.get(appPtyId)
|
||||
@@ -74,6 +105,9 @@ export class SshPtyLivenessState {
|
||||
}
|
||||
|
||||
private invalidateLiveEvidence(appPtyId: string): void {
|
||||
for (const window of this.liveEvidenceWindows) {
|
||||
window.invalidatedPtyIds.add(appPtyId)
|
||||
}
|
||||
const pending = this.pendingLiveEvidenceByPtyId.get(appPtyId)
|
||||
if (!pending) {
|
||||
return
|
||||
|
||||
@@ -12,6 +12,7 @@ export async function listSshPtyProcessesWithLiveEvidence(args: {
|
||||
rememberPtyIncarnation: (relayPtyId: string, incarnationId: unknown) => void
|
||||
}): Promise<PtyProcessInfo[]> {
|
||||
const pendingEvidence = new Map<string, SshPtyLiveEvidence>()
|
||||
const evidenceWindow = args.livenessState.beginLiveEvidenceWindow()
|
||||
const mapProcesses = (value: unknown): PtyProcessInfo[] =>
|
||||
mapSshPtyProcessList(value as PtyProcessInfo[], args.toAppPtyId)
|
||||
try {
|
||||
@@ -22,7 +23,10 @@ export async function listSshPtyProcessesWithLiveEvidence(args: {
|
||||
beforeResolve: (value) => {
|
||||
for (const process of mapProcesses(value)) {
|
||||
if (!pendingEvidence.has(process.id)) {
|
||||
pendingEvidence.set(process.id, args.livenessState.beginLiveEvidence(process.id))
|
||||
pendingEvidence.set(
|
||||
process.id,
|
||||
args.livenessState.beginLiveEvidence(process.id, evidenceWindow)
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -30,7 +34,8 @@ export async function listSshPtyProcessesWithLiveEvidence(args: {
|
||||
const processes = mapProcesses(result)
|
||||
for (const process of processes) {
|
||||
const evidence =
|
||||
pendingEvidence.get(process.id) ?? args.livenessState.beginLiveEvidence(process.id)
|
||||
pendingEvidence.get(process.id) ??
|
||||
args.livenessState.beginLiveEvidence(process.id, evidenceWindow)
|
||||
args.livenessState.settleLiveEvidence(process.id, evidence, true)
|
||||
pendingEvidence.delete(process.id)
|
||||
args.rememberPtyIncarnation(args.toRelayPtyId(process.id), process.incarnationId)
|
||||
@@ -40,5 +45,6 @@ export async function listSshPtyProcessesWithLiveEvidence(args: {
|
||||
for (const [id, evidence] of pendingEvidence) {
|
||||
args.livenessState.settleLiveEvidence(id, evidence, false)
|
||||
}
|
||||
args.livenessState.closeLiveEvidenceWindow(evidenceWindow)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -214,6 +214,57 @@ describe('SSH PTY reattach when the relay requires source restoration', () => {
|
||||
await expect(provider.probePtyLiveness(id)).resolves.toBeNull()
|
||||
})
|
||||
|
||||
it('does not promote an attach response after an in-flight exact exit', async () => {
|
||||
const id = 'ssh:conn-1@@pty-attach-exited'
|
||||
const mux = createMockMux()
|
||||
const provider = new SshPtyProvider('conn-1', mux as never)
|
||||
provider.onExit((payload) => provider.acceptExitedPty(payload.id))
|
||||
mux.request.mockImplementation(async (method: string, _params, options) => {
|
||||
if (method !== 'pty.attach') {
|
||||
return undefined
|
||||
}
|
||||
notificationHandler(mux)('pty.exit', {
|
||||
id: 'pty-attach-exited',
|
||||
code: 0,
|
||||
incarnationId: 'incarnation-attach-exited'
|
||||
})
|
||||
const result = { incarnationId: 'incarnation-attach-exited' }
|
||||
options?.beforeResolve?.(result)
|
||||
return result
|
||||
})
|
||||
|
||||
await provider.attach(id)
|
||||
|
||||
await expect(provider.probePtyLiveness(id)).resolves.toBe(false)
|
||||
})
|
||||
|
||||
it('does not promote an inventory response after an in-flight legacy exit', async () => {
|
||||
const id = 'ssh:conn-1@@pty-inventory-unverifiable'
|
||||
const mux = createMockMux()
|
||||
const provider = new SshPtyProvider('conn-1', mux as never)
|
||||
provider.onExit((payload) => provider.acceptUnverifiablePty(payload.id))
|
||||
mux.request.mockImplementation(async (method: string, _params, options) => {
|
||||
if (method !== 'pty.listProcesses') {
|
||||
return undefined
|
||||
}
|
||||
notificationHandler(mux)('pty.exit', { id: 'pty-inventory-unverifiable', code: 0 })
|
||||
const result = [
|
||||
{
|
||||
id: 'pty-inventory-unverifiable',
|
||||
cwd: '/work',
|
||||
title: 'shell',
|
||||
incarnationId: 'incarnation-inventory-unverifiable'
|
||||
}
|
||||
]
|
||||
options?.beforeResolve?.(result)
|
||||
return result
|
||||
})
|
||||
|
||||
await provider.listProcesses()
|
||||
|
||||
await expect(provider.probePtyLiveness(id)).resolves.toBeNull()
|
||||
})
|
||||
|
||||
it('re-attaches over the live PTY instead of reporting exited', async () => {
|
||||
const mux = createMockMux()
|
||||
// Why: the relay retires the stale delivery record as the restoreRequired response settles,
|
||||
|
||||
@@ -88,7 +88,6 @@ export async function requestSshPtyAttach(args: {
|
||||
activation: PtySourceReceivingActivation
|
||||
) => SshPtyReceivingActivationLease
|
||||
rememberPtyIncarnation?: (relayPtyId: string, incarnationId: unknown) => void
|
||||
observeResult?: (result: SshPtyAttachResult) => void
|
||||
}): Promise<SshPtyAttachResult> {
|
||||
let activationLease: SshPtyReceivingActivationLease | undefined
|
||||
const installFromResult = (result: SshPtyAttachResult): void => {
|
||||
@@ -102,7 +101,6 @@ export async function requestSshPtyAttach(args: {
|
||||
beforeResolve: (value) => {
|
||||
const result = parseSshPtyAttachResult(value)
|
||||
installFromResult(result)
|
||||
args.observeResult?.(result)
|
||||
}
|
||||
})
|
||||
const result = parseSshPtyAttachResult(rawResult)
|
||||
|
||||
Reference in New Issue
Block a user