diff --git a/src/main/daemon/history/journal/terminal-history-journals.ts b/src/main/daemon/history/journal/terminal-history-journals.ts index a80e2a59558..f57b933b681 100644 --- a/src/main/daemon/history/journal/terminal-history-journals.ts +++ b/src/main/daemon/history/journal/terminal-history-journals.ts @@ -1,16 +1,14 @@ import { promises as fsPromises } from 'node:fs' import { join } from 'node:path' +import { errnoCodeOrUnknown } from '../../../../shared/errno-code' import { PrioritySemaphore } from '../../../../shared/priority-semaphore' import { ensurePrivateDirAsync } from '../../endpoint/publication/daemon-private-file-modes' import { DAEMON_HISTORY_RESIZE_MODEL } from '../../protocol/daemon-history-replay-pty' import { TerminalHistoryUnavailableError } from '../../protocol/daemon-history-protocol' import { getHistorySessionDirName } from '../storage/history-paths' import { LOG_HEADER_BYTES } from '../storage/terminal-history-log' -import { mayHaveTerminalHistoryRecoveryProtection } from '../storage/terminal-history-recovery-quarantine' -import { - provesTerminalHistoryPathAbsent, - terminalHistoryPathMayExist -} from '../storage/terminal-history-session-files' +import { hasTerminalHistoryRecoveryProtectionAsync } from '../storage/terminal-history-recovery-quarantine' +import { terminalHistoryPathExists } from '../storage/terminal-history-session-files' import { createTerminalHistoryMetaOwned, newTerminalHistoryMeta, @@ -54,7 +52,9 @@ export class TerminalHistoryJournals { /** * Before any fresh physical spawn: wait for the predecessor's actual I/O, then make the shared * root's metadata hold the ID for prior-stable readers. Throws rather than start unmarked. The - * caller's pending creation owns every step, so its file work is exclusive for this ID. + * caller's pending creation owns every step, so its file work is exclusive for this ID. Only an + * actual marker or saved metadata that cannot be read preserves recovery without a write; a + * lookup that cannot tell refuses this start and changes nothing. */ async prepareHold( sessionId: string, @@ -81,7 +81,7 @@ export class TerminalHistoryJournals { } const dir = join(this.root, getHistorySessionDirName(sessionId)) try { - if (await mayHaveTerminalHistoryRecoveryProtection(this.root, sessionId)) { + if (await hasTerminalHistoryRecoveryProtectionAsync(this.root, sessionId)) { return { sessionId, preservesRecovery: true } } const meta = await readTerminalHistoryMetaInDirAsync(dir) @@ -242,17 +242,20 @@ export class TerminalHistoryJournals { } } -// Only a confirmed missing file is absent; one that cannot be read may still be recovery. +// Only ENOENT is absence; a lookup that cannot tell throws and refuses the start. async function hasReplayableFiles(dir: string): Promise { if ( - (await terminalHistoryPathMayExist(join(dir, 'checkpoint.json'))) || - (await terminalHistoryPathMayExist(join(dir, 'scrollback.bin'))) + (await terminalHistoryPathExists(join(dir, 'checkpoint.json'))) || + (await terminalHistoryPathExists(join(dir, 'scrollback.bin'))) ) { return true } try { return (await fsPromises.stat(join(dir, 'output.log'))).size > LOG_HEADER_BYTES } catch (error) { - return !provesTerminalHistoryPathAbsent(error) + if (errnoCodeOrUnknown(error) === 'ENOENT') { + return false + } + throw error } } diff --git a/src/main/daemon/history/journal/testing/terminal-history-cwd-observation.test.ts b/src/main/daemon/history/journal/testing/terminal-history-cwd-observation.test.ts index a67f4da5a43..19088e9c976 100644 --- a/src/main/daemon/history/journal/testing/terminal-history-cwd-observation.test.ts +++ b/src/main/daemon/history/journal/testing/terminal-history-cwd-observation.test.ts @@ -321,6 +321,8 @@ describe('one admitted operation and its source lineage', () => { await renaming.reached journal.gap('storage_failed') renaming.open() + // The pump still owns the issued rename; its fence returns once that rename has settled. + expect(await journal.flush()).toBe(false) expect(await preflight).toBe(false) expect(await sleep.outcome.promise).toEqual({ phase: 'preflight', reason: 'gap' }) diff --git a/src/main/daemon/history/journal/testing/terminal-history-journals-read-pressure.test.ts b/src/main/daemon/history/journal/testing/terminal-history-journals-read-pressure.test.ts index e969a75b54c..e2f875be238 100644 --- a/src/main/daemon/history/journal/testing/terminal-history-journals-read-pressure.test.ts +++ b/src/main/daemon/history/journal/testing/terminal-history-journals-read-pressure.test.ts @@ -1,33 +1,16 @@ import { afterEach, describe, expect, it, vi } from 'vitest' -import { mkdtempSync, rmSync } from 'node:fs' +import { mkdtempSync, promises as fsPromises, readFileSync, rmSync } from 'node:fs' import { tmpdir } from 'node:os' import { join } from 'node:path' import { TerminalHistoryJournals } from '../terminal-history-journals' import type { TerminalSessionJournal } from '../terminal-session-journal' import type { TerminalHistoryReads } from '../terminal-history-read-admission' -import type * as FileReaderModule from '../../storage/terminal-history-file-reader' import { gateAppends, gateReconstruction } from './terminal-history-read-fixture' -const sourceRead = vi.hoisted(() => ({ during: async (): Promise => {} })) -// A reader's issued source file read can still be in flight when its terminal exits. -vi.mock('../../storage/terminal-history-file-reader', async (importOriginal) => { - const actual = await importOriginal() - return { - ...actual, - readTerminalHistoryBufferAsync: async ( - ...args: Parameters - ) => { - await sourceRead.during() - return actual.readTerminalHistoryBufferAsync(...args) - } - } -}) - const roots: string[] = [] afterEach(() => { vi.restoreAllMocks() - sourceRead.during = async () => {} for (const root of roots.splice(0)) { rmSync(root, { recursive: true, force: true }) } @@ -136,16 +119,25 @@ describe('host-wide read pressure', () => { const old = await open(sessionId) old.record({ kind: 'output', data: 'OLD_ROW\r\n' }) expect(await old.flush()).toBe(true) - const reading = Promise.withResolvers() + // The next file handle to close, the reader's source log, finishes only when released. + const probe = await fsPromises.open(join(old.dir, 'meta.json'), 'r') + const handles = Object.getPrototypeOf(probe) + await probe.close() + const close = handles.close + const closing = Promise.withResolvers() const release = Promise.withResolvers() - sourceRead.during = async () => { - reading.resolve() - await release.promise - } + let held = false + vi.spyOn(handles, 'close').mockImplementation(async function (this: unknown) { + if (!held) { + held = true + closing.resolve() + await release.promise + } + return close.call(this) + }) const read = readThrough(journals.reads, old) - // Sync point only: the read holds its replay lease and its source read is issued. - await reading.promise - sourceRead.during = async () => {} + // Sync point only: the read holds its replay lease and its source file is still closing. + await closing.promise let settled = false void old.settled.then(() => { settled = true @@ -153,12 +145,17 @@ describe('host-wide read pressure', () => { old.finish(0) // The final pump, its clean end included, has finished. await old.flush() + const ended = readFileSync(join(old.dir, 'meta.json'), 'utf8') + const successor = journals.prepareHold(sessionId, { cwd: '/tmp', cols: 80, rows: 24 }) expect(settled).toBe(false) + expect(JSON.parse(ended).endedAt).not.toBeNull() + expect(readFileSync(join(old.dir, 'meta.json'), 'utf8')).toBe(ended) release.resolve() expect(await read).toContain('OLD_ROW') - await old.settled - const hold = await journals.prepareHold(sessionId, { cwd: '/tmp', cols: 80, rows: 24 }) + const hold = await successor + expect(settled).toBe(true) expect(hold.preservesRecovery).toBe(false) + expect(JSON.parse(readFileSync(join(old.dir, 'meta.json'), 'utf8')).endedAt).toBeNull() }, 30_000) }) diff --git a/src/main/daemon/history/storage/terminal-history-metadata.ts b/src/main/daemon/history/storage/terminal-history-metadata.ts index 54bca524b18..6eef782edd9 100644 --- a/src/main/daemon/history/storage/terminal-history-metadata.ts +++ b/src/main/daemon/history/storage/terminal-history-metadata.ts @@ -1,7 +1,6 @@ import { existsSync, promises as fsPromises, readFileSync, writeFileSync } from 'node:fs' import { join } from 'node:path' import { errnoCodeOrUnknown } from '../../../../shared/errno-code' -import { provesTerminalHistoryPathAbsent } from './terminal-history-session-files' import { getHistorySessionDirName } from './history-paths' import { isValidTerminalHistorySize } from '../restoration/terminal-history-dimensions' import { @@ -60,7 +59,10 @@ export function readTerminalHistoryMetaInDir(dir: string): SessionMetaRead { } } -/** Bounded and awaited: only a confirmed missing file is missing; any other failure is unreadable. */ +/** + * Bounded and awaited. Only ENOENT is missing; saved content that cannot be read is unreadable; + * a filesystem lookup that cannot tell either way throws. + */ export async function readTerminalHistoryMetaInDirAsync(dir: string): Promise { let meta: unknown try { @@ -69,7 +71,14 @@ export async function readTerminalHistoryMetaInDirAsync(dir: string): Promise { - return terminalHistoryPathMayExist( + return terminalHistoryPathExists( join(basePath, getHistorySessionDirName(sessionId), RECOVERY_PROTECTION_MARKER) ) } diff --git a/src/main/daemon/history/storage/terminal-history-session-files.ts b/src/main/daemon/history/storage/terminal-history-session-files.ts index 39c7dceaf1b..033b450e83f 100644 --- a/src/main/daemon/history/storage/terminal-history-session-files.ts +++ b/src/main/daemon/history/storage/terminal-history-session-files.ts @@ -54,25 +54,25 @@ export async function readTerminalHistoryLogHeader( } } -/** The kernel's proof a path does not exist; access and read failures prove nothing. */ -export function provesTerminalHistoryPathAbsent(error: unknown): boolean { - const code = errnoCodeOrUnknown(error) - return code === 'ENOENT' || code === 'ENOTDIR' -} - -/** Only a confirmed missing path is absent; one that cannot be read may still be recovery. */ -export async function terminalHistoryPathMayExist(path: string): Promise { +/** + * Whether a path exists. Only ENOENT is absence; a lookup that cannot tell throws, so no caller + * mistakes uncertainty for either a missing file or a confirmed one. + */ +export async function terminalHistoryPathExists(path: string): Promise { try { await fsPromises.stat(path) return true } catch (error) { - return !provesTerminalHistoryPathAbsent(error) + if (errnoCodeOrUnknown(error) === 'ENOENT') { + return false + } + throw error } } export async function hasReplayableTerminalHistorySessionFiles(dir: string): Promise { for (const name of REPLAYABLE_SESSION_FILE_NAMES) { - if (await terminalHistoryPathMayExist(join(dir, name))) { + if (await terminalHistoryPathExists(join(dir, name))) { return true } } diff --git a/src/main/daemon/providers/adapter/testing/daemon-pty-adapter-host-history-sleep.test.ts b/src/main/daemon/providers/adapter/testing/daemon-pty-adapter-host-history-sleep.test.ts index 44b54793eeb..c46e93baa40 100644 --- a/src/main/daemon/providers/adapter/testing/daemon-pty-adapter-host-history-sleep.test.ts +++ b/src/main/daemon/providers/adapter/testing/daemon-pty-adapter-host-history-sleep.test.ts @@ -6,6 +6,7 @@ import { join } from 'node:path' import { DaemonPtyAdapter } from '../daemon-pty-adapter' import { HistoryReader } from '../../../history/restoration/history-reader' import type * as CheckpointReaderModule from '../../../history/restoration/terminal-history-checkpoint-reader' +import type * as FileReaderModule from '../../../history/storage/terminal-history-file-reader' import { getHistorySessionDirName } from '../../../history/storage/history-paths' import { createMockSubprocess, @@ -41,6 +42,22 @@ vi.mock( } ) +// The final Sleep readback of a published checkpoint can still be in flight when its caller leaves. +const finalReadback = vi.hoisted(() => ({ during: async (_path: string): Promise => {} })) +vi.mock('../../../history/storage/terminal-history-file-reader', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + readTerminalHistoryBufferAsync: async ( + ...args: Parameters + ) => { + const bytes = await actual.readTerminalHistoryBufferAsync(...args) + await finalReadback.during(args[0]) + return bytes + } + } +}) + type MockChild = ReturnType let harness: Awaited> | undefined const adapters: DaemonPtyAdapter[] = [] @@ -51,6 +68,7 @@ afterEach(async () => { probe.cwd = null probe.during = () => {} sourceChange.during = () => {} + finalReadback.during = async () => {} for (const adapter of adapters.splice(0)) { adapter.dispose() } @@ -189,6 +207,47 @@ describe('host Sleep through disk-only Wake', () => { expect(child().kill).not.toHaveBeenCalled() }, 30_000) + // The caller leaves while the final checkpoint's readback is in flight, and a same-ID start + // arrives then too: the readback finishes, and the start sees the whole final cut. + it('finishes its final readback after the caller leaves, before a same-ID start', async () => { + const { open, child } = await startHost() + const sessionId = 'sleep-held-readback' + const adapter = open() + await adapter.spawn(request(sessionId)) + const slept = child() + slept._simulateData('HELD_ROW\r\n') + exitAfterOutput(slept, 'HELD_EXIT\r\n') + const readback = Promise.withResolvers() + const release = Promise.withResolvers() + finalReadback.during = async (path) => { + if (path.endsWith('checkpoint.json')) { + finalReadback.during = async () => {} + readback.resolve() + await release.promise + } + } + + const sleeping = adapter.shutdown(sessionId, { + keepHistory: true, + // ast-grep-ignore: test-no-wall-clock -- Sleep's deadline is an absolute epoch the real daemon compares against. + deadlineMs: Date.now() + 3_000 + }) + // Sync point only: the physical exit happened and its final checkpoint is being read back. + await readback.promise + await expect(sleeping).rejects.toThrow() + const successor = open().spawn(request(sessionId)) + release.resolve() + + const woken = await successor + expect(woken.coldRestore?.scrollback).toContain('HELD_ROW') + expect(woken.coldRestore?.scrollback).toContain('HELD_EXIT') + adapters.at(-1)!.write(sessionId, 'successor-input') + await waitFor(() => child().write.mock.calls.some(([data]) => data === 'successor-input')) + expect(child()).not.toBe(slept) + expect(slept.kill).toHaveBeenCalledTimes(1) + expect(child().kill).not.toHaveBeenCalled() + }, 30_000) + // A source that keeps changing under a cold read is not absent, corrupt or a fresh start. it('refuses a Wake whose saved source keeps changing, then restores it on Retry', async () => { const { open, child, historyPath } = await startHost() diff --git a/src/main/daemon/providers/adapter/testing/daemon-pty-adapter-host-history.test.ts b/src/main/daemon/providers/adapter/testing/daemon-pty-adapter-host-history.test.ts index e322b7ba810..d02ddb59c5b 100644 --- a/src/main/daemon/providers/adapter/testing/daemon-pty-adapter-host-history.test.ts +++ b/src/main/daemon/providers/adapter/testing/daemon-pty-adapter-host-history.test.ts @@ -1,6 +1,13 @@ import '../../../pty-subprocess/testing/mock-descendant-sweep' -import { afterEach, describe, expect, it } from 'vitest' -import { mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { + existsSync, + mkdirSync, + promises as fsPromises, + readFileSync, + rmSync, + writeFileSync +} from 'node:fs' import { join } from 'node:path' import { DaemonPtyAdapter } from '../daemon-pty-adapter' import { HeadlessEmulator } from '../../../emulation/headless-emulator' @@ -176,6 +183,47 @@ describe('daemon-owned terminal history', () => { expect(new HistoryReader(root).probeRestorableHistory(id).status).toBe('restorable') }, 30_000) + // AIO-C3: a recovery marker lookup that cannot tell used to start the process with no hold. + it('starts no process while the recovery marker cannot be checked, then starts normally', async () => { + let spawned = 0 + harness = await startDaemonAdapterHarness( + () => { + spawned += 1 + return createMockSubprocess() + }, + {}, + { hostHistory: true } + ) + const root = join(harness.dir, 'history') + adapter = new DaemonPtyAdapter({ + socketPath: harness.socketPath, + tokenPath: harness.tokenPath, + historyPath: root + }) + const request = { sessionId: 'host-history-uncertain', cols: 80, rows: 24, cwd: '/tmp' } + const dir = join(root, getHistorySessionDirName(request.sessionId)) + const stat = fsPromises.stat + const lookup = vi.spyOn(fsPromises, 'stat').mockImplementation(async (...args) => { + if (String(args[0]).endsWith('.unreadable-recovery')) { + throw Object.assign(new Error('i/o error'), { code: 'EIO' }) + } + return stat(...args) + }) + try { + await expect(adapter.spawn(request)).rejects.toThrow('terminal_history_unavailable') + expect(spawned).toBe(0) + // Nothing was reset, marked or fabricated for the ID. + expect(existsSync(join(dir, 'meta.json'))).toBe(false) + expect(existsSync(join(dir, '.unreadable-recovery'))).toBe(false) + } finally { + lookup.mockRestore() + } + + const { id } = await adapter.spawn(request) + expect(spawned).toBe(1) + expect(new HistoryReader(root).probeRestorableHistory(id).status).toBe('restorable') + }, 30_000) + // A start whose process never existed used to leave an ended predecessor marked restorable. it('restores the clean-ended predecessor when the fresh process never forked', async () => { let fail = true diff --git a/src/main/daemon/terminal-host/testing/terminal-host-concurrent-create.test.ts b/src/main/daemon/terminal-host/testing/terminal-host-concurrent-create.test.ts index 00bf3ed31c1..4edc53d4138 100644 --- a/src/main/daemon/terminal-host/testing/terminal-host-concurrent-create.test.ts +++ b/src/main/daemon/terminal-host/testing/terminal-host-concurrent-create.test.ts @@ -153,4 +153,71 @@ describe('a start that never produced a process', () => { expect(savedMeta()).toMatchObject({ endedAt: null, exitCode: null }) await host.dispose() }) + + // A same-ID start attempted while the rollback's rename is in flight waits behind it: canceling + // it then starts nothing, and a start that keeps waiting runs over the restored predecessor. + it('holds a same-ID start until a never-forked start has restored its predecessor', async () => { + const root = mkdtempSync(join(tmpdir(), 'orca-hold-rollback-')) + historyRoots.push(root) + const sessionId = 'hold-rollback-successor' + const dir = join(root, getHistorySessionDirName(sessionId)) + mkdirSync(dir, { recursive: true }) + const ended = { + ...newTerminalHistoryMeta({ cwd: '/tmp', cols: 80, rows: 24 }, null), + endedAt: '2026-11-01T11:00:00.000Z', + exitCode: 0 + } + writeTerminalHistoryMeta(dir, ended) + const savedMeta = () => JSON.parse(readFileSync(join(dir, 'meta.json'), 'utf8')) + let rollingBack = false + const rollback = Promise.withResolvers() + const release = Promise.withResolvers() + const rename = fsPromises.rename + vi.spyOn(fsPromises, 'rename').mockImplementation(async (from, to) => { + if (rollingBack) { + rollingBack = false + rollback.resolve() + await release.promise + } + return rename(from, to) + }) + let fail = true + const startedOver: unknown[] = [] + const host = new TerminalHost({ + historyJournals: new TerminalHistoryJournals(root), + spawnSubprocess: async (opts) => { + if (fail) { + rollingBack = true + opts.onCreationDisposition?.({ kind: 'no-fork' }) + throw new Error('spawn failed') + } + startedOver.push(savedMeta()) + return mockSubprocess() + } + }) + + const failed = host.createOrAttach(createOptions(sessionId)).then( + () => 'started', + () => savedMeta() + ) + // Sync point only: the rollback's restore rename is issued and held. + await rollback.promise + fail = false + const abort = new AbortController() + const canceled = host.createOrAttach({ + ...createOptions(sessionId), + cancelSignal: abort.signal + }) + const waiting = host.createOrAttach(createOptions(sessionId)) + abort.abort() + + await expect(canceled).rejects.toThrow('Attach canceled') + expect(startedOver).toEqual([]) + release.resolve() + expect(await failed).toEqual(ended) + await waiting + expect(startedOver).toEqual([expect.objectContaining({ endedAt: null, exitCode: null })]) + expect(savedMeta()).toMatchObject({ endedAt: null, exitCode: null }) + await host.dispose() + }) })