diff --git a/packages/cli/src/serve/conversations/conversation-runtime-ownership.test.ts b/packages/cli/src/serve/conversations/conversation-runtime-ownership.test.ts index 27ed79e631e..e6bdd3cb19d 100644 --- a/packages/cli/src/serve/conversations/conversation-runtime-ownership.test.ts +++ b/packages/cli/src/serve/conversations/conversation-runtime-ownership.test.ts @@ -21,6 +21,47 @@ import { } from '../live/discovery.js'; import { LIVE_HOST_PROTOCOL_VERSION } from '../live/types.js'; +const recordReadFailure = vi.hoisted(() => ({ + path: undefined as string | undefined, + operation: undefined as 'open' | 'readFile' | undefined, + code: undefined as string | undefined, + injected: false, +})); + +vi.mock('node:fs/promises', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + open: vi.fn(async (...args: Parameters) => { + const matches = String(args[0]) === recordReadFailure.path; + if ( + matches && + !recordReadFailure.injected && + recordReadFailure.operation === 'open' + ) { + recordReadFailure.injected = true; + throw Object.assign(new Error(recordReadFailure.code), { + code: recordReadFailure.code, + }); + } + const handle = await actual.open(...args); + if ( + matches && + !recordReadFailure.injected && + recordReadFailure.operation === 'readFile' + ) { + recordReadFailure.injected = true; + vi.spyOn(handle, 'readFile').mockRejectedValueOnce( + Object.assign(new Error(recordReadFailure.code), { + code: recordReadFailure.code, + }), + ); + } + return handle; + }), + }; +}); + const temporaryDirectories: string[] = []; const childProcesses = new Set(); const currentNonce = 'conversation_owner_nonce_current_01'; @@ -39,7 +80,47 @@ async function readRecord(stableBaseDir: string) { ) as { version: number; pid: number; instanceNonce: string }; } +function failRecordReadOnce( + recordPath: string, + operation: 'open' | 'readFile', + code: string, +): void { + recordReadFailure.path = recordPath; + recordReadFailure.operation = operation; + recordReadFailure.code = code; + recordReadFailure.injected = false; +} + +async function writeForeignRecord( + stableBaseDir: string, + kind: 'ownership' | 'Live discovery', +): Promise { + if (kind === 'ownership') { + const previous = createConversationRuntimeOwnership({ + stableBaseDir, + pid: 999_998, + instanceNonce: 'conversation_owner_nonce_read_failure', + isProcessAlive: () => false, + }); + await previous.acquire(); + return getConversationRuntimeOwnerPath(stableBaseDir); + } + await fs.mkdir(stableBaseDir, { recursive: true, mode: 0o700 }); + await writeLiveDiscoveryFile(stableBaseDir, { + url: 'http://127.0.0.1:3210', + protocolVersion: LIVE_HOST_PROTOCOL_VERSION, + pid: 999_997, + instanceNonce: 'legacy_live_owner_nonce_read_failure', + }); + return getLiveDiscoveryPath(stableBaseDir); +} + afterEach(async () => { + vi.restoreAllMocks(); + recordReadFailure.path = undefined; + recordReadFailure.operation = undefined; + recordReadFailure.code = undefined; + recordReadFailure.injected = false; for (const child of childProcesses) { child.kill('SIGKILL'); } @@ -313,6 +394,58 @@ try { expect(wait).toHaveBeenCalledOnce(); }); + it.each([ + ['ownership', 'open'], + ['ownership', 'readFile'], + ['Live discovery', 'open'], + ['Live discovery', 'readFile'], + ] as const)( + 'recovers after a transient %s record %s failure', + async (kind, operation) => { + const stableBaseDir = await temporaryStableBase(); + const recordPath = await writeForeignRecord(stableBaseDir, kind); + const ownership = createConversationRuntimeOwnership({ + stableBaseDir, + pid: process.pid, + instanceNonce: currentNonce, + isProcessAlive: () => false, + handoffGraceMs: 0, + }); + + failRecordReadOnce( + recordPath, + operation, + operation === 'open' ? 'EMFILE' : 'EIO', + ); + await expect(ownership.acquire()).rejects.toMatchObject({ + code: 'conversation_runtime_unavailable', + retryable: true, + }); + + await expect(ownership.acquire()).resolves.toEqual({ reclaimed: true }); + }, + ); + + it.each(['ownership', 'Live discovery'] as const)( + 'treats an ELOOP opening the %s record as terminal compromise', + async (kind) => { + const stableBaseDir = await temporaryStableBase(); + const recordPath = await writeForeignRecord(stableBaseDir, kind); + const ownership = createConversationRuntimeOwnership({ + stableBaseDir, + pid: process.pid, + instanceNonce: currentNonce, + isProcessAlive: () => false, + }); + + failRecordReadOnce(recordPath, 'open', 'ELOOP'); + await expect(ownership.acquire()).rejects.toMatchObject({ + code: 'conversation_runtime_ownership_compromised', + retryable: false, + }); + }, + ); + it('rejects an active foreign legacy Live owner before writing its record', async () => { const stableBaseDir = await temporaryStableBase(); await fs.mkdir(stableBaseDir, { recursive: true, mode: 0o700 }); diff --git a/packages/cli/src/serve/conversations/conversation-runtime-ownership.ts b/packages/cli/src/serve/conversations/conversation-runtime-ownership.ts index 66b6a3cb091..d0d06d293ba 100644 --- a/packages/cli/src/serve/conversations/conversation-runtime-ownership.ts +++ b/packages/cli/src/serve/conversations/conversation-runtime-ownership.ts @@ -226,7 +226,10 @@ async function readOwnerRecord( : fsConstants.O_RDONLY | fsConstants.O_NOFOLLOW, ); } catch (error) { - throw new UnsafeOwnershipStateError(undefined, { cause: error }); + if ((error as NodeJS.ErrnoException).code === 'ELOOP') { + throw new UnsafeOwnershipStateError(undefined, { cause: error }); + } + throw error; } try { const handleStat = await handle.stat(); @@ -243,9 +246,10 @@ async function readOwnerRecord( ) { throw new UnsafeOwnershipStateError(); } + const serialized = await handle.readFile('utf8'); let parsed: unknown; try { - parsed = JSON.parse(await handle.readFile('utf8')); + parsed = JSON.parse(serialized); } catch (error) { throw new UnsafeOwnershipStateError(undefined, { cause: error }); } diff --git a/packages/cli/src/serve/live/discovery.ts b/packages/cli/src/serve/live/discovery.ts index 08ed0252199..4c4b5dcdb58 100644 --- a/packages/cli/src/serve/live/discovery.ts +++ b/packages/cli/src/serve/live/discovery.ts @@ -188,8 +188,11 @@ async function readExistingRecord( ? fsConstants.O_RDONLY : fsConstants.O_RDONLY | fsConstants.O_NOFOLLOW, ); - } catch { - throw new LiveDiscoveryStateError(); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ELOOP') { + throw new LiveDiscoveryStateError(error); + } + throw error; } try { const handleStat = await handle.stat(); @@ -206,11 +209,12 @@ async function readExistingRecord( ) { throw new LiveDiscoveryStateError(); } + const serialized = await handle.readFile('utf8'); let parsed: unknown; try { - parsed = JSON.parse(await handle.readFile('utf8')); - } catch { - throw new LiveDiscoveryStateError(); + parsed = JSON.parse(serialized); + } catch (error) { + throw new LiveDiscoveryStateError(error); } if ( typeof parsed !== 'object' ||