diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index aa27644990..9c86fe9f77 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -162,6 +162,18 @@ jobs: if: github.event_name != 'pull_request' || steps.e2e-marker.outputs.cache-hit != 'true' run: TEST_FILE=restart-persistence.test USE_MOCK=true xvfb-run -a pnpm --filter @roo-code/vscode-e2e exec node ./out/runTest.js + - name: Upload restart-persistence failure artifacts + # runTest.js preserves a failed scenario workspace (phase results, + # VS Code logs, task files) under apps/vscode-e2e/test-failures/ + # instead of deleting it, so the interrupted-write state from + # issue #2130-style failures stays inspectable. + if: failure() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: restart-persistence-failure + path: apps/vscode-e2e/test-failures/ + if-no-files-found: ignore + - name: Explain skipped mocked E2E pass marker if: steps.e2e-marker.outputs.cache-hit != 'true' && steps.run-e2e.outcome == 'success' && steps.vscode-fallback.outputs.used == 'true' run: echo "Skipping mocked E2E pass marker because tests ran against a stale cached VS Code binary (VS Code download endpoints unreachable)." diff --git a/.gitignore b/.gitignore index 1d4b255cea..e6692fa08d 100644 --- a/.gitignore +++ b/.gitignore @@ -28,6 +28,7 @@ AGENTS.local.md apps/vscode-e2e/playwright-report-electron/ apps/vscode-e2e/test-results-electron/ apps/vscode-e2e/heap-results/ +apps/vscode-e2e/test-failures/ # Docs docs/_site/ diff --git a/apps/vscode-e2e/src/restart/scenarioWorkspace.ts b/apps/vscode-e2e/src/restart/scenarioWorkspace.ts index 9ce49deaf0..b8b768ef31 100644 --- a/apps/vscode-e2e/src/restart/scenarioWorkspace.ts +++ b/apps/vscode-e2e/src/restart/scenarioWorkspace.ts @@ -56,3 +56,56 @@ export async function removeScenarioWorkspace(scenarioWorkspace: ScenarioWorkspa await fs.rm(root, { recursive: true, force: true }) } + +const FAILURE_ARTIFACTS_DIR = "test-failures" + +/** + * Copies the diagnosable parts of a failed scenario workspace into a stable + * directory inside the checkout, then removes the temporary scenario root. + * CI uploads that directory as a run artifact, so the failure that used to be + * deleted in the cleanup path stays inspectable: phase results, VS Code logs, + * and the extension's global storage with the task files. + * + * The directory is keyed by the unique scenario root name and cleared before + * copying, so retried runs cannot mix artifacts from different failures. + */ +export async function preserveScenarioWorkspace(scenarioWorkspace: ScenarioWorkspace): Promise { + const destinationRoot = path.resolve( + __dirname, + "../../", + FAILURE_ARTIFACTS_DIR, + path.basename(scenarioWorkspace.root), + ) + await fs.rm(destinationRoot, { recursive: true, force: true }) + await fs.mkdir(destinationRoot, { recursive: true }) + + const selections: Array<{ label: string; source: string; destination: string }> = [ + { + label: "phase results", + source: scenarioWorkspace.results, + destination: path.join(destinationRoot, "phase-results"), + }, + { + label: "VS Code logs", + source: path.join(scenarioWorkspace.userData, "logs"), + destination: path.join(destinationRoot, "vscode-logs"), + }, + { + label: "task files", + source: path.join(scenarioWorkspace.userData, "User", "globalStorage"), + destination: path.join(destinationRoot, "global-storage"), + }, + ] + + for (const selection of selections) { + try { + await fs.cp(selection.source, selection.destination, { recursive: true, force: true }) + } catch (error) { + // A missing subdirectory must not hide the other artifacts. + console.error(`Failed to preserve ${selection.label} from ${selection.source}:`, error) + } + } + + await removeScenarioWorkspace(scenarioWorkspace) + return destinationRoot +} diff --git a/apps/vscode-e2e/src/runTest.ts b/apps/vscode-e2e/src/runTest.ts index 88c687bc76..f24f37d388 100644 --- a/apps/vscode-e2e/src/runTest.ts +++ b/apps/vscode-e2e/src/runTest.ts @@ -21,7 +21,11 @@ import { addSearchFilesResultFixtures } from "./fixtures/search-files" import { addSubtaskFixtures } from "./fixtures/subtasks" import { addUseMcpToolResultFixtures } from "./fixtures/use-mcp-tool" import { addWriteToFileResultFixtures } from "./fixtures/write-to-file" -import { createScenarioWorkspace, removeScenarioWorkspace } from "./restart/scenarioWorkspace" +import { + createScenarioWorkspace, + preserveScenarioWorkspace, + removeScenarioWorkspace, +} from "./restart/scenarioWorkspace" import { runRestartScenario } from "./restart/vscodeCoordinator" function getCliFlagValue(flag: string) { @@ -95,6 +99,7 @@ async function main() { let testWorkspace: string | undefined let scenarioWorkspace: Awaited> | undefined + let runFailed = false try { // Create a temporary workspace folder for regular tests. Restart scenarios own @@ -218,6 +223,7 @@ async function main() { }) } } catch (error) { + runFailed = true console.error("Failed to run tests", error) process.exitCode = 1 } finally { @@ -225,7 +231,19 @@ async function main() { await fs.rm(testWorkspace, { recursive: true, force: true }) } if (scenarioWorkspace) { - await removeScenarioWorkspace(scenarioWorkspace) + // A failed restart scenario must stay inspectable: the phase + // results, VS Code logs, and task files are copied to a stable + // directory for CI artifact upload before the temp root goes away. + if (runFailed) { + try { + const preservedAt = await preserveScenarioWorkspace(scenarioWorkspace) + console.error(`Restart scenario workspace preserved for failure inspection: ${preservedAt}`) + } catch (preserveError) { + console.error("Failed to preserve the restart scenario workspace:", preserveError) + } + } else { + await removeScenarioWorkspace(scenarioWorkspace) + } } await mock?.stop() } diff --git a/apps/vscode-e2e/src/suite/restart-persistence.test.ts b/apps/vscode-e2e/src/suite/restart-persistence.test.ts index 0d508273e3..47a42d0172 100644 --- a/apps/vscode-e2e/src/suite/restart-persistence.test.ts +++ b/apps/vscode-e2e/src/suite/restart-persistence.test.ts @@ -54,12 +54,13 @@ async function runCreate(api: RooCodeAPI): Promise { await waitUntilCompleted({ api, taskId }) assert.strictEqual(sawMarker, true, `Completion should include ${MARKER}`) - // TaskCompleted fires before the final saveClineMessages() write is - // necessarily durable on disk. Wait until the completion marker is - // persisted in ui_messages.json before quitting, so the verify phase - // never reads a task directory whose final write has not landed yet. + // TaskCompleted and the UI marker do not acknowledge all three files. + // Check both persisted histories and the store's published history item. + // TaskHistoryStore publishes newly created cache entries only after writing them. const createdTaskId: string = taskId await waitFor(() => api.hasTaskUiMessageText(createdTaskId, MARKER)) + await waitForMarkedCompletion(api, createdTaskId) + await waitFor(async () => (await api.getTaskHistoryItem(createdTaskId)) !== undefined) const result: PhaseResult = { version: PHASE_RESULT_VERSION, diff --git a/docs/architecture/file-replacement-restart-model.md b/docs/architecture/file-replacement-restart-model.md new file mode 100644 index 0000000000..7b15980cbb --- /dev/null +++ b/docs/architecture/file-replacement-restart-model.md @@ -0,0 +1,78 @@ +# File replacement and restart model + +`scripts/check-file-replacement-restart.ts` is an independent bounded submodel +under `pnpm lifecycle:model-check`. It preserves the completion-readiness model's +separate scope. It checks three files, one initial publication per file, one later +replacement of any one file, and one process crash at every publication or +replacement boundary. Exploration runs to a fixed point with a 10,000-state cap. + +## Contract and production mapping + +| Model action | Production boundary | +| ---------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `publish-ui`, `publish-api`, `publish-history` | Successful writes of `ui_messages.json`, `api_conversation_history.json`, and `history_item.json` | +| `acknowledge-all-three-files` | Explicit readiness conjunction in the restart E2E: `hasTaskUiMessageText`, `hasTaskApiConversationHistorySequence`, and `getTaskHistoryItem`. For a newly created task, `TaskHistoryStore.upsertCore` publishes its cache entry after `writeTaskFile` succeeds. This is a test readiness boundary, not a new public runtime event. | +| `stage-replacement-*` | `safeWriteJson` writes and flushes a `.new_*.tmp` file under the advisory lock | +| `atomic-commit` | Single temp-to-canonical rename | +| `eperm-backup-rename`, `fallback-commit` | Windows-compatible EPERM fallback: canonical-to-backup, then temp-to-canonical | +| `process-crash` | Process exits without rollback, cleanup, or lock release | +| `lock-becomes-stale` | `proper-lockfile` staleness check using the shared 31-second window | +| `fresh-host-startup`, `locked-reader-recovery` | `TaskHistoryStore.reconcile` recovers a missing history record before stat/read; `readFileWithMissingRetry` invokes the same recovery after a second ENOENT for UI/API reads | +| `recovery-io-failure` | Recovery scan, access, stat, or restore fails; rejection preserves committed backups for a later retry | +| `explicit-delete-under-lock` | `deleteJsonFile` removes matching replacements/backups before the canonical file, serialized with writers and recovery; store deletion remains best effort | + +The UI marker can become visible while history publication is still pending. +`Task.saveClineMessages` persists UI messages before updating metadata, and logs +metadata failures independently. Neither `TaskCompleted` nor UI-marker visibility +promises that all three files exist. The model explicitly reaches that partial +publication state and permits interruption there. It requires restart discovery +only after the readiness conjunction, not after an unfinished first write. + +After acknowledgment, the latest committed version of each required file must +remain in either the canonical file or its backup until explicit deletion. +Completed fresh reads must return those committed versions and find the history +record without a new write. A failed recovery must preserve every file version. +Successful explicit deletion must leave no recoverable version to resurrect. +The model requires reachable atomic and fallback successes, restart recovery, +recovery after failure, partial publication, and reads after deletion. + +The checker first runs the old two-rename protocol without reader recovery and +requires a #2130 counterexample. It then exhaustively checks the fixed protocol. +This is an abstract protocol check, not execution of the production writer. + +## Refinement evidence + +`src/utils/__tests__/safeWriteJson.interruption.test.ts` runs the real bundled +writer in a child process, terminates it with SIGKILL, and exercises both commit +paths before and after the replacement rename for all three filenames. Initial +publication uses the production writer. UI/API tests call production readers; +history-record cases initialize a fresh `TaskHistoryStore` before any recovery +write. One case leaves the crash lock's timestamp unchanged, proving the fresh +store can wait for staleness instead of requiring the test to age the lock. + +`safeWriteJson.test.ts` injects directory scan, access, backup stat, and restore +failures, checking backup preservation and retry. The store deletion tests cover +single and batch deletion, leftover cleanup, unrelated-file preservation, and +cleanup failure retaining the canonical copy. The existing two-extension-host +restart E2E remains the production activation/rehydration boundary check. + +## Assumptions and limits + +This is process-crash safety on filesystems with atomic same-directory rename, +not power-loss safety. Parent directories are not fsynced. Locks are advisory; +all writers, recoverers, and history-file deleters must use the shared protocol. +Live holders refresh their locks and must not be mistaken for crashed holders; +filesystem clock behavior and lock compromise are outside the model. Recovery +waits up to 35 lock retries at one-second intervals, then reports a lock failure. +An indefinitely active writer or persistent filesystem fault can still prevent +recovery; no liveness guarantee is claimed under such faults. + +Recovery is abstracted as a locked restore and cleanup; repeated crashes during +recovery, multiple orphaned versions, external file removal, arbitrary overlapping +writers, and deletion interrupted between individual unlinks are not exhaustively +modeled. Focused tests cover newest-backup selection and filesystem failures. +Deletion does not prohibit an explicit later creation of a task at the same path +or a stale writer that deliberately publishes new data after deletion. The +no-resurrection property concerns recovery from leftovers after successful +deletion. Lifecycle status/delegation reducers and completion-event semantics +are unchanged. diff --git a/docs/architecture/task-lifecycle-model.md b/docs/architecture/task-lifecycle-model.md index b7b39e600d..081cd96a0c 100644 --- a/docs/architecture/task-lifecycle-model.md +++ b/docs/architecture/task-lifecycle-model.md @@ -206,6 +206,14 @@ CI runs `pnpm lifecycle:model-check` in `.github/workflows/code-qa.yml` after li ## Gap audit +The independent [file replacement/restart model](./file-replacement-restart-model.md) +checks the process-crash replacement protocol separately from completion readiness. +It models three task files, initial publication, one replacement, one crash, +locked recovery on fresh reads/startup, and successful explicit deletion. Its +contract, production mapping, old-protocol counterexample, refinement tests, and +filesystem/lock assumptions are documented there. A passing completion-readiness +model alone does not establish this file-level contract. + This section is a summary tracker. Issue links are historical provenance rather than specifications. Coverage status uses these evidence classes: The [Task lifecycle verification GAP report](./task-lifecycle-gap-report.md) is the authoritative register. It holds the exhaustive repository inventory, the ranked stable burn-down register, the source-based methodology, and the follow-up checklist. This page remains the executable model-suite specification. diff --git a/package.json b/package.json index b0c290b97c..a67f824dcc 100644 --- a/package.json +++ b/package.json @@ -20,7 +20,7 @@ "test:agent-evals:unit": "node --import tsx --test scripts/agent-evals/*.test.ts", "agent-evals:check-types": "tsc -p scripts/agent-evals/tsconfig.json", "test:mutation-ci": "node --test scripts/stryker-diff.test.mjs", - "lifecycle:model-check": "tsx scripts/check-task-lifecycle.ts && tsx scripts/check-task-store-concurrency.ts && tsx scripts/check-provider-handoff-scheduler.ts && pnpm cleanup-protocol:model-check && pnpm parser-scope:model-check && tsx scripts/check-completion-persistence.ts && tsx scripts/check-delegated-mode-readers.ts && pnpm tool-turn:model-check", + "lifecycle:model-check": "tsx scripts/check-task-lifecycle.ts && tsx scripts/check-task-store-concurrency.ts && tsx scripts/check-provider-handoff-scheduler.ts && pnpm cleanup-protocol:model-check && pnpm parser-scope:model-check && tsx scripts/check-completion-persistence.ts && tsx scripts/check-file-replacement-restart.ts && tsx scripts/check-delegated-mode-readers.ts && pnpm tool-turn:model-check", "webview-recovery:model-check": "node --import tsx scripts/check-webview-recovery.ts", "webview-recovery:check-types": "tsc -p scripts/tsconfig.webview-recovery.json", "fanout-protocol:model-check": "tsx scripts/check-task-fanout-protocol.ts", diff --git a/scripts/check-file-replacement-restart.ts b/scripts/check-file-replacement-restart.ts new file mode 100644 index 0000000000..c6e7410441 --- /dev/null +++ b/scripts/check-file-replacement-restart.ts @@ -0,0 +1,200 @@ +/** Independent bounded abstraction; see docs/architecture/file-replacement-restart-model.md. */ +import assert from "node:assert/strict" + +type Version = 0 | 1 | 2 +type Phase = "publishing" | "idle" | "temp" | "backup" | "committed" | "crashed" | "reading" | "read" | "deleted" +interface FileState { + canonical: Version + backup: Version + temp: Version + committed: Version +} +interface State { + files: FileState[] + phase: Phase + target: number + acknowledged: boolean + lock: "none" | "live" | "fresh-crashed" | "stale" + readIndex: number + recoveryFailed: boolean + deleted: boolean +} +interface Edge { + action: string + state: State +} +const FILES = ["ui", "api", "history"] as const +const MAX_STATES = 10_000 +const landmarks = new Set() + +function edges(state: State, fixed: boolean): Edge[] { + const result: Edge[] = [] + function add(action: string, change: (next: State) => void) { + const next = structuredClone(state) + change(next) + result.push({ action, state: next }) + } + if (state.phase === "publishing") { + // Separate file publications: UI visibility does not acknowledge history publication. + for (const [index, file] of state.files.entries()) { + if (!file.canonical) + add(`publish-${FILES[index]}`, (next) => { + next.files[index] = { canonical: 1, backup: 0, temp: 0, committed: 1 } + if (next.files.every((entry) => entry.canonical)) next.phase = "idle" + }) + } + } + if (state.phase === "idle" && !state.acknowledged) { + add("acknowledge-all-three-files", (next) => { + next.acknowledged = true + }) + } + if (state.phase === "idle" && state.acknowledged) { + for (const index of [0, 1, 2]) + add(`stage-replacement-${FILES[index]}`, (next) => { + next.target = index + next.files[index].temp = 2 + next.phase = "temp" + next.lock = "live" + }) + } + if (state.phase === "temp") { + if (fixed) + add("atomic-commit", (next) => { + next.files[next.target].canonical = 2 + next.files[next.target].committed = 2 + next.files[next.target].temp = 0 + next.phase = "committed" + }) + add(fixed ? "eperm-backup-rename" : "old-backup-rename", (next) => { + const file = next.files[next.target] + file.backup = file.canonical + file.canonical = 0 + next.phase = "backup" + }) + } + if (state.phase === "backup") + add("fallback-commit", (next) => { + const file = next.files[next.target] + file.canonical = file.temp + file.committed = file.temp + file.temp = 0 + next.phase = "committed" + }) + // One process crash, at every publication/replacement boundary, including post-commit cleanup. + if (["publishing", "idle", "temp", "backup", "committed"].includes(state.phase)) { + add("process-crash", (next) => { + next.phase = "crashed" + next.lock = state.lock === "live" ? "fresh-crashed" : "none" + }) + } + if (state.lock === "fresh-crashed") + add("lock-becomes-stale", (next) => { + next.lock = "stale" + }) + if (state.phase === "crashed") + add("fresh-host-startup", (next) => { + next.phase = "reading" + }) + if (state.phase === "reading") { + const file = state.files[state.readIndex] + if (file.canonical || !fixed || (!file.backup && !file.temp)) { + add(`read-${FILES[state.readIndex]}`, (next) => { + next.readIndex++ + if (next.readIndex === FILES.length) next.phase = "read" + }) + } else if (state.lock !== "fresh-crashed") { + // Failed scan/access/stat/rename preserves every committed byte and permits retry. + if (!state.recoveryFailed) + add("recovery-io-failure", (next) => { + next.recoveryFailed = true + }) + add("locked-reader-recovery", (next) => { + const recovering = next.files[next.readIndex] + recovering.canonical = recovering.backup + recovering.backup = 0 + recovering.temp = 0 + next.lock = "none" + }) + } + } + if (["crashed", "read"].includes(state.phase) && state.lock !== "fresh-crashed") { + add("explicit-delete-under-lock", (next) => { + next.files = next.files.map(() => ({ canonical: 0, backup: 0, temp: 0, committed: 0 })) + next.phase = "deleted" + next.deleted = true + next.lock = "none" + }) + } + if (state.phase === "deleted") + add("read-after-delete", (next) => { + next.phase = "read" + }) + return result +} + +function violation(state: State): string | undefined { + if (state.deleted && state.files.some((file) => file.canonical || file.backup)) return "deleted history resurrected" + if (state.acknowledged && !state.deleted) { + if (state.files.some((file) => file.canonical !== file.committed && file.backup !== file.committed)) { + return "last committed content lost" + } + if (state.phase === "read" && state.files.some((file) => file.canonical !== file.committed)) { + return "acknowledged task not discoverable/recoverable on fresh startup (#2130)" + } + } + return undefined +} + +function explore(fixed: boolean): number { + const initial: State = { + files: FILES.map(() => ({ canonical: 0, backup: 0, temp: 0, committed: 0 })), + phase: "publishing", + target: 0, + acknowledged: false, + lock: "none", + readIndex: 0, + recoveryFailed: false, + deleted: false, + } + const queue = [{ state: initial, trace: [] as string[] }] + const seen = new Set([JSON.stringify(initial)]) + for (let index = 0; index < queue.length; index++) { + const { state, trace } = queue[index] + const error = violation(state) + if (error) { + if (!fixed && error.includes("#2130")) { + console.log(`Old protocol #2130 counterexample: ${trace.join(" -> ")}`) + return seen.size + } + throw new Error(`${error}: ${trace.join(" -> ")}\n${JSON.stringify(state)}`) + } + if (fixed) { + if (state.phase === "crashed" && state.files[0].canonical && !state.files[2].canonical) + landmarks.add("ui-visible-history-pending") + if (state.phase === "read" && state.acknowledged && !state.deleted) landmarks.add("acknowledged-restart") + if (state.phase === "read" && state.recoveryFailed && !state.deleted) + landmarks.add("retry-after-recovery-failure") + if (state.phase === "read" && state.deleted) landmarks.add("no-resurrection") + if (state.phase === "committed" && !state.files[state.target].backup) landmarks.add("atomic-success") + if (state.phase === "committed" && state.files[state.target].backup) landmarks.add("fallback-success") + } + for (const edge of edges(state, fixed)) { + if (fixed && edge.action === "recovery-io-failure") assert.deepEqual(edge.state.files, state.files) + const key = JSON.stringify(edge.state) + if (seen.has(key)) continue + seen.add(key) + assert.ok(seen.size <= MAX_STATES, "Replacement/restart state budget exceeded") + queue.push({ state: edge.state, trace: [...trace, edge.action] }) + } + } + assert.ok(fixed, "Old protocol must reproduce #2130") + return seen.size +} + +explore(false) +const checked = explore(true) +assert.equal(landmarks.size, 6, `Unreachable replacement/restart landmarks: ${[...landmarks]}`) +console.log( + `File replacement/restart model passed: ${checked} states, 3 files, initial publication + 1 replacement, 1 crash, 6/6 landmarks`, +) diff --git a/src/core/task-persistence/TaskHistoryStore.ts b/src/core/task-persistence/TaskHistoryStore.ts index 346d697d9a..c1c50cca1e 100644 --- a/src/core/task-persistence/TaskHistoryStore.ts +++ b/src/core/task-persistence/TaskHistoryStore.ts @@ -7,8 +7,10 @@ import deepEqual from "fast-deep-equal" import { historyItemSchema, type HistoryItem } from "@roo-code/types" import { GlobalFileNames } from "../../shared/globalFileNames" -import { LOCK_STALE_MS, withFileLock } from "../../utils/fileLock" +import { LOCK_STALE_MS } from "../../utils/fileLock" import { safeWriteJson } from "../../utils/safeWriteJson" +import { deleteJsonFile, recoverMissingJsonFile } from "../../utils/jsonFileRecovery" +import { getErrorCode, readFileWithMissingRetry } from "./readFileWithMissingRetry" import { getStorageBasePath } from "../../utils/storage" import { assertValidTransition, settleRejectedCreateSubtaskAction, type HistoryItemStatus } from "./taskLifecycle" import { computeHistoryDelta, DeltaRejectedError, mergeHistoryDelta } from "./taskStoreConcurrency" @@ -280,8 +282,10 @@ export class TaskHistoryStore { // Remove per-task file (best-effort) try { const filePath = await this.getTaskFilePath(taskId) - await withFileLock(filePath, (absoluteFilePath) => fs.unlink(absoluteFilePath)) - } catch { + await deleteJsonFile(filePath) + } catch (error) { + if (getErrorCode(error) !== "ENOENT") + console.warn("[TaskHistoryStore] Failed to delete task history:", error) // File may already be deleted } @@ -308,8 +312,10 @@ export class TaskHistoryStore { // Remove per-task file (best-effort) try { const filePath = await this.getTaskFilePath(taskId) - await withFileLock(filePath, (absoluteFilePath) => fs.unlink(absoluteFilePath)) - } catch { + await deleteJsonFile(filePath) + } catch (error) { + if (getErrorCode(error) !== "ENOENT") + console.warn("[TaskHistoryStore] Failed to delete task history:", error) // File may already be deleted } } @@ -351,7 +357,11 @@ export class TaskHistoryStore { for (const taskId of onDiskIds) { try { const taskFilePath = await this.getTaskFilePath(taskId) - const { mtimeMs } = await fs.stat(taskFilePath) + const { mtimeMs } = await fs.stat(taskFilePath).catch(async (error: unknown) => { + if (getErrorCode(error) !== "ENOENT") throw error + await recoverMissingJsonFile(taskFilePath, { waitForStaleLock: false }) + return fs.stat(taskFilePath) + }) liveIds.add(taskId) if ( !options.forceRefresh && @@ -369,7 +379,12 @@ export class TaskHistoryStore { this.cache.set(taskId, item) } } - } catch { + } catch (error) { + if (getErrorCode(error) !== "ENOENT") { + liveIds.add(taskId) + console.warn("[TaskHistoryStore] Failed to reconcile task history:", error) + continue + } // File may be temporarily absent during a peer's atomic // rename window in safeWriteJson. The advisory lock is // held for the entire write, so its presence means a @@ -660,7 +675,7 @@ export class TaskHistoryStore { const intentPath = await this.getDelegationRepairIntentPath() let parsed: unknown try { - parsed = JSON.parse(await fs.readFile(intentPath, "utf8")) as unknown + parsed = JSON.parse(await readFileWithMissingRetry(intentPath)) as unknown } catch (error) { if (this.isFileNotFoundError(error)) { return null @@ -720,7 +735,7 @@ export class TaskHistoryStore { private async removeDelegationRepairIntent(): Promise { try { - await fs.unlink(await this.getDelegationRepairIntentPath()) + await deleteJsonFile(await this.getDelegationRepairIntentPath()) } catch (error) { console.warn("[TaskHistoryStore] Failed to remove completed delegation repair intent:", error) } @@ -828,10 +843,21 @@ export class TaskHistoryStore { try { await fs.access(filePath) // File already exists, skip (don't overwrite existing per-task files) - } catch { - // File doesn't exist, write it - await safeWriteJson(filePath, item) - this.cache.set(item.id, item) + } catch (error) { + if (getErrorCode(error) !== "ENOENT") throw error + // Recovery or a peer may supply a committed record after the access check. + let persisted = item + await safeWriteJson(filePath, item, { + merge: (existing) => { + if (existing !== null) { + const recovered = historyItemSchema.parse(existing) + if (recovered.id !== item.id) throw new Error("Migration task ID mismatch") + persisted = recovered + } + return persisted + }, + }) + this.cache.set(item.id, persisted) } } @@ -888,7 +914,7 @@ export class TaskHistoryStore { const filePath = await this.getTaskFilePath(taskId) try { - const raw = await fs.readFile(filePath, "utf8") + const raw = await readFileWithMissingRetry(filePath) const item: HistoryItem = JSON.parse(raw) return item.id ? item : null } catch { diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.deleteSemantics.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.deleteSemantics.spec.ts index 5071c9de73..875c7dcb9a 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.deleteSemantics.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.deleteSemantics.spec.ts @@ -13,6 +13,8 @@ import type { HistoryItem } from "@roo-code/types" import { TaskHistoryStore } from "../TaskHistoryStore" import { withFileLock } from "../../../utils/fileLock" import { GlobalFileNames } from "../../../shared/globalFileNames" +import { readFileWithMissingRetry } from "../readFileWithMissingRetry" +import { safeWriteJson } from "../../../utils/safeWriteJson" vi.mock("../../../utils/storage", () => ({ getStorageBasePath: vi.fn().mockImplementation((defaultPath: string) => defaultPath), @@ -90,6 +92,61 @@ describe("TaskHistoryStore best-effort deletion semantics", () => { return store } + it.each(["delete", "deleteMany"] as const)( + "%s removes backups before deletion so readers and merges cannot resurrect history", + async (operation) => { + const store = createStore() + await store.initialize() + const history = makeHistoryItem({ id: "with-leftovers" }) + await store.upsert(history) + const filePath = historyFilePath(storagePath, history.id) + const dir = path.dirname(filePath) + const backup = path.join(dir, `.${path.basename(filePath)}.bak_1_saved.tmp`) + const replacement = path.join(dir, `.${path.basename(filePath)}.new_1_pending.tmp`) + const unrelated = path.join(dir, ".other.json.bak_1_saved.tmp") + await fs.writeFile(backup, JSON.stringify(history)) + await fs.writeFile(replacement, "{}") + await fs.writeFile(unrelated, "keep") + if (operation === "delete") await store.delete(history.id) + else await store.deleteMany([history.id]) + + await expect(readFileWithMissingRetry(filePath)).rejects.toMatchObject({ code: "ENOENT" }) + expect(await fs.readdir(dir)).toEqual([path.basename(unrelated)]) + await safeWriteJson( + filePath, + {}, + { + merge: (existing) => { + expect(existing).toBeNull() + return { fresh: true } + }, + }, + ) + expect(JSON.parse(await fs.readFile(filePath, "utf8"))).toEqual({ fresh: true }) + }, + ) + + it.each(["delete", "deleteMany"] as const)( + "%s preserves the canonical file if backup cleanup fails", + async (operation) => { + const store = createStore() + await store.initialize() + const history = makeHistoryItem({ id: "blocked-cleanup" }) + await store.upsert(history) + const filePath = historyFilePath(storagePath, history.id) + const backup = path.join(path.dirname(filePath), `.${path.basename(filePath)}.bak_1_saved.tmp`) + await fs.writeFile(backup, JSON.stringify(history)) + vi.mocked(fs.unlink).mockImplementation(async (target) => { + if (target === backup) throw Object.assign(new Error("Cleanup failed"), { code: "EACCES" }) + return actualFs.unlink(target) + }) + if (operation === "delete") await store.delete(history.id) + else await store.deleteMany([history.id]) + expect(JSON.parse(await fs.readFile(filePath, "utf8"))).toEqual(history) + expect(JSON.parse(await fs.readFile(backup, "utf8"))).toEqual(history) + }, + ) + describe("delete()", () => { it("unlinks under the shared per-file lock, evicts cache and mtime, and writes through once", async () => { const store = createStore() diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.realConcurrency.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.realConcurrency.spec.ts index d35dfc49cf..e3e4445f35 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.realConcurrency.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.realConcurrency.spec.ts @@ -91,6 +91,24 @@ function createAction(actionId: string, message: string) { } describe("TaskHistoryStore real cross-host locking", () => { + it("preserves a recovered record instead of overwriting it during legacy migration", async () => { + const storagePath = await fs.mkdtemp(path.join(os.tmpdir(), "task-history-migrate-")) + const store = new TaskHistoryStore(storagePath) + try { + const legacy = item("migrated") + const committed = { ...legacy, task: "committed newer task", totalCost: 7 } + const taskDir = path.join(storagePath, "tasks", legacy.id) + await fs.mkdir(taskDir, { recursive: true }) + await fs.writeFile(path.join(taskDir, ".history_item.json.bak_1.tmp"), JSON.stringify(committed)) + await store.migrateFromGlobalState([legacy]) + expect(store.get(legacy.id)).toEqual(committed) + expect(JSON.parse(await fs.readFile(path.join(taskDir, "history_item.json"), "utf8"))).toEqual(committed) + } finally { + store.dispose() + await fs.rm(storagePath, { recursive: true, force: true }) + } + }) + it("preserves independent stale-cache deltas through the real per-file lock", async () => { const storagePath = await fs.mkdtemp(path.join(os.tmpdir(), "task-history-real-lock-")) const storeA = new TaskHistoryStore(storagePath) diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts index e37fd1a25e..0d1f0de9d5 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts @@ -3,6 +3,7 @@ import * as fs from "fs/promises" import * as path from "path" import * as os from "os" +import * as recovery from "../../../utils/jsonFileRecovery" import type { HistoryItem } from "@roo-code/types" @@ -167,12 +168,28 @@ describe("TaskHistoryStore reconcileDelegationState", () => { }) afterEach(async () => { + vi.restoreAllMocks() safeWriteJsonMock.mockImplementation(writeJson) for (const disposable of disposables) disposable.dispose() disposables.clear() await fs.rm(tmpDir, { recursive: true, force: true }).catch(() => {}) }) + it.each(["EACCES", "EIO", "ELOCKED", "missing"])("handles failed startup recovery: %s", async (code) => { + const item = makeItem({ id: "recovery-failure", status: "active" }) + await seedItems([item]) + await store.initialize() + store.dispose() + const target = path.join(tmpDir, "tasks", item.id, GlobalFileNames.historyItem) + await fs.rename(target, path.join(path.dirname(target), `.${path.basename(target)}.bak_1.tmp`)) + const recover = vi.spyOn(recovery, "recoverMissingJsonFile") + if (code === "missing") recover.mockResolvedValue(false) + else recover.mockRejectedValue(Object.assign(new Error("recovery failed"), { code })) + await store.reconcile() + expect(recover).toHaveBeenCalledWith(target, { waitForStaleLock: false }) + expect(store.get(item.id)).toEqual(code === "missing" ? undefined : item) + }) + it("repairs orphaned delegation: delegated parent whose child does not exist → active", async () => { const parent = makeItem({ id: "parent-1", status: "delegated", awaitingChildId: "missing-child" }) await seedItems([parent]) @@ -444,26 +461,36 @@ describe("TaskHistoryStore reconcileDelegationState", () => { await expect(fs.access(intentPath)).rejects.toThrow() }) - it("replays a both-at-target intent without writing task files", async () => { - const child = makeItem({ id: "child-at-target", status: "interrupted", parentTaskId: "parent-at-target" }) - const parent = makeItem({ id: "parent-at-target", status: "active" }) - await seedItems([parent, child]) - const intentPath = path.join(tmpDir, "tasks", GlobalFileNames.delegationRepairIntent) - await fs.writeFile( - intentPath, - JSON.stringify(makeRepairIntent({ ...parent, status: "delegated", awaitingChildId: child.id }, child)), - ) - - const writeCalls: string[] = [] - safeWriteJsonMock.mockImplementation(async (filePath, data) => { - writeCalls.push(filePath) - await writeJson(filePath, data) - }) - await store.initialize() + it.each([false, true])( + "replays a both-at-target intent without writing task files (backup: %s)", + async (backup) => { + const child = makeItem({ id: "child-at-target", status: "interrupted", parentTaskId: "parent-at-target" }) + const parent = makeItem({ id: "parent-at-target", status: "active" }) + await seedItems([parent, child]) + const intentPath = path.join(tmpDir, "tasks", GlobalFileNames.delegationRepairIntent) + await fs.writeFile( + intentPath, + JSON.stringify(makeRepairIntent({ ...parent, status: "delegated", awaitingChildId: child.id }, child)), + ) - expect(writeCalls.filter((filePath) => filePath.endsWith(GlobalFileNames.historyItem))).toEqual([]) - await expect(fs.access(intentPath)).rejects.toThrow() - }) + if (backup) { + await fs.rename( + intentPath, + path.join(path.dirname(intentPath), `.${path.basename(intentPath)}.bak_1.tmp`), + ) + } + + const writeCalls: string[] = [] + safeWriteJsonMock.mockImplementation(async (filePath, data) => { + writeCalls.push(filePath) + await writeJson(filePath, data) + }) + await store.initialize() + + expect(writeCalls.filter((filePath) => filePath.endsWith(GlobalFileNames.historyItem))).toEqual([]) + await expect(fs.access(intentPath)).rejects.toThrow() + }, + ) it("removes the repair-intent file after successful replay", async () => { const child = makeItem({ id: "child-deferred-index", status: "active", parentTaskId: "parent-deferred-index" }) diff --git a/src/core/task-persistence/__tests__/persistenceTrace.spec.ts b/src/core/task-persistence/__tests__/persistenceTrace.spec.ts new file mode 100644 index 0000000000..c4a58afce4 --- /dev/null +++ b/src/core/task-persistence/__tests__/persistenceTrace.spec.ts @@ -0,0 +1,59 @@ +import { describe, it, expect, vi, afterEach } from "vitest" + +import { setPersistenceTraceSink, tracePersistenceEvent } from "../persistenceTrace" + +function parseTraceLine(line: string): Record { + return JSON.parse(line.replace("[persistence-trace] ", "")) as Record +} + +describe("persistenceTrace", () => { + afterEach(() => { + setPersistenceTraceSink(undefined) + vi.restoreAllMocks() + }) + + it("emits a JSON line with the timestamp, process ID, stage, and fields to the sink", () => { + const lines: string[] = [] + setPersistenceTraceSink((line) => lines.push(line)) + + tracePersistenceEvent("ui-messages-write-start", { taskId: "task-1", filePath: "/tmp/ui_messages.json" }) + + expect(lines).toHaveLength(1) + const event = parseTraceLine(lines[0]) + expect(event.stage).toBe("ui-messages-write-start") + expect(event.taskId).toBe("task-1") + expect(event.pid).toBe(process.pid) + expect(typeof event.ts).toBe("string") + expect(event.filePath).toBe("/tmp/ui_messages.json") + }) + + it("marks the marker-check result", () => { + const lines: string[] = [] + setPersistenceTraceSink((line) => lines.push(line)) + + tracePersistenceEvent("ui-messages-marker-check", { taskId: "task-2", result: "match" }) + + const event = parseTraceLine(lines[0]) + expect(event.stage).toBe("ui-messages-marker-check") + expect(event.result).toBe("match") + }) + + it("falls back to the console when no sink is installed", () => { + const consoleSpy = vi.spyOn(console, "log").mockImplementation(() => {}) + + tracePersistenceEvent("shutdown") + + expect(consoleSpy).toHaveBeenCalledTimes(1) + const line = String(consoleSpy.mock.calls[0]?.[0]) + expect(line).toContain("[persistence-trace]") + expect(parseTraceLine(line).stage).toBe("shutdown") + }) + + it("never lets a failing sink break the caller", () => { + setPersistenceTraceSink(() => { + throw new Error("sink failed") + }) + + expect(() => tracePersistenceEvent("ui-messages-marker-check", { result: "match" })).not.toThrow() + }) +}) diff --git a/src/core/task-persistence/__tests__/readFileWithMissingRetry.spec.ts b/src/core/task-persistence/__tests__/readFileWithMissingRetry.spec.ts new file mode 100644 index 0000000000..bf2a7a3dd6 --- /dev/null +++ b/src/core/task-persistence/__tests__/readFileWithMissingRetry.spec.ts @@ -0,0 +1,29 @@ +import * as fs from "fs/promises" +import * as os from "os" +import * as path from "path" + +import { readFileWithMissingRetry } from "../readFileWithMissingRetry" + +describe("readFileWithMissingRetry", () => { + let dir: string + beforeEach(async () => { + dir = await fs.mkdtemp(path.join(os.tmpdir(), "missing-retry-")) + }) + afterEach(async () => { + await fs.rm(dir, { recursive: true, force: true }) + }) + + it("reads an existing file", async () => { + const target = path.join(dir, "data.json") + await fs.writeFile(target, "[]") + await expect(readFileWithMissingRetry(target)).resolves.toBe("[]") + }) + + it("preserves ENOENT after the second read when nothing can be recovered", async () => { + await expect(readFileWithMissingRetry(path.join(dir, "missing.json"))).rejects.toMatchObject({ code: "ENOENT" }) + }) + + it("propagates non-ENOENT errors", async () => { + await expect(readFileWithMissingRetry(dir)).rejects.toMatchObject({ code: "EISDIR" }) + }) +}) diff --git a/src/core/task-persistence/__tests__/taskMessages.spec.ts b/src/core/task-persistence/__tests__/taskMessages.spec.ts index 6b968fe96c..5917997931 100644 --- a/src/core/task-persistence/__tests__/taskMessages.spec.ts +++ b/src/core/task-persistence/__tests__/taskMessages.spec.ts @@ -21,6 +21,7 @@ vi.mock("../../../utils/safeWriteJson", () => ({ // Import after mocks import { saveTaskMessages, readTaskMessages } from "../taskMessages" import { formatErrorDetail } from "../readFileWithMissingRetry" +import { setPersistenceTraceSink } from "../persistenceTrace" let tmpBaseDir: string @@ -35,6 +36,7 @@ beforeEach(async () => { afterEach(() => { vi.useRealTimers() vi.restoreAllMocks() + setPersistenceTraceSink(undefined) }) describe("taskMessages.saveTaskMessages", () => { @@ -63,7 +65,11 @@ describe("taskMessages.saveTaskMessages", () => { expect(hoisted.safeWriteJsonMock).toHaveBeenCalledTimes(1) const [, persisted] = hoisted.safeWriteJsonMock.mock.calls[0] expect(persisted).toEqual(messages) - expect(hoisted.safeWriteJsonMock.mock.calls[0][2]).toBeUndefined() + const options = hoisted.safeWriteJsonMock.mock.calls[0][2] as { merge?: unknown; trace?: unknown } | undefined + expect(options?.merge).toBeUndefined() + // The trace callback is always installed so restart logs can follow + // the write stages. + expect(options?.trace).toBeTypeOf("function") }) it("persists messages without modification when no metadata", async () => { @@ -101,6 +107,39 @@ describe("taskMessages.saveTaskMessages", () => { ]) }) + it("emits persistence trace events with the task ID for the write stages", async () => { + const traceLines: string[] = [] + hoisted.safeWriteJsonMock.mockImplementationOnce(async (_path, _data, options) => { + for (const stage of ["recover", "write-start", "backup-rename", "replacement-rename", "write-complete"]) { + options.trace({ stage, detail: stage === "backup-rename" ? "eperm-fallback" : undefined }) + } + }) + setPersistenceTraceSink((line) => traceLines.push(line)) + + await saveTaskMessages({ + messages: [{ ts: 1, type: "say", say: "text", text: "traced" }], + taskId: "task-trace", + globalStoragePath: tmpBaseDir, + }) + + const events = traceLines.map( + (line) => JSON.parse(line.replace("[persistence-trace] ", "")) as Record, + ) + expect(events.map((event) => event.stage)).toEqual([ + "ui-messages-recovery", + "ui-messages-write-start", + "ui-messages-backup-rename", + "ui-messages-replacement-rename", + "ui-messages-write-complete", + ]) + for (const event of events) { + expect(event.taskId).toBe("task-trace") + expect(event.pid).toBe(process.pid) + } + const backupRename = events.find((event) => event.stage === "ui-messages-backup-rename") + expect(backupRename?.detail).toBe("eperm-fallback") + }) + it("propagates a forced write failure without swallowing or replacing it", async () => { // Fault injection: simulate an interrupted/failed final write (e.g. the // extension host quitting mid-save or the disk refusing the write). The diff --git a/src/core/task-persistence/persistenceTrace.ts b/src/core/task-persistence/persistenceTrace.ts new file mode 100644 index 0000000000..591e489494 --- /dev/null +++ b/src/core/task-persistence/persistenceTrace.ts @@ -0,0 +1,64 @@ +import type { LogFunction } from "../../utils/outputChannelLogger" + +/** + * Structured trace points for task-persistence file writes and shutdown. + * + * The restart-persistence investigation in issue #2130 could not tell which + * writer touched ui_messages.json or when, because nothing logged the + * replacement-protocol steps. These trace lines record the write stages, the + * marker check, and shutdown with the task ID and the process ID, so CI logs + * and extension logs can order writer activity across a restart. + * + * The default sink is the console. Extension activation installs a sink that + * also mirrors every line into the extension output channel; tests can + * install their own sink to capture events. Tracing must never break + * persistence, so sink failures are swallowed. + */ + +export type PersistenceTraceStage = + | "ui-messages-recovery" + | "ui-messages-write-start" + | "ui-messages-backup-rename" + | "ui-messages-replacement-rename" + | "ui-messages-write-complete" + | "ui-messages-marker-check" + | "shutdown" + | "json-recovery" + | "json-lock-wait" + +export type PersistenceTraceFields = { + taskId?: string + filePath?: string + result?: string + detail?: string +} + +type PersistenceTraceSink = LogFunction<[line: string]> + +let traceSink: PersistenceTraceSink | undefined + +/** + * Installs the sink used for every subsequent trace event. Pass undefined to + * restore the default console sink. + */ +export function setPersistenceTraceSink(sink: PersistenceTraceSink | undefined): void { + traceSink = sink +} + +export function tracePersistenceEvent(stage: PersistenceTraceStage, fields: PersistenceTraceFields = {}): void { + const line = `[persistence-trace] ${JSON.stringify({ + ts: new Date().toISOString(), + pid: process.pid, + stage, + ...fields, + })}` + try { + if (traceSink) { + traceSink(line) + } else { + console.log(line) + } + } catch { + // A failing sink must never break the write, read, or shutdown path. + } +} diff --git a/src/core/task-persistence/readFileWithMissingRetry.ts b/src/core/task-persistence/readFileWithMissingRetry.ts index 1036049470..61edfcb3aa 100644 --- a/src/core/task-persistence/readFileWithMissingRetry.ts +++ b/src/core/task-persistence/readFileWithMissingRetry.ts @@ -1,18 +1,19 @@ +import { getErrorCode } from "../../utils/errorCode" +export { getErrorCode } from "../../utils/errorCode" + import * as fs from "fs/promises" +import { recoverMissingJsonFile } from "../../utils/jsonFileRecovery" -// safeWriteJson replaces a file in two renames (target to backup, new file to -// target) because a single rename over an open file fails with EPERM on -// Windows. A reader can see ENOENT between the two renames, so retry once. +// safeWriteJson commits a replacement with a single atomic rename. Only the +// Windows EPERM fallback replaces a file in two renames (target to backup, +// new file to target), because a single rename over an open file fails with +// EPERM on Windows. A reader can see ENOENT between those two renames, so +// retry once, then recover under the writer's lock if a process crash left +// recoverable backups. Recovery may retry lock acquisition for up to 35 seconds. const READ_RETRY_MIN_MS = 1 const READ_RETRY_MAX_MS = 10 const READ_RETRY_RANGE_MS = READ_RETRY_MAX_MS - READ_RETRY_MIN_MS + 1 -export function getErrorCode(error: unknown): string | undefined { - return typeof error === "object" && error !== null && "code" in error && typeof error.code === "string" - ? error.code - : undefined -} - /** * Formats the actionable detail of an underlying error for inclusion in * wrapping error messages: the filesystem error code when present (e.g. @@ -41,6 +42,12 @@ export async function readFileWithMissingRetry(filePath: string): Promise setTimeout(resolve, retryDelay)) - return fs.readFile(filePath, "utf8") + try { + return await fs.readFile(filePath, "utf8") + } catch (retryError) { + if (getErrorCode(retryError) !== "ENOENT") throw retryError + if (!(await recoverMissingJsonFile(filePath))) throw retryError + return fs.readFile(filePath, "utf8") + } } } diff --git a/src/core/task-persistence/taskMessages.ts b/src/core/task-persistence/taskMessages.ts index b67ad2b7f8..bb89bbf4f2 100644 --- a/src/core/task-persistence/taskMessages.ts +++ b/src/core/task-persistence/taskMessages.ts @@ -2,11 +2,20 @@ import * as path from "path" import type { ClineMessage } from "@roo-code/types" -import { safeWriteJson } from "../../utils/safeWriteJson" +import { safeWriteJson, type SafeWriteJsonTraceStage } from "../../utils/safeWriteJson" import { GlobalFileNames } from "../../shared/globalFileNames" import { getTaskDirectoryPath } from "../../utils/storage" import { ensureMessageIdentifiers, mergeClineMessageSnapshots } from "./mergeMessageSnapshots" import { formatErrorDetail, getErrorCode, readFileWithMissingRetry } from "./readFileWithMissingRetry" +import { tracePersistenceEvent, type PersistenceTraceStage } from "./persistenceTrace" + +const uiMessagesTraceStages: Record = { + recover: "ui-messages-recovery", + "write-start": "ui-messages-write-start", + "backup-rename": "ui-messages-backup-rename", + "replacement-rename": "ui-messages-replacement-rename", + "write-complete": "ui-messages-write-complete", +} export type TaskMessagesReadErrorKind = "not_found" | "invalid" | "io_error" @@ -83,17 +92,21 @@ export async function saveTaskMessages({ const taskDir = await getTaskDirectoryPath(globalStoragePath, taskId) const filePath = path.join(taskDir, GlobalFileNames.uiMessages) let savedMessages = messages - await safeWriteJson( - filePath, - messages, - merge - ? { - merge: (existing, incoming) => { - savedMessages = mergeClineMessageSnapshots(existing, incoming) as ClineMessage[] - return savedMessages - }, - } - : undefined, - ) + await safeWriteJson(filePath, messages, { + ...(merge && { + merge: (existing, incoming) => { + savedMessages = mergeClineMessageSnapshots(existing, incoming) as ClineMessage[] + return savedMessages + }, + }), + // Trace the replacement-protocol stages with the task ID so restart + // logs can order UI-message writer activity across processes. + trace: (event) => + tracePersistenceEvent(uiMessagesTraceStages[event.stage], { + taskId, + filePath, + detail: event.detail, + }), + }) return savedMessages } diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index 87c37cf596..8fff156425 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -118,7 +118,6 @@ import { type DiagnosticsProviderSourceSnapshot, } from "../../services/diagnostics" -import { fileExistsAtPath } from "../../utils/fs" import { setTtsEnabled, setTtsSpeed } from "../../utils/tts" import { getWorkspaceGitInfo } from "../../utils/git" import { getWorkspacePath } from "../../utils/path" @@ -153,6 +152,7 @@ import { LifecycleTransitionError, } from "../task-persistence" import { readTaskMessages } from "../task-persistence/taskMessages" +import { readFileWithMissingRetry } from "../task-persistence/readFileWithMissingRetry" import { getNonce } from "./getNonce" import { getUri } from "./getUri" import { omitOriginalContentFromExtensionMessage } from "./stripOriginalContent" @@ -2719,21 +2719,13 @@ export class ClineProvider const taskDirPath = await getTaskDirectoryPath(globalStoragePath, id) const apiConversationHistoryFilePath = path.join(taskDirPath, GlobalFileNames.apiConversationHistory) const uiMessagesFilePath = path.join(taskDirPath, GlobalFileNames.uiMessages) - const fileExists = await fileExistsAtPath(apiConversationHistoryFilePath) - let apiConversationHistory: Anthropic.MessageParam[] = [] - if (fileExists) { - try { - apiConversationHistory = JSON.parse(await fs.readFile(apiConversationHistoryFilePath, "utf8")) - } catch (error) { - console.warn( - `[getTaskWithId] api_conversation_history.json corrupted for task ${id}, returning empty history: ${error instanceof Error ? error.message : String(error)}`, - ) - } - } else { + try { + apiConversationHistory = JSON.parse(await readFileWithMissingRetry(apiConversationHistoryFilePath)) + } catch (error) { console.warn( - `[getTaskWithId] api_conversation_history.json missing for task ${id}, returning empty history`, + `[getTaskWithId] Could not read API conversation history for task ${id}, returning empty history: ${error instanceof Error ? error.message : String(error)}`, ) } @@ -2750,11 +2742,13 @@ export class ClineProvider historyItem: HistoryItem aggregatedCosts: AggregatedCosts }> { - const { historyItem } = await this.getTaskWithId(taskId) + const historyItem = this.getTaskHistoryItem(taskId) + if (!historyItem) throw new Error("Task not found") const aggregatedCosts = await aggregateTaskCostsRecursive(taskId, async (id: string) => { - const result = await this.getTaskWithId(id) - return result.historyItem + const item = this.getTaskHistoryItem(id) + if (!item) throw new Error("Task not found") + return item }) return { historyItem, aggregatedCosts } diff --git a/src/core/webview/__tests__/ClineProvider.spec.ts b/src/core/webview/__tests__/ClineProvider.spec.ts index 9826fc206b..97a25a7038 100644 --- a/src/core/webview/__tests__/ClineProvider.spec.ts +++ b/src/core/webview/__tests__/ClineProvider.spec.ts @@ -7946,6 +7946,82 @@ describe("ClineProvider - Comprehensive Edit/Delete Edge Cases", () => { }) describe("getTaskWithId", () => { + it("retries a transiently missing API history before returning an empty conversation", async () => { + const historyItem = { + id: "retry-api-history", + task: "test task", + ts: 1, + number: 1, + tokensIn: 0, + tokensOut: 0, + totalCost: 0, + } + provider.taskHistoryStore["cache"].set(historyItem.id, historyItem) + provider["taskHistoryStoreInitialized"] = true + const fsp = await import("fs/promises") + let reads = 0 + vi.mocked(fsp.readFile).mockImplementation(async (file) => { + if (!String(file).endsWith("api_conversation_history.json")) return "" + if (++reads === 1) throw Object.assign(new Error("Replacement in progress"), { code: "ENOENT" }) + return JSON.stringify([{ role: "assistant", content: "committed" }]) + }) + + const result = await provider.getTaskWithId(historyItem.id) + + expect(result.apiConversationHistory).toEqual([{ role: "assistant", content: "committed" }]) + }) + + it.each(["ENOENT", "EACCES"])("returns empty history after %s without removing the task", async (code) => { + const item = { + id: "failed-api-read", + task: "test", + ts: 1, + number: 1, + tokensIn: 0, + tokensOut: 0, + totalCost: 0, + } + provider.taskHistoryStore["cache"].set(item.id, item) + provider["taskHistoryStoreInitialized"] = true + const fsp = await import("fs/promises") + const read = vi.mocked(fsp.readFile) + let reads = 0 + read.mockImplementation(async (file) => { + if (!String(file).endsWith("api_conversation_history.json")) return "" + reads++ + throw Object.assign(new Error("read failed"), { code }) + }) + const result = await provider.getTaskWithId(item.id) + expect(result.apiConversationHistory).toEqual([]) + expect(provider.taskHistoryStore.get(item.id)).toEqual(item) + expect(reads).toBe(code === "ENOENT" ? 2 : 1) + }) + + it("aggregates parent and child costs without reading API history", async () => { + const parent = { + id: "cost-parent", + task: "test", + ts: 1, + number: 1, + tokensIn: 0, + tokensOut: 0, + totalCost: 2, + childIds: ["cost-child"], + } + provider.taskHistoryStore["cache"].set(parent.id, parent) + provider.taskHistoryStore["cache"].set("cost-child", { + ...parent, + id: "cost-child", + totalCost: 3, + childIds: [], + }) + provider["taskHistoryStoreInitialized"] = true + const read = vi.spyOn(provider, "getTaskWithId") + const result = await provider.getTaskWithAggregatedCosts(parent.id) + expect(result.aggregatedCosts.totalCost).toBe(5) + expect(read).not.toHaveBeenCalled() + }) + it("does not restore a deleted file-backed task from legacy history", async () => { const historyItem = { id: "deleted-task", diff --git a/src/extension.ts b/src/extension.ts index 6ce42aacd5..34865230c4 100644 --- a/src/extension.ts +++ b/src/extension.ts @@ -27,6 +27,7 @@ import { initializeNetworkProxy } from "./utils/networkProxy" import { Package } from "./shared/package" import { formatLanguage } from "./shared/language" import { ContextProxy } from "./core/config/ContextProxy" +import { setPersistenceTraceSink, tracePersistenceEvent } from "./core/task-persistence/persistenceTrace" import { ClineProvider } from "./core/webview/ClineProvider" import { registerCopilotModelBroadcast } from "./services/language-models/copilotModelBroadcast" import { ClineProviderFactory } from "./core/webview/ClineProviderFactory" @@ -125,6 +126,10 @@ export async function activate(context: vscode.ExtensionContext) { extensionContext = context outputChannel = vscode.window.createOutputChannel(Package.outputChannel) context.subscriptions.push(outputChannel) + // Mirror persistence trace lines (task ID + process ID) into the output + // channel and the console, so a restart-persistence investigation can + // order UI-message writer activity across extension-host restarts. + setPersistenceTraceSink(createDualLogger(createOutputChannelLogger(outputChannel))) outputChannel.appendLine(`${Package.name} extension activated - ${JSON.stringify(Package)}`) // Initialize network proxy configuration early, before any network requests. @@ -402,6 +407,7 @@ export async function activate(context: vscode.ExtensionContext) { // This method is called when your extension is deactivated. export async function deactivate() { + tracePersistenceEvent("shutdown") outputChannel.appendLine(`${Package.name} extension deactivated`) if (cloudService && CloudService.hasInstance()) { diff --git a/src/extension/__tests__/api-has-task-ui-message-text.spec.ts b/src/extension/__tests__/api-has-task-ui-message-text.spec.ts index 697a2e0809..931ce3d6d4 100644 --- a/src/extension/__tests__/api-has-task-ui-message-text.spec.ts +++ b/src/extension/__tests__/api-has-task-ui-message-text.spec.ts @@ -5,6 +5,7 @@ import { API } from "../api" import { ClineProvider } from "../../core/webview/ClineProvider" import { readTaskMessages, TaskMessagesReadError } from "../../core/task-persistence/taskMessages" import { makeClineProviderFactory } from "../../test-utils/provider" +import { setPersistenceTraceSink } from "../../core/task-persistence/persistenceTrace" vi.mock("vscode") vi.mock("../../core/webview/ClineProvider") @@ -73,9 +74,19 @@ describe("API#hasTaskUiMessageText", () => { "returns false instead of throwing on a %s read failure", async (kind) => { // The completion write may not have landed yet; callers poll until it does. - vi.mocked(readTaskMessages).mockRejectedValue(new TaskMessagesReadError(kind, `read failed: ${kind}`)) - - await expect(api.hasTaskUiMessageText("task-1", "RESTART_PERSISTENCE_MARKER")).resolves.toBe(false) + const lines: string[] = [] + setPersistenceTraceSink((line) => lines.push(line)) + vi.mocked(readTaskMessages).mockRejectedValue( + new TaskMessagesReadError(kind, "private content", { code: "EIO" }), + ) + try { + await expect(api.hasTaskUiMessageText("task-1", "RESTART_PERSISTENCE_MARKER")).resolves.toBe(false) + expect(lines).toHaveLength(1) + expect(lines[0]).toContain(`"detail":"${kind}:EIO"`) + expect(lines[0]).not.toContain("private content") + } finally { + setPersistenceTraceSink(undefined) + } }, ) diff --git a/src/extension/api.ts b/src/extension/api.ts index 623b0492d7..606bb7725d 100644 --- a/src/extension/api.ts +++ b/src/extension/api.ts @@ -33,7 +33,9 @@ import type { Mode } from "../shared/modes" import type { ClineProvider } from "../core/webview/ClineProvider" import type { ClineProviderFactory } from "../core/webview/ClineProviderFactory" import { withoutInvalidDisplaySettings } from "../core/config/displaySettings" -import { readTaskMessages } from "../core/task-persistence/taskMessages" +import { readTaskMessages, TaskMessagesReadError } from "../core/task-persistence/taskMessages" +import { getErrorCode } from "../core/task-persistence/readFileWithMissingRetry" +import { tracePersistenceEvent } from "../core/task-persistence/persistenceTrace" import { Terminal } from "../integrations/terminal/Terminal" import { TerminalRegistry } from "../integrations/terminal/TerminalRegistry" import { getCommands } from "../services/command/commands" @@ -361,10 +363,20 @@ export class API extends EventEmitter implements RooCodeAPI { taskId, globalStoragePath: this.sidebarProvider.contextProxy.globalStorageUri.fsPath, }) - } catch { + } catch (error: unknown) { + tracePersistenceEvent("ui-messages-marker-check", { + taskId, + result: "read-error", + detail: + error instanceof TaskMessagesReadError + ? `${error.kind}:${getErrorCode(error.originalError) ?? "unknown"}` + : (getErrorCode(error) ?? "unknown"), + }) return false } - return uiMessages.some((message) => message.text?.includes(text)) + const hasText = uiMessages.some((message) => message.text?.includes(text)) + tracePersistenceEvent("ui-messages-marker-check", { taskId, result: hasText ? "match" : "no-match" }) + return hasText } public getCurrentTaskStack() { diff --git a/src/utils/__tests__/fileLock.test.ts b/src/utils/__tests__/fileLock.test.ts new file mode 100644 index 0000000000..92637fe7a0 --- /dev/null +++ b/src/utils/__tests__/fileLock.test.ts @@ -0,0 +1,34 @@ +import * as lockfile from "proper-lockfile" +import { acquireFileLock, LOCK_STALE_MS, withFileLock } from "../fileLock" + +vi.mock("proper-lockfile", () => ({ lock: vi.fn() })) + +describe("file lock protocol", () => { + beforeEach(() => vi.clearAllMocks()) + + it("allows recovery to outwait staleness with bounded constant retries", async () => { + vi.mocked(lockfile.lock).mockResolvedValue(async () => {}) + await acquireFileLock("history.json", { waitForStaleLock: true }) + expect(lockfile.lock).toHaveBeenCalledWith( + expect.any(String), + expect.objectContaining({ + stale: LOCK_STALE_MS, + update: 10000, + realpath: false, + retries: { retries: 35, factor: 1, minTimeout: 1000, maxTimeout: 1000 }, + }), + ) + }) + + it("releases the lock when the operation fails", async () => { + const release = vi.fn(async () => {}) + vi.mocked(lockfile.lock).mockResolvedValue(release) + const error = new Error("operation failed") + await expect( + withFileLock("history.json", async () => { + throw error + }), + ).rejects.toBe(error) + expect(release).toHaveBeenCalledOnce() + }) +}) diff --git a/src/utils/__tests__/jsonFileRecovery.test.ts b/src/utils/__tests__/jsonFileRecovery.test.ts new file mode 100644 index 0000000000..1bd90d8935 --- /dev/null +++ b/src/utils/__tests__/jsonFileRecovery.test.ts @@ -0,0 +1,109 @@ +import * as fs from "fs/promises" +import * as os from "os" +import * as path from "path" + +import { acquireFileLock } from "../fileLock" +import { setPersistenceTraceSink } from "../../core/task-persistence/persistenceTrace" +import { deleteJsonFile, recoverMissingJsonFile } from "../jsonFileRecovery" +import { readFileWithMissingRetry } from "../../core/task-persistence/readFileWithMissingRetry" + +vi.mock("fs/promises", async () => { + const actual = await vi.importActual("fs/promises") + return { ...actual, readdir: vi.fn(actual.readdir), rename: vi.fn(actual.rename), unlink: vi.fn(actual.unlink) } +}) + +describe("reader recovery", () => { + let dir: string + let target: string + let backup: string + beforeEach(async () => { + dir = await fs.mkdtemp(path.join(os.tmpdir(), "json-reader-recovery-")) + target = path.join(dir, "history_item.json") + backup = path.join(dir, ".history_item.json.bak_1_saved.tmp") + await fs.writeFile(backup, JSON.stringify({ committed: true })) + }) + afterEach(async () => { + setPersistenceTraceSink(undefined) + vi.restoreAllMocks() + await fs.rm(dir, { recursive: true, force: true }) + }) + + it.each(["hint-scan", "locked-scan", "restore"] as const)( + "preserves history after a failed %s and recovers on retry", + async (step) => { + const sink = vi.fn() + setPersistenceTraceSink(sink) + const error = Object.assign(new Error("Private filesystem detail"), { code: "EIO" }) + if (step === "restore") vi.mocked(fs.rename).mockRejectedValueOnce(error) + else { + if (step === "locked-scan") { + const actual = await vi.importActual("fs/promises") + vi.mocked(fs.readdir).mockImplementationOnce(actual.readdir) + } + vi.mocked(fs.readdir).mockRejectedValueOnce(error) + } + await expect(readFileWithMissingRetry(target)).rejects.toBe(error) + expect(sink).toHaveBeenCalledWith(expect.stringContaining('"result":"failed"')) + expect(JSON.parse(await fs.readFile(backup, "utf8"))).toEqual({ committed: true }) + await expect(fs.readFile(target)).rejects.toMatchObject({ code: "ENOENT" }) + expect(JSON.parse(await readFileWithMissingRetry(target))).toEqual({ committed: true }) + }, + ) + + it("rechecks the canonical file under lock and never replaces a newer commit with a stale backup", async () => { + await fs.writeFile(target, JSON.stringify({ newest: true })) + await recoverMissingJsonFile(target) + expect(JSON.parse(await fs.readFile(target, "utf8"))).toEqual({ newest: true }) + await expect(fs.readFile(backup)).rejects.toMatchObject({ code: "ENOENT" }) + }) + + it("traces reader restoration and lock waiting", async () => { + const sink = vi.fn() + setPersistenceTraceSink(sink) + await readFileWithMissingRetry(target) + expect(sink).toHaveBeenCalledWith(expect.stringContaining('"stage":"json-lock-wait"')) + expect(sink).toHaveBeenCalledWith(expect.stringContaining('"result":"restored"')) + }) + + it("does not steal a live writer lock during startup recovery", async () => { + const release = await acquireFileLock(target) + try { + await expect(recoverMissingJsonFile(target, { waitForStaleLock: false })).rejects.toMatchObject({ + code: "ELOCKED", + }) + expect(await fs.readFile(backup, "utf8")).toContain("committed") + await expect(fs.access(target)).rejects.toMatchObject({ code: "ENOENT" }) + } finally { + await release() + } + await expect(recoverMissingJsonFile(target)).resolves.toBe(true) + }) + + it("returns false when only an uncommitted replacement exists", async () => { + await fs.rename(backup, path.join(dir, ".history_item.json.new_1.tmp")) + await expect(recoverMissingJsonFile(target)).resolves.toBe(false) + }) + + it("deletes backups before the canonical file and prevents resurrection", async () => { + await fs.writeFile(target, "{}") + await deleteJsonFile(target) + await expect(recoverMissingJsonFile(target)).resolves.toBe(false) + await expect(fs.access(target)).rejects.toMatchObject({ code: "ENOENT" }) + await expect(deleteJsonFile(target)).resolves.toBeUndefined() + }) + + it("preserves the canonical file when deleting a backup fails", async () => { + await fs.writeFile(target, "committed") + const error = Object.assign(new Error("unlink failed"), { code: "EACCES" }) + vi.mocked(fs.unlink).mockRejectedValueOnce(error) + await expect(deleteJsonFile(target)).rejects.toBe(error) + expect(await fs.readFile(target, "utf8")).toBe("committed") + await expect(fs.access(backup)).resolves.toBeUndefined() + }) + + it("does not create a missing task directory", async () => { + await fs.rm(dir, { recursive: true }) + await expect(recoverMissingJsonFile(target)).resolves.toBe(false) + await expect(fs.stat(dir)).rejects.toMatchObject({ code: "ENOENT" }) + }) +}) diff --git a/src/utils/__tests__/safeWriteJson.interruption.test.ts b/src/utils/__tests__/safeWriteJson.interruption.test.ts new file mode 100644 index 0000000000..c9c5993f84 --- /dev/null +++ b/src/utils/__tests__/safeWriteJson.interruption.test.ts @@ -0,0 +1,242 @@ +import { spawn, type ChildProcess } from "child_process" +import { createRequire } from "module" +import { fileURLToPath } from "url" +import * as fs from "fs/promises" +import * as os from "os" +import * as path from "path" + +import { safeWriteJson } from "../safeWriteJson" +import { readFileWithMissingRetry } from "../../core/task-persistence/readFileWithMissingRetry" +import { TaskHistoryStore } from "../../core/task-persistence/TaskHistoryStore" +import { readTaskMessages } from "../../core/task-persistence/taskMessages" +import { readApiMessages } from "../../core/task-persistence/apiMessages" + +vi.mock("../storage", () => ({ + getStorageBasePath: (defaultPath: string) => defaultPath, + getTaskDirectoryPath: async (base: string, taskId: string) => path.join(base, "tasks", taskId), +})) + +const requireFromTest = createRequire(import.meta.url) +const esbuildPath = requireFromTest.resolve("esbuild") +const safeWriteJsonEntryPath = fileURLToPath(new URL("../safeWriteJson.ts", import.meta.url)) + +// The child performs a real safeWriteJson from this repository. The source is +// bundled to CommonJS at child startup, so the bundle's require("fs/promises") +// resolves to the same exports object the script patched above, and the +// patched rename is the one the writer actually calls. +const CHILD_SCRIPT = `"use strict" +const fs = require("fs") +const path = require("path") + +const targetPath = path.resolve(process.env.INTERRUPTION_TARGET_PATH) +const fsPromises = require("fs/promises") +const actualRename = fsPromises.rename + +// Fault injection: hard-kill before the replacement lands. Any rename +// whose destination is the canonical path is the commit step, whatever +// replacement protocol the writer uses; exit before performing it. +let commitAttempts = 0 +fsPromises.rename = async function interruptionRename(oldPath, newPath) { + if (path.resolve(String(newPath)) === targetPath) { + commitAttempts++ + if (process.env.INTERRUPTION_MODE === "fallback" && commitAttempts === 1) { + return Promise.reject(Object.assign(new Error("Injected EPERM"), { code: "EPERM" })) + } + if (process.env.INTERRUPTION_BOUNDARY === "after-commit") { + await actualRename.apply(fsPromises, arguments) + } + fs.writeSync(1, "EXIT_AT_REPLACEMENT_RENAME\\n") + process.kill(process.pid, "SIGKILL") + } + return actualRename.apply(fsPromises, arguments) +} + +require(process.env.ESBUILD_PATH).buildSync({ + entryPoints: [process.env.SAFE_WRITE_JSON_ENTRY], + bundle: true, + platform: "node", + format: "cjs", + target: "node22", + outfile: process.env.INTERRUPTION_BUNDLE_PATH, + logLevel: "silent", +}) + +const payload = JSON.parse(process.env.INTERRUPTION_PAYLOAD) +require(process.env.INTERRUPTION_BUNDLE_PATH) + .safeWriteJson(targetPath, payload) + .then(() => { + fs.writeSync(1, "WRITE_COMPLETED\\n") + process.exit(0) + }) + .catch((error) => { + fs.writeSync(2, "WRITE_FAILED: " + ((error && error.message) || String(error)) + "\\n") + process.exit(1) + }) +` + +async function waitForChildExit(child: ChildProcess): Promise<{ code: number | null; stdout: string; stderr: string }> { + const stdoutChunks: Buffer[] = [] + const stderrChunks: Buffer[] = [] + child.stdout?.on("data", (chunk: Buffer) => stdoutChunks.push(chunk)) + child.stderr?.on("data", (chunk: Buffer) => stderrChunks.push(chunk)) + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + child.kill("SIGKILL") + reject(new Error("Interruption child did not exit within 20 seconds")) + }, 20_000) + child.once("error", reject) + child.once("close", (code) => { + clearTimeout(timeout) + resolve({ + code, + stdout: Buffer.concat(stdoutChunks).toString("utf8"), + stderr: Buffer.concat(stderrChunks).toString("utf8"), + }) + }) + }) +} + +describe("safeWriteJson subprocess interruption fault injection", () => { + let tempDir: string + let storageDir: string + let childDir: string + let targetPath: string + let childScriptPath: string + let childBundlePath: string + + beforeEach(async () => { + tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "safe-write-interruption-")) + storageDir = path.join(tempDir, "tasks", "restart-task") + childDir = path.join(tempDir, "child") + await fs.mkdir(storageDir, { recursive: true }) + await fs.mkdir(childDir) + targetPath = path.join(storageDir, "ui_messages.json") + childScriptPath = path.join(childDir, "interrupted-writer.cjs") + childBundlePath = path.join(childDir, "safeWriteJson.bundle.cjs") + await fs.writeFile(childScriptPath, CHILD_SCRIPT, "utf8") + }) + + afterEach(async () => { + await fs.rm(tempDir, { recursive: true, force: true }) + }) + + it.each([ + ...(["ui_messages.json", "api_conversation_history.json", "history_item.json"] as const).flatMap((fileName) => + (["atomic", "fallback"] as const).flatMap((mode) => + (["before-commit", "after-commit"] as const).map((boundary) => ({ + fileName, + mode, + boundary, + ageLock: true, + })), + ), + ), + ])( + "recovers $fileName after $mode interruption $boundary without another write (age lock: $ageLock)", + async ({ fileName, mode, boundary, ageLock }) => { + targetPath = path.join(storageDir, fileName) + // A previously committed write contains the restart marker. + const historyItem = { + id: "restart-task", + number: 1, + ts: 1, + task: "RESTART_PERSISTENCE_MARKER", + tokensIn: 0, + tokensOut: 0, + totalCost: 0, + status: "completed", + } + const message = { + ts: 1, + type: "say", + say: "text", + text: "RESTART_PERSISTENCE_MARKER", + role: "assistant", + content: "RESTART_PERSISTENCE_MARKER", + } + const committed = fileName === "history_item.json" ? historyItem : [message] + const replacement = + fileName === "history_item.json" + ? { ...historyItem, ts: 2, task: `INTERRUPTED_WRITE_${mode}` } + : [{ ...message, ts: 2, text: `INTERRUPTED_WRITE_${mode}`, content: `INTERRUPTED_WRITE_${mode}` }] + await safeWriteJson(targetPath, committed) + + // The condition introduced by the previous fix: the committed + // content is readable before the interrupted write starts. + const before = await readFileWithMissingRetry(targetPath) + expect(before).toContain("RESTART_PERSISTENCE_MARKER") + + // A real child process writes the same path and terminates right + // before the replacement rename, mirroring the reported crash + // window between the two renames of the old protocol. + const child = spawn(process.execPath, [childScriptPath], { + env: { + ...process.env, + INTERRUPTION_TARGET_PATH: targetPath, + INTERRUPTION_MODE: mode, + INTERRUPTION_BOUNDARY: boundary, + INTERRUPTION_PAYLOAD: JSON.stringify(replacement), + ESBUILD_PATH: esbuildPath, + SAFE_WRITE_JSON_ENTRY: safeWriteJsonEntryPath, + INTERRUPTION_BUNDLE_PATH: childBundlePath, + }, + stdio: ["ignore", "pipe", "pipe"], + }) + const exit = await waitForChildExit(child) + + expect(exit.stdout).toContain("EXIT_AT_REPLACEMENT_RENAME") + expect(exit.stdout).not.toContain("WRITE_COMPLETED") + expect(exit.stderr).not.toContain("WRITE_FAILED") + // Windows emulates SIGKILL with exit code 1; POSIX reports a signal exit with no code. + expect(exit.code).toBe(process.platform === "win32" ? 1 : null) + + // Atomic replacement retains the canonical file; the two-step + // fallback retains committed content in its backup until recovery. + const interruptedFiles = await fs.readdir(storageDir) + expect(interruptedFiles.filter((entry) => entry.includes(".new_"))).toHaveLength( + boundary === "before-commit" ? 1 : 0, + ) + if (mode === "fallback" && boundary === "before-commit") { + await expect(fs.readFile(targetPath)).rejects.toMatchObject({ code: "ENOENT" }) + const backups = interruptedFiles.filter((entry) => entry.includes(".bak_")) + expect(backups).toHaveLength(1) + expect(JSON.parse(await fs.readFile(path.join(storageDir, backups[0]), "utf8"))).toEqual(committed) + } + + // SIGKILL leaves the lock behind. Age it to exercise real stale-lock + // acquisition without making the test wait through the stale window. + const staleTime = new Date(Date.now() - 60_000) + if (ageLock) await fs.utimes(`${targetPath}.lock`, staleTime, staleTime) + const expected = boundary === "before-commit" ? committed : replacement + // Fresh startup must discover a missing history record before any writer runs. + if (fileName === "history_item.json") { + const store = new TaskHistoryStore(tempDir) + try { + await store.initialize() + expect(store.get("restart-task")).toEqual(expected) + } finally { + store.dispose() + } + } else if (fileName === "ui_messages.json") { + expect(await readTaskMessages({ taskId: "restart-task", globalStoragePath: tempDir })).toEqual(expected) + } else { + expect(await readApiMessages({ taskId: "restart-task", globalStoragePath: tempDir })).toEqual(expected) + } + expect(JSON.parse(await readFileWithMissingRetry(targetPath))).toEqual(expected) + const recovered = { ui_messages: [{ ts: 3, type: "say", text: "RECOVERED_AFTER_RESTART" }] } + await safeWriteJson(targetPath, recovered, { + merge: (existing, incoming) => { + expect(existing).toEqual(expected) + return incoming + }, + }) + + expect(JSON.parse(await readFileWithMissingRetry(targetPath))).toEqual(recovered) + const leftovers = (await fs.readdir(storageDir)).filter( + (entry) => entry.includes(".new_") || entry.includes(".bak_"), + ) + expect(leftovers).toEqual([]) + }, + 45_000, + ) +}) diff --git a/src/utils/__tests__/safeWriteJson.test.ts b/src/utils/__tests__/safeWriteJson.test.ts index 84cb4d2cf6..30ac500212 100644 --- a/src/utils/__tests__/safeWriteJson.test.ts +++ b/src/utils/__tests__/safeWriteJson.test.ts @@ -27,12 +27,14 @@ vi.mock("fs/promises", async () => { rename: vi.fn(actual.rename), unlink: vi.fn(actual.unlink), access: vi.fn(actual.access), + stat: vi.fn(actual.stat), + open: vi.fn(actual.open), mkdtemp: vi.fn(actual.mkdtemp), rm: vi.fn(actual.rm), readdir: vi.fn(actual.readdir), mkdir: vi.fn(actual.mkdir), } - // fs.stat and fs.lstat are available via { ...actual } + // fs.lstat is available via { ...actual } return mockedFs }) @@ -160,26 +162,33 @@ describe("safeWriteJson", () => { expect(content).toEqual({ initial: "content" }) }) - test("should handle failure when renaming filePath to tempBackupFilePath (filePath exists)", async () => { - const initialData = { message: "Initial content, should remain" } - const newData = { message: "New content, should not be written" } + test("should fall back to the two-step replacement when the commit rename fails with EPERM", async () => { + const initialData = { message: "Initial content" } + const newData = { message: "Windows-style EPERM replacement" } // Overwrite the pre-created file with specific initial data await fsPromisesActuals.writeFile!(currentTestFilePath, JSON.stringify(initialData)) // fs.rename is already vi.fn() — use vi.mocked to avoid double-wrapping via vi.spyOn + // First call: the single-rename commit fails with the Windows-style EPERM + // (a reader holds the target open). The fallback then renames the target + // to a backup and the replacement into place, using real renames. vi.mocked(fs.rename).mockImplementationOnce(async () => { - throw new Error("Rename to backup failed") + throw Object.assign(new Error("EPERM: operation not permitted"), { code: "EPERM" }) }) - await expect(safeWriteJson(currentTestFilePath, newData)).rejects.toThrow("Rename to backup failed") + await safeWriteJson(currentTestFilePath, newData) - // Verify the original file still exists with initial content + // Verify the new content was committed by the fallback const content = await readFileContent(currentTestFilePath) - expect(content).toEqual(initialData) + expect(content).toEqual(newData) + + // Verify the fallback backup was cleaned up + const leftovers = (await fs.readdir(tempDir)).filter((entry) => entry.includes(".bak_")) + expect(leftovers).toEqual([]) }) - test("should handle failure when renaming tempNewFilePath to filePath (filePath exists, backup succeeded)", async () => { + test("should restore the committed file when the EPERM fallback's replacement rename fails", async () => { const initialData = { message: "Initial content, should be restored" } const newData = { message: "New content" } @@ -193,20 +202,21 @@ describe("safeWriteJson", () => { vi.mocked(fs.rename).mockImplementation(async (oldPath, newPath) => { renameCallCount++ if (renameCallCount === 1) { - // First call: filePath -> tempBackupFilePath (should succeed) - return fsPromisesActuals.rename!(oldPath, newPath) + // First call: the single-rename commit fails with EPERM, which + // triggers the two-step fallback. + throw Object.assign(new Error("EPERM: operation not permitted"), { code: "EPERM" }) } else if (renameCallCount === 2) { - // Second call: tempNewFilePath -> filePath (should fail) - throw new Error("Rename from temp to final failed") - } else if (renameCallCount === 3) { - // Third call: tempBackupFilePath -> filePath (rollback, should succeed) + // Second call: filePath -> tempBackupFilePath (fallback backup, succeeds) return fsPromisesActuals.rename!(oldPath, newPath) + } else if (renameCallCount === 3) { + // Third call: tempNewFilePath -> filePath (fallback replacement, fails) + throw new Error("Fallback replacement rename failed") } - // Default: use original implementation + // Fourth call: tempBackupFilePath -> filePath (rollback, succeeds) return fsPromisesActuals.rename!(oldPath, newPath) }) - await expect(safeWriteJson(currentTestFilePath, newData)).rejects.toThrow("Rename from temp to final failed") + await expect(safeWriteJson(currentTestFilePath, newData)).rejects.toThrow("Fallback replacement rename failed") // Verify the file was restored to initial content const content = await readFileContent(currentTestFilePath) @@ -293,13 +303,18 @@ describe("safeWriteJson", () => { expect(content).toEqual(data) }) - test("should handle failure when deleting tempBackupFilePath (filePath exists, all renames succeed)", async () => { + test("should handle failure when deleting tempBackupFilePath (EPERM fallback, all renames succeed)", async () => { const initialData = { message: "Initial content" } const newData = { message: "Successfully written new content" } // Overwrite the pre-created file with specific initial data await fsPromisesActuals.writeFile!(currentTestFilePath, JSON.stringify(initialData)) + // Force the two-step fallback so a backup exists, then let its deletion fail + vi.mocked(fs.rename).mockImplementationOnce(async () => { + throw Object.assign(new Error("EPERM: operation not permitted"), { code: "EPERM" }) + }) + // fs.unlink is already vi.fn() — use vi.mocked to avoid double-wrapping via vi.spyOn vi.mocked(fs.unlink).mockImplementationOnce(async () => { throw new Error("Failed to delete backup file") @@ -321,6 +336,11 @@ describe("safeWriteJson", () => { await fsPromisesActuals.writeFile!(currentTestFilePath, JSON.stringify(initialData)) + // Force the two-step fallback so a backup exists, then fail its deletion + vi.mocked(fs.rename).mockImplementationOnce(async () => { + throw Object.assign(new Error("EPERM: operation not permitted"), { code: "EPERM" }) + }) + // fs.unlink is already vi.fn() — use vi.mocked to avoid double-wrapping via vi.spyOn vi.mocked(fs.unlink).mockImplementation(async (filePath) => { if (filePath.toString().includes(".bak_")) { @@ -339,7 +359,7 @@ describe("safeWriteJson", () => { }) // The expected error message might need to change if the mock behaves differently. - test("should handle failure when renaming tempNewFilePath to filePath (filePath initially exists)", async () => { + test("should keep the committed file when the commit rename fails with a non-EPERM error", async () => { // currentTestFilePath exists due to beforeEach. const initialData = { message: "Initial content" } const newData = { message: "New content" } @@ -347,20 +367,14 @@ describe("safeWriteJson", () => { await fsPromisesActuals.writeFile!(currentTestFilePath, JSON.stringify(initialData)) // fs.rename is already vi.fn() — use vi.mocked to avoid double-wrapping via vi.spyOn - let renameCallCount = 0 - vi.mocked(fs.rename).mockImplementation(async (oldPath, newPath) => { - renameCallCount++ - if (renameCallCount === 2) { - // Second call: tempNewFilePath -> filePath (should fail) - throw new Error("Rename failed") - } - // For all other calls, use the original implementation - return fsPromisesActuals.rename!(oldPath, newPath) + vi.mocked(fs.rename).mockImplementationOnce(async () => { + throw new Error("Rename failed") }) await expect(safeWriteJson(currentTestFilePath, newData)).rejects.toThrow("Rename failed") - // The file should be restored to its initial content + // No backup is made on the single-rename path, and the committed file + // keeps its previous content const content = await readFileContent(currentTestFilePath) expect(content).toEqual(initialData) }) @@ -425,7 +439,7 @@ describe("safeWriteJson", () => { // Test for rollback failure scenario test("should log error and re-throw original if rollback fails", async () => { - const initialData = { message: "Initial, should be lost if rollback fails" } + const initialData = { message: "Initial, preserved if rollback fails" } const newData = { message: "New content" } await fsPromisesActuals.writeFile!(currentTestFilePath, JSON.stringify(initialData)) @@ -436,18 +450,36 @@ describe("safeWriteJson", () => { let renameCallCount = 0 vi.mocked(fs.rename).mockImplementation(async (oldPath, newPath) => { renameCallCount++ - if (renameCallCount === 2) { - // Second call: tempNewFilePath -> filePath (fail) - throw new Error("Primary rename failed") + if (renameCallCount === 1) { + // First call: the single-rename commit fails with EPERM, which + // triggers the two-step fallback. + throw Object.assign(new Error("Primary rename failed"), { code: "EPERM" }) + } else if (renameCallCount === 2) { + // Second call: filePath -> tempBackupFilePath (fallback backup, succeeds) + return fsPromisesActuals.rename!(oldPath, newPath) } else if (renameCallCount === 3) { - // Third call: tempBackupFilePath -> filePath (rollback, also fail) + // Third call: tempNewFilePath -> filePath (fallback replacement, fails) + throw new Error("Fallback replacement rename failed") + } else if (renameCallCount === 4) { + // Fourth call: tempBackupFilePath -> filePath (rollback, also fails) throw new Error("Rollback rename failed") } return fsPromisesActuals.rename!(oldPath, newPath) }) - // Should throw the original error, not the rollback error - await expect(safeWriteJson(currentTestFilePath, newData)).rejects.toThrow("Primary rename failed") + // The fallback's replacement error rejects the write; the rollback error + // is only logged. + await expect(safeWriteJson(currentTestFilePath, newData)).rejects.toThrow("Fallback replacement rename failed") + + const backups = (await fs.readdir(tempDir)).filter((entry) => entry.includes(".bak_")) + expect(backups).toHaveLength(1) + expect(await readFileContent(path.join(tempDir, backups[0]))).toEqual(initialData) + await safeWriteJson(currentTestFilePath, newData, { + merge: (existing) => { + expect(existing).toEqual(initialData) + return newData + }, + }) // Verify console.error was called for the rollback failure expect(consoleErrorSpy).toHaveBeenCalledWith( @@ -531,4 +563,259 @@ describe("safeWriteJson", () => { const content = await readFileContent(currentTestFilePath) expect(content).toEqual({ c: 3 }) }) + + // Interrupted-write recovery tests + it.each(["EMFILE", "EIO"])( + "should preserve committed history when the recovery scan fails with %s", + async (code) => { + await fs.unlink(currentTestFilePath) + const backupPath = path.join(tempDir, `.${path.basename(currentTestFilePath)}.bak_1_saved.tmp`) + const committed = { history: ["saved"] } + await fs.writeFile(backupPath, JSON.stringify(committed)) + const error = Object.assign(new Error("Recovery scan failed"), { code }) + vi.mocked(fs.readdir).mockRejectedValueOnce(error) + const merge = vi.fn((existing: unknown) => existing) + const trace = vi.fn() + + await expect(safeWriteJson(currentTestFilePath, {}, { merge, trace })).rejects.toBe(error) + + expect(merge).not.toHaveBeenCalled() + expect(trace).toHaveBeenCalledExactlyOnceWith({ stage: "recover", detail: `failed:${code}` }) + expect(await readFileContent(backupPath)).toEqual(committed) + await expect(fs.readFile(currentTestFilePath)).rejects.toMatchObject({ code: "ENOENT" }) + // Once the fault clears, a later merge still sees the committed state. + await safeWriteJson(currentTestFilePath, {}, { merge }) + expect(merge).toHaveBeenCalledExactlyOnceWith(committed, {}) + expect(await readFileContent(currentTestFilePath)).toEqual(committed) + }, + ) + + test("should allow a write when the recovery directory scan reports ENOENT", async () => { + vi.mocked(fs.readdir).mockRejectedValueOnce(Object.assign(new Error("Directory missing"), { code: "ENOENT" })) + await safeWriteJson(currentTestFilePath, { saved: true }) + expect(await readFileContent(currentTestFilePath)).toEqual({ saved: true }) + }) + + test("should restore an orphaned backup when the canonical path is missing after an interrupted write", async () => { + const committedData = { message: "Committed before the interruption" } + + // Simulate the crash state of the two-step replacement: the target was + // renamed to a backup, the replacement never landed, and the + // replacement temp file was left behind. The canonical path is absent. + await fsPromisesActuals.unlink!(currentTestFilePath) + const backupPath = path.join(tempDir, `.${path.basename(currentTestFilePath)}.bak_1_abc.tmp`) + const orphanedNewPath = path.join(tempDir, `.${path.basename(currentTestFilePath)}.new_1_def.tmp`) + await fsPromisesActuals.writeFile!(backupPath, JSON.stringify(committedData)) + await fsPromisesActuals.writeFile!(orphanedNewPath, "{}") + + // A merge write must observe the restored committed content, not null. + const incoming = { message: "New content" } + let mergedWith: unknown + await safeWriteJson(currentTestFilePath, incoming, { + merge: (existing) => { + mergedWith = existing + return incoming + }, + }) + + expect(mergedWith).toEqual(committedData) + const content = await readFileContent(currentTestFilePath) + expect(content).toEqual(incoming) + + // The restored backup is consumed and the leftover replacement file is + // garbage collected. + const leftovers = (await fs.readdir(tempDir)).filter( + (entry) => entry.includes(".new_") || entry.includes(".bak_"), + ) + expect(leftovers).toEqual([]) + }) + + it.each(["EACCES", "EIO", "EPERM", undefined])( + "should preserve committed content when canonical access fails with %s", + async (code) => { + const backupPath = path.join(tempDir, `.${path.basename(currentTestFilePath)}.bak_1_stale.tmp`) + await fs.writeFile(backupPath, JSON.stringify({ stale: true })) + // Directory access succeeds; only the canonical-path check fails. + const error = Object.assign(new Error("Canonical access failed"), { code }) + vi.mocked(fs.access).mockResolvedValueOnce(undefined).mockRejectedValueOnce(error) + const merge = vi.fn((existing: unknown) => existing) + const trace = vi.fn() + + await expect(safeWriteJson(currentTestFilePath, {}, { merge, trace })).rejects.toBe(error) + + expect(trace).toHaveBeenCalledExactlyOnceWith({ stage: "recover", detail: `failed:${code ?? "unknown"}` }) + expect(merge).not.toHaveBeenCalled() + expect(await readFileContent(currentTestFilePath)).toEqual({ initial: "content" }) + expect(await readFileContent(backupPath)).toEqual({ stale: true }) + }, + ) + + it.each(["ENOENT", "EACCES", "EIO"])( + "should preserve backups when another backup stat fails with %s", + async (code) => { + await fs.unlink(currentTestFilePath) + const unreadableBackup = path.join(tempDir, `.${path.basename(currentTestFilePath)}.bak_1_bad.tmp`) + const readableBackup = path.join(tempDir, `.${path.basename(currentTestFilePath)}.bak_2_good.tmp`) + await fs.writeFile(unreadableBackup, JSON.stringify({ stale: true })) + await fs.writeFile(readableBackup, JSON.stringify({ committed: true })) + const actual = await vi.importActual("fs/promises") + vi.mocked(fs.stat).mockImplementation(async (backupPath) => { + if (backupPath === unreadableBackup) { + throw Object.assign(new Error("Backup stat failed"), { code }) + } + return actual.stat(backupPath) + }) + const merge = vi.fn((existing: unknown) => existing) + + if (code === "ENOENT") { + await safeWriteJson(currentTestFilePath, {}, { merge }) + expect(merge).toHaveBeenCalledWith({ committed: true }, {}) + } else { + await expect(safeWriteJson(currentTestFilePath, {}, { merge })).rejects.toMatchObject({ code }) + expect(merge).not.toHaveBeenCalled() + expect(await readFileContent(unreadableBackup)).toEqual({ stale: true }) + expect(await readFileContent(readableBackup)).toEqual({ committed: true }) + } + }, + ) + + it.each(["rename", "stat"] as const)( + "should abort recovery and preserve history when backup %s fails", + async (operation) => { + await fs.unlink(currentTestFilePath) + const backupPath = path.join(tempDir, `.${path.basename(currentTestFilePath)}.bak_1_saved.tmp`) + const committed = { history: ["saved"] } + await fs.writeFile(backupPath, JSON.stringify(committed)) + const error = Object.assign(new Error("Recovery failed"), { code: "EACCES" }) + vi.mocked(fs[operation]).mockRejectedValueOnce(error) + const merge = vi.fn() + const trace = vi.fn() + + await expect(safeWriteJson(currentTestFilePath, {}, { merge, trace })).rejects.toBe(error) + + expect(trace).toHaveBeenCalledExactlyOnceWith({ stage: "recover", detail: "failed:EACCES" }) + expect(merge).not.toHaveBeenCalled() + expect(await readFileContent(backupPath)).toEqual(committed) + await expect(fs.readFile(currentTestFilePath)).rejects.toMatchObject({ code: "ENOENT" }) + await safeWriteJson(currentTestFilePath, {}, { merge: (existing) => existing }) + expect(await readFileContent(currentTestFilePath)).toEqual(committed) + }, + ) + + test("should restore by mtime and preserve unrelated siblings", async () => { + await fs.unlink(currentTestFilePath) + const prefix = `.${path.basename(currentTestFilePath)}` + const newest = path.join(tempDir, `${prefix}.bak_1_newest.tmp`) + const older = path.join(tempDir, `${prefix}.bak_9_older.tmp`) + const unrelated = path.join(tempDir, ".other.json.bak_1_saved.tmp") + const wrongSuffix = path.join(tempDir, `${prefix}.bak_1_saved.txt`) + await fs.writeFile(newest, JSON.stringify({ newest: true })) + await fs.writeFile(older, JSON.stringify({ older: true })) + await fs.writeFile(unrelated, "keep") + await fs.writeFile(wrongSuffix, "keep") + await fs.utimes(older, new Date(1000), new Date(1000)) + await fs.utimes(newest, new Date(2000), new Date(2000)) + + await safeWriteJson(currentTestFilePath, {}, { merge: (existing) => existing }) + + expect(await readFileContent(currentTestFilePath)).toEqual({ newest: true }) + expect(await fs.readFile(unrelated, "utf8")).toBe("keep") + expect(await fs.readFile(wrongSuffix, "utf8")).toBe("keep") + }) + + test("should open the replacement with write access for flushing", async () => { + const open = vi.mocked(fs.open) + open.mockClear() + + await safeWriteJson(currentTestFilePath, { flushed: true }) + + expect(open).toHaveBeenCalledExactlyOnceWith(expect.stringContaining(".new_"), "r+") + expect(await readFileContent(currentTestFilePath)).toEqual({ flushed: true }) + }) + + it.each([false, true])("should close the flush handle before commit when sync fails: %s", async (failSync) => { + const actual = await vi.importActual("fs/promises") + const steps: string[] = [] + vi.mocked(fs.open).mockImplementation(async (filePath, flags) => { + const handle = await actual.open(filePath, flags) + const sync = handle.sync.bind(handle) + const close = handle.close.bind(handle) + vi.spyOn(handle, "sync").mockImplementation(async () => { + steps.push("sync") + if (failSync) throw Object.assign(new Error("Flush failed"), { code: "EIO" }) + await sync() + }) + vi.spyOn(handle, "close").mockImplementation(async () => { + steps.push("close") + await close() + }) + return handle + }) + await safeWriteJson(currentTestFilePath, { saved: true }, { trace: ({ stage }) => steps.push(stage) }) + expect(steps).toEqual(["write-start", "sync", "close", "replacement-rename", "write-complete"]) + expect(await readFileContent(currentTestFilePath)).toEqual({ saved: true }) + }) + + test("should garbage collect orphaned replacement files when the canonical path exists", async () => { + const initialData = { message: "Initial content" } + + await fsPromisesActuals.writeFile!(currentTestFilePath, JSON.stringify(initialData)) + + // Leftovers from an interrupted write whose commit rename never ran. + const backupPath = path.join(tempDir, `.${path.basename(currentTestFilePath)}.bak_1_abc.tmp`) + const orphanedNewPath = path.join(tempDir, `.${path.basename(currentTestFilePath)}.new_1_def.tmp`) + await fsPromisesActuals.writeFile!(backupPath, JSON.stringify({ message: "stale backup" })) + await fsPromisesActuals.writeFile!(orphanedNewPath, "{}") + + const newData = { message: "New content" } + await safeWriteJson(currentTestFilePath, newData) + + const content = await readFileContent(currentTestFilePath) + expect(content).toEqual(newData) + + const leftovers = (await fs.readdir(tempDir)).filter( + (entry) => entry.includes(".new_") || entry.includes(".bak_"), + ) + expect(leftovers).toEqual([]) + }) + + test("should report replacement-protocol trace events when a trace callback is provided", async () => { + const traceEvents: Array<{ stage: string; detail?: string }> = [] + const newData = { message: "traced write" } + + await safeWriteJson(currentTestFilePath, newData, { + trace: (event) => traceEvents.push(event), + }) + + expect(traceEvents.map((event) => event.stage)).toEqual(["write-start", "replacement-rename", "write-complete"]) + const content = await readFileContent(currentTestFilePath) + expect(content).toEqual(newData) + }) + + test("should trace the EPERM fallback backup rename", async () => { + const traceEvents: Array<{ stage: string; detail?: string }> = [] + const initialData = { message: "Initial content" } + const newData = { message: "Traced fallback write" } + + await fsPromisesActuals.writeFile!(currentTestFilePath, JSON.stringify(initialData)) + + vi.mocked(fs.rename).mockImplementationOnce(async () => { + throw Object.assign(new Error("EPERM: operation not permitted"), { code: "EPERM" }) + }) + + await safeWriteJson(currentTestFilePath, newData, { + trace: (event) => traceEvents.push(event), + }) + + expect(traceEvents.map((event) => event.stage)).toEqual([ + "write-start", + "replacement-rename", + "backup-rename", + "replacement-rename", + "write-complete", + ]) + expect(traceEvents.find((event) => event.stage === "backup-rename")?.detail).toBe("eperm-fallback") + const content = await readFileContent(currentTestFilePath) + expect(content).toEqual(newData) + }) }) diff --git a/src/utils/errorCode.ts b/src/utils/errorCode.ts new file mode 100644 index 0000000000..56fdf885df --- /dev/null +++ b/src/utils/errorCode.ts @@ -0,0 +1,6 @@ +/** Return a filesystem error code without assuming the thrown value is an Error. */ +export function getErrorCode(error: unknown): string | undefined { + return typeof error === "object" && error !== null && "code" in error && typeof error.code === "string" + ? error.code + : undefined +} diff --git a/src/utils/fileLock.ts b/src/utils/fileLock.ts index 9f7cad7653..49807ea154 100644 --- a/src/utils/fileLock.ts +++ b/src/utils/fileLock.ts @@ -1,4 +1,5 @@ import * as path from "path" +import { tracePersistenceEvent } from "../core/task-persistence/persistenceTrace" import * as lockfile from "proper-lockfile" /** @@ -8,6 +9,16 @@ import * as lockfile from "proper-lockfile" */ export const LOCK_STALE_MS = 31_000 +const RECOVERY_RETRY_MS = 1000 +const RECOVERY_GRACE_MS = 4000 + +export interface FileLockOptions { + /** Recovery may wait out a crashed writer's staleness window without stealing a live lock. */ + waitForStaleLock?: boolean + /** Startup scans defer locked files instead of waiting under the store mutex. */ + immediate?: boolean +} + /** * Acquire the advisory lock for one file path using the exact protocol * `safeWriteJson` uses, so operations that hold this lock serialize with @@ -15,20 +26,30 @@ export const LOCK_STALE_MS = 31_000 * returned function exactly once and must not acquire the same lock again * while holding it. */ -export async function acquireFileLock(filePath: string): Promise<() => Promise> { +export async function acquireFileLock(filePath: string, options?: FileLockOptions): Promise<() => Promise> { const absoluteFilePath = path.resolve(filePath) + if (options?.waitForStaleLock) tracePersistenceEvent("json-lock-wait", { filePath: absoluteFilePath }) try { return await lockfile.lock(absoluteFilePath, { stale: LOCK_STALE_MS, update: 10000, // Update mtime every 10 seconds to prevent staleness if operation is long realpath: false, // the file may not exist yet, which is acceptable - retries: { - // Configuration for retrying lock acquisition - retries: 5, // Number of retries after the initial attempt - factor: 2, // Exponential backoff factor (e.g., 100ms, 200ms, 400ms, ...) - minTimeout: 100, // Minimum time to wait before the first retry (in ms) - maxTimeout: 1000, // Maximum time to wait for any single retry (in ms) - }, + retries: options?.immediate + ? 0 + : options?.waitForStaleLock + ? { + retries: Math.ceil((LOCK_STALE_MS + RECOVERY_GRACE_MS) / RECOVERY_RETRY_MS), + factor: 1, + minTimeout: RECOVERY_RETRY_MS, + maxTimeout: RECOVERY_RETRY_MS, + } + : { + // Configuration for retrying lock acquisition + retries: 5, // Number of retries after the initial attempt + factor: 2, // Exponential backoff factor (e.g., 100ms, 200ms, 400ms, ...) + minTimeout: 100, // Minimum time to wait before the first retry (in ms) + maxTimeout: 1000, // Maximum time to wait for any single retry (in ms) + }, onCompromised: (err) => { console.error(`Lock at ${absoluteFilePath} was compromised:`, err) throw err @@ -49,9 +70,10 @@ export async function acquireFileLock(filePath: string): Promise<() => Promise( filePath: string, operation: (absoluteFilePath: string) => Promise, + options?: FileLockOptions, ): Promise { const absoluteFilePath = path.resolve(filePath) - const releaseLock = await acquireFileLock(absoluteFilePath) + const releaseLock = await acquireFileLock(absoluteFilePath, options) let result: T try { diff --git a/src/utils/jsonFileRecovery.ts b/src/utils/jsonFileRecovery.ts new file mode 100644 index 0000000000..2350f30685 --- /dev/null +++ b/src/utils/jsonFileRecovery.ts @@ -0,0 +1,210 @@ +import * as fs from "fs/promises" +import * as path from "path" + +import { getErrorCode } from "./errorCode" +import { tracePersistenceEvent } from "../core/task-persistence/persistenceTrace" + +import { withFileLock } from "./fileLock" + +function getTempFilePrefix(absoluteFilePath: string, kind: "new" | "bak"): string { + return `.${path.basename(absoluteFilePath)}.${kind}_` +} + +function isReplacementLeftover(absoluteFilePath: string, entry: string): boolean { + return ( + entry.endsWith(".tmp") && + (entry.startsWith(getTempFilePrefix(absoluteFilePath, "new")) || + entry.startsWith(getTempFilePrefix(absoluteFilePath, "bak"))) + ) +} + +/** + * Recover a missing canonical file for a reader or startup scan, without a new + * write. A live writer retains its lock; a crashed writer's lock must age out. + * The unlocked scan is only a hint: recovery rechecks everything under the lock. + */ +export async function recoverMissingJsonFile( + filePath: string, + options: { waitForStaleLock?: boolean } = {}, +): Promise { + const absoluteFilePath = path.resolve(filePath) + let entries: string[] + try { + entries = await fs.readdir(path.dirname(absoluteFilePath)) + } catch (error: unknown) { + if (getErrorCode(error) === "ENOENT") return false + tracePersistenceEvent("json-recovery", { + filePath: absoluteFilePath, + result: "failed", + detail: getErrorCode(error), + }) + throw error + } + if (!entries.some((entry) => isReplacementLeftover(absoluteFilePath, entry))) return false + try { + return await withFileLock( + absoluteFilePath, + async (lockedPath) => { + const recovery = await recoverInterruptedReplacement(lockedPath) + if (recovery.restoredFrom) { + tracePersistenceEvent("json-recovery", { filePath: absoluteFilePath, result: "restored" }) + return true + } + // A peer may have committed while this reader waited for the lock. + try { + await fs.access(lockedPath) + return true + } catch (error) { + if (getErrorCode(error) === "ENOENT") return false + throw error + } + }, + { waitForStaleLock: options.waitForStaleLock ?? true, immediate: options.waitForStaleLock === false }, + ) + } catch (error) { + tracePersistenceEvent("json-recovery", { + filePath: absoluteFilePath, + result: "failed", + detail: getErrorCode(error), + }) + throw error + } +} + +/** + * Delete recoverable leftovers before the canonical file under the writer's + * lock. Failure preserves the canonical copy, rather than authorizing recovery + * of a stale backup after a successful deletion. Missing files are harmless. + */ +export async function deleteJsonFile(filePath: string): Promise { + await withFileLock(filePath, async (absoluteFilePath) => { + let entries: string[] + try { + entries = await fs.readdir(path.dirname(absoluteFilePath)) + } catch (error: unknown) { + if (getErrorCode(error) === "ENOENT") return + throw error + } + const leftovers = entries.filter((entry) => isReplacementLeftover(absoluteFilePath, entry)) + for (const target of [ + ...leftovers.map((entry) => path.join(path.dirname(absoluteFilePath), entry)), + absoluteFilePath, + ]) { + try { + await fs.unlink(target) + } catch (error: unknown) { + if (getErrorCode(error) !== "ENOENT") throw error + } + } + }) +} + +/** + * Heals the canonical path after an interrupted previous write. Must run + * under the advisory lock and before any data is read or written: temp and + * backup files are only ever created while that lock is held, so anything + * matching their name patterns belongs to a writer that can no longer + * commit or clean it up (a crashed process, or a stale lock that was + * stolen). + * + * The two-step replacement used by the EPERM fallback (and by versions + * before the single-rename commit) can be interrupted after the target was + * renamed to its backup but before the replacement landed. A process exit + * at that point leaves the canonical path absent, which readers see as a + * permanent ENOENT that no retry can repair, so the newest backup is + * restored first. Leftover replacement and backup files are garbage + * collected afterwards. + * + * @returns Whether leftover files were found, and the backup path that was + * restored, if any. + */ +export async function recoverInterruptedReplacement( + absoluteFilePath: string, +): Promise<{ foundLeftovers: boolean; restoredFrom: string | null }> { + const dirPath = path.dirname(absoluteFilePath) + const backupPrefix = getTempFilePrefix(absoluteFilePath, "bak") + + let entries: string[] + try { + entries = await fs.readdir(dirPath) + } catch (error: unknown) { + if (getErrorCode(error) === "ENOENT") { + return { foundLeftovers: false, restoredFrom: null } + } + // A failed scan cannot establish that no committed backups need recovery. + throw error + } + + const orphanedPaths = entries + .filter((entry) => isReplacementLeftover(absoluteFilePath, entry)) + .map((entry) => path.join(dirPath, entry)) + + if (orphanedPaths.length === 0) { + return { foundLeftovers: false, restoredFrom: null } + } + + let restoredFrom: string | null = null + let canonicalPathMissing = false + + try { + await fs.access(absoluteFilePath) + } catch (error: unknown) { + if (getErrorCode(error) !== "ENOENT") { + // Unknown canonical state must not authorize deleting committed backups. + throw error + } + canonicalPathMissing = true + } + + if (canonicalPathMissing) { + // The canonical path is absent. An interrupted two-step replacement + // leaves the committed content in the newest backup, so restore it + // and keep readers on the last committed state. + const backupStats = ( + await Promise.all( + orphanedPaths + .filter((orphanPath) => path.basename(orphanPath).startsWith(backupPrefix)) + .map(async (backupPath) => { + try { + return { backupPath, mtimeMs: (await fs.stat(backupPath)).mtimeMs } + } catch (error: unknown) { + if (getErrorCode(error) === "ENOENT") { + return null + } + // An unreadable backup could be the newest committed state. + // Intentionally block writes until it becomes readable or is + // explicitly removed, rather than risk losing committed history. + throw error + } + }), + ) + ).filter((entry): entry is { backupPath: string; mtimeMs: number } => entry !== null) + if (backupStats.length > 0) { + backupStats.sort((a, b) => b.mtimeMs - a.mtimeMs) + const newestBackupPath = backupStats[0].backupPath + try { + await fs.rename(newestBackupPath, absoluteFilePath) + restoredFrom = newestBackupPath + } catch (restoreError) { + console.error( + `Failed to restore interrupted-write backup ${newestBackupPath} to ${absoluteFilePath}:`, + restoreError, + ) + throw restoreError + } + } + } + + for (const orphanPath of orphanedPaths) { + if (orphanPath === restoredFrom) { + continue + } + try { + await fs.unlink(orphanPath) + } catch (cleanupError) { + console.error(`Failed to clean up interrupted-write leftover ${orphanPath}:`, cleanupError) + } + } + + return { foundLeftovers: true, restoredFrom } +} diff --git a/src/utils/outputChannelLogger.ts b/src/utils/outputChannelLogger.ts index 3b52449aee..c4e8be21eb 100644 --- a/src/utils/outputChannelLogger.ts +++ b/src/utils/outputChannelLogger.ts @@ -1,6 +1,6 @@ import * as vscode from "vscode" -export type LogFunction = (...args: unknown[]) => void +export type LogFunction = (...args: Args) => void /** * Creates a logging function that writes to a VSCode output channel diff --git a/src/utils/safeWriteJson.ts b/src/utils/safeWriteJson.ts index 773ba57c06..e54b8c51d3 100644 --- a/src/utils/safeWriteJson.ts +++ b/src/utils/safeWriteJson.ts @@ -3,7 +3,20 @@ import * as fsSync from "fs" import * as path from "path" import { JsonStreamStringify } from "json-stream-stringify" +import { getErrorCode } from "./errorCode" import { acquireFileLock } from "./fileLock" +import { recoverInterruptedReplacement } from "./jsonFileRecovery" + +/** + * Stages reported through the optional `trace` callback. They mark the + * replacement-protocol steps that matter when diagnosing interrupted or + * failing writes: the recovery pass, the temp-file write, the backup rename + * of the EPERM fallback, the commit rename, and completion. + */ +export type SafeWriteJsonTraceStage = + "recover" | "write-start" | "backup-rename" | "replacement-rename" | "write-complete" + +export type SafeWriteJsonTrace = (event: { stage: SafeWriteJsonTraceStage; detail?: string }) => void /** * Options for safeWriteJson function @@ -26,14 +39,29 @@ export interface SafeWriteJsonOptions { * cannot be parsed. */ merge?: (existing: unknown, incoming: unknown) => unknown + + /** + * When provided, receives one event per replacement-protocol step (see + * SafeWriteJsonTraceStage). Callers such as the task-message save use it + * to emit diagnostic trace lines; it never influences the write itself. + */ + trace?: SafeWriteJsonTrace +} + +function createTempFilePath(absoluteFilePath: string, kind: "new" | "bak"): string { + return path.join( + path.dirname(absoluteFilePath), + `.${path.basename(absoluteFilePath)}.${kind}_${Date.now()}_${Math.random().toString(36).substring(2)}.tmp`, + ) } /** * Safely writes JSON data to a file. * - Creates parent directories if they don't exist * - Uses 'proper-lockfile' for inter-process advisory locking to prevent concurrent writes to the same path. - * - Writes to a temporary file first. - * - If the target file exists, it's backed up before being replaced. + * - Recovers an interrupted previous write before touching the file: restores an orphaned backup when the canonical path is absent, and removes leftover temp files. + * - Writes to a temporary file first, then commits it with a single atomic rename, so the committed target stays readable through an interrupted write. + * - When a single rename over the target fails with EPERM (a reader holds the file open on Windows), falls back to backing the target up before replacing it. * - Attempts to roll back and clean up in case of errors. * - Supports pretty-printing with indentation while maintaining streaming efficiency. * @@ -76,6 +104,18 @@ async function safeWriteJson(filePath: string, data: unknown, options?: SafeWrit let actualTempBackupFilePath: string | null = null try { + // Heal any state an interrupted previous write left behind: an absent + // canonical path with an orphaned backup, or leftover replacement + // files. Runs first so a merge callback reads the restored content. + const recovery = await recoverInterruptedReplacement(absoluteFilePath).catch((error: unknown) => { + const code = getErrorCode(error) ?? "unknown" + options?.trace?.({ stage: "recover", detail: `failed:${code}` }) + throw error + }) + if (recovery.foundLeftovers) { + options?.trace?.({ stage: "recover", detail: recovery.restoredFrom ?? "cleaned-leftovers" }) + } + // If a merge callback was provided, read the current file under the lock // and let the caller merge before we write. Must be inside try/finally // so a throwing merge still releases the lock. @@ -84,8 +124,7 @@ async function safeWriteJson(filePath: string, data: unknown, options?: SafeWrit try { existing = JSON.parse(await fs.readFile(absoluteFilePath, "utf8")) } catch (error: unknown) { - const code = - error && typeof error === "object" && "code" in error ? (error as { code: string }).code : undefined + const code = getErrorCode(error) if (!(error instanceof SyntaxError) && code !== "ENOENT") { throw error } @@ -93,47 +132,46 @@ async function safeWriteJson(filePath: string, data: unknown, options?: SafeWrit data = options.merge(existing, data) } - // Step 1: Write data to a new temporary file. - actualTempNewFilePath = path.join( - path.dirname(absoluteFilePath), - `.${path.basename(absoluteFilePath)}.new_${Date.now()}_${Math.random().toString(36).substring(2)}.tmp`, - ) + // Step 1: Write data to a new temporary file and flush it to storage, + // making a best-effort flush before committing the replacement. + options?.trace?.({ stage: "write-start" }) + actualTempNewFilePath = createTempFilePath(absoluteFilePath, "new") await _streamDataToFile(actualTempNewFilePath, data, options?.prettyPrint) + await _flushFileToDisk(actualTempNewFilePath) - // Step 2: Check if the target file exists. If so, rename it to a backup path. + // Step 2: Commit the replacement with a single atomic rename. The + // committed target stays in place until this rename lands, so a + // process exit before or during the commit can no longer leave the + // canonical path absent. try { - // Check for target file existence - await fs.access(absoluteFilePath) - // Target exists, create a backup path and rename. - actualTempBackupFilePath = path.join( - path.dirname(absoluteFilePath), - `.${path.basename(absoluteFilePath)}.bak_${Date.now()}_${Math.random().toString(36).substring(2)}.tmp`, - ) - await fs.rename(absoluteFilePath, actualTempBackupFilePath) - } catch (accessError: unknown) { - if ( - typeof accessError !== "object" || - accessError === null || - !("code" in accessError) || - accessError.code !== "ENOENT" - ) { - // An error other than "file not found" occurred during access check. - throw accessError + options?.trace?.({ stage: "replacement-rename" }) + await fs.rename(actualTempNewFilePath, absoluteFilePath) + } catch (renameError: unknown) { + if (getErrorCode(renameError) !== "EPERM") { + throw renameError } - // Target file does not exist, so no backup is made. actualTempBackupFilePath remains null. - } - // Step 3: Rename the new temporary file to the target file path. - // This is the main "commit" step. - await fs.rename(actualTempNewFilePath, absoluteFilePath) + // A single rename over a file that a reader holds open fails with + // EPERM on Windows, so fall back to the two-step replacement: + // rename the target to a backup, then the replacement into place. + // The rollback in the catch below restores the backup when the + // second rename fails, and the recovery pass at the start of the + // next write repairs a crash between the two renames. + options?.trace?.({ stage: "backup-rename", detail: "eperm-fallback" }) + actualTempBackupFilePath = createTempFilePath(absoluteFilePath, "bak") + await fs.rename(absoluteFilePath, actualTempBackupFilePath) + options?.trace?.({ stage: "replacement-rename", detail: "eperm-fallback" }) + await fs.rename(actualTempNewFilePath, absoluteFilePath) + } // If we reach here, the new file is successfully in place. // The original actualTempNewFilePath is now the main file, so we shouldn't try to clean it up as "temp". // Mark as "used" or "committed" actualTempNewFilePath = null + options?.trace?.({ stage: "write-complete" }) - // Step 4: If a backup was created, attempt to delete it. + // Step 3: If a backup was created, attempt to delete it. if (actualTempBackupFilePath) { try { await fs.unlink(actualTempBackupFilePath) @@ -181,17 +219,8 @@ async function safeWriteJson(filePath: string, data: unknown, options?: SafeWrit } } - // Cleanup the .bak file if it still needs to be (i.e., wasn't successfully restored) - if (actualTempBackupFilePath) { - try { - await fs.unlink(actualTempBackupFilePath) - } catch (cleanupError) { - console.error( - `[Catch] Failed to clean up temporary backup file ${actualTempBackupFilePath}:`, - cleanupError, - ) - } - } + // A failed rollback leaves the backup as the only committed copy. + // Preserve it for recovery by a later writer. throw originalError // This MUST be the error that rejects the promise. } finally { // Release the lock in the main finally block. @@ -234,4 +263,23 @@ async function _streamDataToFile(targetPath: string, data: unknown, prettyPrint }) } +/** + * Flushes a fully written temp file to storage before it is committed. + * Best effort: durability against power loss is a bonus here, process-crash + * safety comes from the rename protocol itself, and a filesystem that + * refuses fsync must not block the write. + */ +async function _flushFileToDisk(targetPath: string): Promise { + try { + const handle = await fs.open(targetPath, "r+") + try { + await handle.sync() + } finally { + await handle.close() + } + } catch (flushError) { + console.error(`Failed to flush ${targetPath} to storage before commit:`, flushError) + } +} + export { safeWriteJson }