From 0473a3091c75fd32ad5609af3e4fdc035427318c Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Fri, 23 Oct 2026 18:21:31 +0800 Subject: [PATCH] fix(recovery): retain file outcomes and cleanup ownership Use shared owned-file and directory primitives throughout Wakaru discovery, snapshots, report parsing and publication. Preserve primary read failures and completed fingerprints when final close fails or receives cancellation. Retain exact cleanup capabilities before retiring workspaces, block new work while cleanup is unresolved, and bind error status to the aggregate owner so partial cleanup and later retry keep resource attribution current. --- .../recovery/RecoveryFiles.ownership.test.ts | 167 ++++++++++++++ src/javascript/recovery/RecoveryFiles.test.ts | 23 +- src/javascript/recovery/RecoveryFiles.ts | 167 +++++++++----- src/javascript/recovery/WakaruCommand.ts | 18 +- src/javascript/recovery/WakaruExecution.ts | 29 ++- src/javascript/recovery/WakaruProvider.ts | 180 ++++++++------- src/javascript/recovery/WakaruPublication.ts | 10 +- src/javascript/recovery/WakaruReport.ts | 28 ++- .../wakaruFileOwnership.test.ts | 206 ++++++++++++++++++ 9 files changed, 678 insertions(+), 150 deletions(-) create mode 100644 src/javascript/recovery/RecoveryFiles.ownership.test.ts create mode 100644 tests/boundary/javascript-recovery/wakaruFileOwnership.test.ts diff --git a/src/javascript/recovery/RecoveryFiles.ownership.test.ts b/src/javascript/recovery/RecoveryFiles.ownership.test.ts new file mode 100644 index 0000000000..80b578ce8c --- /dev/null +++ b/src/javascript/recovery/RecoveryFiles.ownership.test.ts @@ -0,0 +1,167 @@ +import type { Dir } from "node:fs"; +import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js"; +import { createHash } from "node:crypto"; +import { writeFile, type FileHandle } from "node:fs/promises"; +import { dirname, join } from "node:path"; +import { afterEach, expect, it, onTestFinished, vi } from "vitest"; +import { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js"; +import { ArtifactFileCleanupFailure } from "../../artifacts/ArtifactFileOperation.js"; +import { createTestTempDirectory } from "../../../tests/fixtures/temporaryDirectory.js"; +import { inventoryRecoveryFiles, readRecoveryFile } from "./RecoveryFiles.js"; + +const closeFault = vi.hoisted(() => ({ + path: undefined as string | undefined, + denied: false, + handles: [] as FileHandle[], + failure: new Error("controlled recovery descriptor close failure"), +})); + +const directoryFault = vi.hoisted(() => ({ + path: undefined as string | undefined, + readFailure: undefined as Error | undefined, + cancel: undefined as AbortController | undefined, + closeAttempts: 0, +})); + +vi.mock("node:fs/promises", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + opendir: async (...args: Parameters) => { + const handle: Dir = await actual.opendir(...args); + if (args[0] === directoryFault.path) { + const close = handle.close.bind(handle); + onTestFinished(async () => { + await close(); + }); + const read = handle.read.bind(handle); + handle.read = async () => { + if (directoryFault.readFailure !== undefined) + throw directoryFault.readFailure; + return read(); + }; + handle.close = async () => { + directoryFault.closeAttempts++; + directoryFault.cancel?.abort(); + throw new Error("controlled recovery directory close failure"); + }; + } + return handle; + }, + open: async (...args: Parameters) => { + const handle = await actual.open(...args); + if (args[0] === closeFault.path) { + const close = handle.close.bind(handle); + closeFault.handles.push(handle); + onTestFinished(async () => { + if (handle.fd >= 0) await close(); + }); + handle.close = async () => { + if (closeFault.denied) throw closeFault.failure; + await close(); + }; + } + return handle; + }, + }; +}); + +afterEach(() => { + directoryFault.path = undefined; + directoryFault.readFailure = undefined; + directoryFault.cancel = undefined; + directoryFault.closeAttempts = 0; + closeFault.path = undefined; + closeFault.denied = false; + closeFault.handles = []; +}); + +const selectedFile = async () => { + const root = await createTestTempDirectory("rea-recovery-file-owner-"); + const path = join(root, "bundle.js"); + const bytes = Buffer.from("module.exports = 42;\n"); + await writeFile(path, bytes); + closeFault.path = path; + return { path, bytes }; +}; + +it.each(["completed", "size-limit"] as const)( + "retains the exact descriptor and %s read outcome through a failed close", + async (mode) => { + const { path, bytes } = await selectedFile(); + const resources = new ArtifactResourceScope(); + closeFault.denied = true; + const failure = await readRecoveryFile( + path, + mode === "completed" ? 1024 : 0, + undefined, + resources, + ).catch((cause: unknown) => cause); + expect(failure).toBeInstanceOf(ArtifactFileCleanupFailure); + if (!(failure instanceof ArtifactFileCleanupFailure)) return; + const physical = failure.fileCleanupFailure; + expect(physical.owner.handle).toBe(closeFault.handles[0]); + expect(physical.cleanupCause).toBe(closeFault.failure); + if (mode === "completed") + expect(physical.outcome).toEqual({ + kind: "completed", + value: { + sha256: createHash("sha256").update(bytes).digest("hex"), + bytes: bytes.length, + }, + }); + else + expect(physical.outcome).toMatchObject({ + kind: "failed", + cause: { maximum: 0 }, + }); + expect(resources.hasPending).toBe(true); + await expect(resources.retryCleanup()).rejects.toBeDefined(); + expect(closeFault.handles[0]?.fd).toBeGreaterThanOrEqual(0); + closeFault.denied = false; + await resources.retryCleanup(); + expect(resources.hasPending).toBe(false); + expect(closeFault.handles[0]?.fd).toBe(-1); + }, +); + +it.each([ + [false, false], + [true, false], + [false, true], + [true, true], +] as const)( + "retains an uncertain directory close and its primary read outcome (read denied: %s, cancel on close: %s)", + async (readDenied, cancelOnClose) => { + const { path } = await selectedFile(); + const root = dirname(path); + directoryFault.path = root; + const readFailure = new Error("controlled directory read failure"); + if (readDenied) directoryFault.readFailure = readFailure; + const resources = new ArtifactResourceScope(); + const controller = new AbortController(); + if (cancelOnClose) directoryFault.cancel = controller; + const failure = await inventoryRecoveryFiles( + root, + controller.signal, + resources, + ).catch((cause: unknown) => cause); + expect(failure).toBeInstanceOf(ArtifactReaderFailure); + if (!(failure instanceof ArtifactReaderFailure)) return; + expect(failure.cleanup?.resources).toContain(root); + if (readDenied) { + expect(failure.cause).toBeInstanceOf(ArtifactReaderFailure); + expect((failure.cause as ArtifactReaderFailure).cause).toBe(readFailure); + } + if (cancelOnClose && !readDenied) { + expect(failure.reason).toBe("cancelled"); + expect(failure.cancellation?.reason).toBe(controller.signal.reason); + } + expect(resources.hasPending).toBe(true); + expect(directoryFault.closeAttempts).toBe(1); + await expect(resources.retryCleanup()).rejects.toBeDefined(); + // Node has no public directory validity signal, so an unconfirmed close cannot be retried as if it were open. + expect(directoryFault.closeAttempts).toBe(1); + expect(resources.hasPending).toBe(true); + }, +); diff --git a/src/javascript/recovery/RecoveryFiles.test.ts b/src/javascript/recovery/RecoveryFiles.test.ts index 10c2c5d689..b878dd192d 100644 --- a/src/javascript/recovery/RecoveryFiles.test.ts +++ b/src/javascript/recovery/RecoveryFiles.test.ts @@ -7,6 +7,7 @@ import { AnalysisInputError, AnalysisOutputError, } from "../../domain/analysisErrorCore.js"; +import { ArtifactFileCleanupFailure } from "../../artifacts/ArtifactFileOperation.js"; import { OwnedFileHandle } from "../../filesystem/OwnedFileHandle.js"; import { openRegularFile, @@ -92,12 +93,24 @@ it("keeps a failed admission close owned when cancellation is also requested", a new Error("close failed"), owner, ); - vi.mocked(openRegularFile).mockRejectedValueOnce(failure); const controller = new AbortController(); - controller.abort(); - await expect( - readRecoveryFile("/selected.js", 1024, controller.signal), - ).rejects.toBe(failure); + vi.mocked(openRegularFile).mockImplementationOnce(async () => { + controller.abort(); + throw failure; + }); + const observed = await readRecoveryFile( + "/selected.js", + 1024, + controller.signal, + ).catch((cause: unknown) => cause); + expect(observed).toBeInstanceOf(ArtifactFileCleanupFailure); + if (!(observed instanceof ArtifactFileCleanupFailure)) return; + expect(observed.fileCleanupFailure.owner).toBe(owner); + expect(observed.fileCleanupFailure.outcome).toEqual({ + kind: "failed", + cause: failure.cause, + }); + expect(observed.fileCleanupFailure.cleanupCause).toBe(failure.cleanupCause); }); it("reports one file selected under two letter cases as an input error", async (context) => { diff --git a/src/javascript/recovery/RecoveryFiles.ts b/src/javascript/recovery/RecoveryFiles.ts index 28571a7345..1da3e3e7de 100644 --- a/src/javascript/recovery/RecoveryFiles.ts +++ b/src/javascript/recovery/RecoveryFiles.ts @@ -1,4 +1,12 @@ -import { regularFileCancellation } from "../../filesystem/OwnedRegularFile.js"; +import { + artifactFileCancellationCause, + ArtifactFileCancelled, + ArtifactFileCleanupFailure, + withArtifactRegularFile, +} from "../../artifacts/ArtifactFileOperation.js"; +import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js"; +import { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js"; +import { OwnedDirectoryHandle } from "../../filesystem/OwnedDirectoryHandle.js"; import { createHash } from "node:crypto"; import { lstat, mkdir, opendir, writeFile } from "node:fs/promises"; import { dirname, isAbsolute, join, relative, resolve, sep } from "node:path"; @@ -13,8 +21,7 @@ import { } from "../../domain/analysisErrorCore.js"; import { NonRegularFileReadError, - openRegularFile, - RegularFileAdmissionFailure, + sameRegularFileState, } from "../../filesystem/RegularFile.js"; import { RECOVERY_LIMITS, wakaruFixedLimitRecovery } from "./WakaruRelease.js"; @@ -48,19 +55,56 @@ const scanRecoveryFile = async ( maximum: number, consume: (chunk: Buffer) => void, signal?: AbortSignal, + resources?: ArtifactResourceScope, ) => { - // A FIFO without a writer must be rejected before it can block the open. - const handle = await openRegularFile(path, { - symlinks: "reject", - signal, - }).catch((cause: unknown) => { - // A descriptor whose close failed stays owned by its error, even when - // cancellation was also requested. - if (cause instanceof RegularFileAdmissionFailure) throw cause; - const cancellation = regularFileCancellation(cause); - if (cancellation !== undefined) + try { + return await withArtifactRegularFile( + path, + resources ?? new ArtifactResourceScope(), + signal, + async (handle) => { + const hash = createHash("sha256"); + const before = await handle.stat(); + if (before.size > maximum) + throw new RecoveryFileSizeError( + `Expected a regular file no larger than ${String(maximum)} bytes: ${path}`, + maximum, + ); + let bytes = 0; + for await (const raw of handle.createReadStream({ autoClose: false })) { + checkRecoveryCancellation(signal); + const chunk = Buffer.isBuffer(raw) ? raw : Buffer.from(raw); + bytes += chunk.length; + if (bytes > maximum) + throw new RecoveryFileSizeError( + `File exceeds ${String(maximum)} bytes: ${path}`, + maximum, + ); + hash.update(chunk); + consume(chunk); + } + const after = await handle.stat(); + const current = await lstat(path); + if ( + before.size !== bytes || + [after, current].some((state) => !sameRegularFileState(before, state)) + ) + throw new AnalysisOutputError( + OPERATION, + `File identity changed while reading: ${path}`, + ); + return { sha256: hash.digest("hex"), bytes }; + }, + ); + } catch (cause: unknown) { + if ( + cause instanceof ArtifactFileCleanupFailure || + cause instanceof ArtifactFileCancelled + ) + throw cause; + if (cause instanceof ArtifactReaderFailure && cause.reason === "cancelled") throw new AnalysisCancelledError(OPERATION, { - cause: cancellation.reason, + cause: artifactFileCancellationCause(cause), }); if (cause instanceof NonRegularFileReadError) throw new AnalysisOutputError( @@ -68,45 +112,6 @@ const scanRecoveryFile = async ( `Expected a regular file no larger than ${String(maximum)} bytes: ${path}`, ); throw cause; - }); - const hash = createHash("sha256"); - try { - const before = await handle.stat(); - if (before.size > maximum) - throw new RecoveryFileSizeError( - `Expected a regular file no larger than ${String(maximum)} bytes: ${path}`, - maximum, - ); - let bytes = 0; - for await (const raw of handle.createReadStream({ autoClose: false })) { - checkRecoveryCancellation(signal); - const chunk = Buffer.isBuffer(raw) ? raw : Buffer.from(raw); - bytes += chunk.length; - if (bytes > maximum) - throw new RecoveryFileSizeError( - `File exceeds ${String(maximum)} bytes: ${path}`, - maximum, - ); - hash.update(chunk); - consume(chunk); - } - const after = await handle.stat(); - const current = await lstat(path); - if ( - before.dev !== current.dev || - before.ino !== current.ino || - before.size !== bytes || - before.mtimeMs !== after.mtimeMs || - before.ctimeMs !== after.ctimeMs || - before.size !== after.size - ) - throw new AnalysisOutputError( - OPERATION, - `File identity changed while reading: ${path}`, - ); - return { sha256: hash.digest("hex"), bytes }; - } finally { - await handle.close(); } }; @@ -115,13 +120,15 @@ export const fingerprintRecoveryFile = async ( path: string, maximum: number, signal?: AbortSignal, -) => scanRecoveryFile(path, maximum, () => {}, signal); + resources?: ArtifactResourceScope, +) => scanRecoveryFile(path, maximum, () => {}, signal, resources); /** Read only a regular file through one handle with a bounded, stable identity. */ export const readRecoveryFile = async ( path: string, maximum: number, signal?: AbortSignal, + resources?: ArtifactResourceScope, ): Promise => { const chunks: Buffer[] = []; const result = await scanRecoveryFile( @@ -129,6 +136,7 @@ export const readRecoveryFile = async ( maximum, (chunk) => chunks.push(chunk), signal, + resources, ); return Buffer.concat(chunks, result.bytes); }; @@ -148,6 +156,7 @@ const snapshotRecoveryInput = async ( relativePath: string, maximum: number, signal?: AbortSignal, + resources?: ArtifactResourceScope, ): Promise => { if (!isAbsolute(path)) throw recoveryInputFailure( @@ -156,9 +165,14 @@ const snapshotRecoveryInput = async ( ); let bytes: Buffer; try { - bytes = await readRecoveryFile(path, maximum, signal); + bytes = await readRecoveryFile(path, maximum, signal, resources); } catch (cause: unknown) { - if (cause instanceof AnalysisCancelledError) throw cause; + if ( + cause instanceof AnalysisCancelledError || + cause instanceof ArtifactFileCleanupFailure || + cause instanceof ArtifactFileCancelled + ) + throw cause; // The operation-wide input allowance is a deterministic byte budget, not // a malformed argument; the same selection fails again unchanged. if (cause instanceof RecoveryFileSizeError) @@ -227,6 +241,7 @@ export const snapshotRecoveryInputs = async ( paths: readonly [string, ...string[]], directory: string, signal?: AbortSignal, + resources?: ArtifactResourceScope, ): Promise => { let commonDirectory = dirname(resolve(paths[0])); for (const path of paths) { @@ -277,6 +292,7 @@ export const snapshotRecoveryInputs = async ( relativePath, remaining, signal, + resources, ); snapshots.push(snapshot); remaining -= snapshot.bytes; @@ -292,6 +308,7 @@ export const snapshotRecoveryInputs = async ( export const inventoryRecoveryFiles = async ( root: string, signal?: AbortSignal, + resources: ArtifactResourceScope = new ArtifactResourceScope(), ): Promise< readonly { relativePath: string; path: string; bytes: number }[] > => { @@ -306,9 +323,15 @@ export const inventoryRecoveryFiles = async ( `Unsafe staging directory: ${directory}`, ); const handle = await opendir(directory); + const owner = new OwnedDirectoryHandle(handle); const children: string[] = []; + let outcome: + | { readonly kind: "completed" } + | { readonly kind: "failed"; readonly cause: unknown }; try { - for await (const entry of handle) { + for (;;) { + const entry = await handle.read(); + if (entry === null) break; checkRecoveryCancellation(signal); entries++; if (entries > RECOVERY_LIMITS.outputEntries) @@ -321,9 +344,32 @@ export const inventoryRecoveryFiles = async ( ); children.push(entry.name); } - } finally { - await handle.close().catch(() => undefined); + outcome = { kind: "completed" }; + } catch (cause: unknown) { + outcome = { kind: "failed", cause }; } + const cleanup = await resources.release({ + kind: "directory-handle", + handle: owner, + resource: directory, + }); + if (cleanup.kind === "failed") { + const failure = resources.cleanupFailure(outcome, [cleanup]); + if (outcome.kind === "completed" && signal?.aborted === true) + throw new ArtifactReaderFailure( + "cancelled", + "Recovery inventory cancelled during directory close", + { + cause: failure, + cancellation: { reason: signal.reason }, + cleanup: failure.cleanup, + cleanupLeases: failure.cleanupLeases, + }, + ); + throw failure; + } + if (outcome.kind === "failed") throw outcome.cause; + checkRecoveryCancellation(signal); for (const name of children) { checkRecoveryCancellation(signal); const relativePath = prefix === "" ? name : `${prefix}/${name}`; @@ -369,7 +415,8 @@ export const recoveryFailureMessage = (cause: unknown): string => /** Cancellation remains distinct from malformed source or provider output. */ export const checkRecoveryCancellation = (signal?: AbortSignal): void => { - if (signal?.aborted === true) throw new AnalysisCancelledError(OPERATION); + if (signal?.aborted === true) + throw new AnalysisCancelledError(OPERATION, { cause: signal.reason }); }; /** Keep both tool invocations within one operation's execution budget. */ diff --git a/src/javascript/recovery/WakaruCommand.ts b/src/javascript/recovery/WakaruCommand.ts index 69a050daa5..3227b9fb45 100644 --- a/src/javascript/recovery/WakaruCommand.ts +++ b/src/javascript/recovery/WakaruCommand.ts @@ -7,6 +7,8 @@ import { AnalysisCapabilityUnavailableError, AnalysisOutputError, } from "../../domain/analysisErrorCore.js"; +import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js"; +import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js"; import { ProviderAdapterError } from "../../domain/providerAdapterError.js"; import { ProviderCleanupError } from "../../domain/providerCleanupError.js"; import { prlimitArgumentVector } from "../../process/LinuxResourceLimits.js"; @@ -68,6 +70,8 @@ export class WakaruCleanupFailure extends ProviderCleanupError { /** Resolve and fingerprint explicitly configured tools; acquire nothing at startup. */ export const resolveWakaruCommand = async ( environment: Readonly>, + resources?: ArtifactResourceScope, + signal?: AbortSignal, ) => { const configured = environment.REA_WAKARU_COMMAND; if ( @@ -87,6 +91,8 @@ export const resolveWakaruCommand = async ( const fingerprint = await fingerprintRecoveryFile( command, RECOVERY_LIMITS.outputBytes, + signal, + resources, ); const limiter = environment.REA_JAVASCRIPT_PRLIMIT_COMMAND ?? "/usr/bin/prlimit"; @@ -97,6 +103,11 @@ export const resolveWakaruCommand = async ( throw new TypeError(`Resource limiter is not a regular file: ${limiter}`); return { command, limiter, ...fingerprint }; } catch (cause: unknown) { + if ( + cause instanceof ArtifactReaderFailure && + (cause.cleanup !== undefined || cause.reason === "cancelled") + ) + throw cause; const reason = `Configured Wakaru or util-linux prlimit is unavailable: ${recoveryFailureMessage(cause)}`; throw new AnalysisCapabilityUnavailableError("wakaru", OPERATION, reason, { cause, @@ -115,6 +126,7 @@ type WakaruCommandContext = { signal?: AbortSignal; launcher?: WakaruLauncher; staging?: string; + resources?: ArtifactResourceScope; }; /** Run one bounded command and verify cleanup before consuming its output. */ @@ -183,7 +195,11 @@ export const runWakaruCommand = async (context: WakaruCommandContext) => { ), ); if (context.staging !== undefined) - await inventoryRecoveryFiles(context.staging, context.signal); + await inventoryRecoveryFiles( + context.staging, + context.signal, + context.resources, + ); } checkRecoveryDeadline(context.deadline, context.signal); if (oversized) diff --git a/src/javascript/recovery/WakaruExecution.ts b/src/javascript/recovery/WakaruExecution.ts index 7ec93ded50..5da9f86359 100644 --- a/src/javascript/recovery/WakaruExecution.ts +++ b/src/javascript/recovery/WakaruExecution.ts @@ -4,6 +4,7 @@ import { type ExecutionOptions, } from "../../application/AnalysisProvider.js"; import type { SafeOutputTree } from "../../artifacts/SafeOutputTree.js"; +import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js"; import { AnalysisCapabilityUnavailableError, AnalysisOutputError, @@ -43,12 +44,21 @@ type RecoveryWorkspace = { launcher: WakaruLauncher | undefined; options: ExecutionOptions | undefined; deadline: number; + resources: ArtifactResourceScope; }; /** Consume the pinned producer representation in a private workspace. */ const executeWakaruRecovery = async (context: RecoveryWorkspace) => { - const { root, input, sources, engine, environment, launcher, options } = - context; + const { + root, + input, + sources, + engine, + environment, + launcher, + options, + resources, + } = context; const staging = join(root, "modules"); const commandContext = { ...engine, @@ -57,6 +67,7 @@ const executeWakaruRecovery = async (context: RecoveryWorkspace) => { environment: environment, ...(options?.signal === undefined ? {} : { signal: options.signal }), ...(launcher === undefined ? {} : { launcher: launcher }), + resources, }; const version = await runWakaruCommand({ ...commandContext, @@ -90,7 +101,11 @@ const executeWakaruRecovery = async (context: RecoveryWorkspace) => { staging, ], }); - const files = await inventoryRecoveryFiles(staging, options?.signal); + const files = await inventoryRecoveryFiles( + staging, + options?.signal, + resources, + ); const snapshots = new Map(); for (const source of sources) snapshots.set( @@ -99,6 +114,7 @@ const executeWakaruRecovery = async (context: RecoveryWorkspace) => { source.snapshot_path, source.bytes, options?.signal, + resources, ), ); const parsed = await parseWakaruReports({ @@ -106,13 +122,15 @@ const executeWakaruRecovery = async (context: RecoveryWorkspace) => { provenancePath: join(staging, "provenance.json"), snapshots, files, + ...(options?.signal === undefined ? {} : { signal: options.signal }), + resources, }); return { version, admitted, run, parsed }; }; /** Publish validated recovery and retain exact source and engine bindings. */ export const prepareWakaruExecution = async (context: RecoveryWorkspace) => { - const { root, tree, input, sources, engine, options } = context; + const { root, tree, input, sources, engine, options, resources } = context; const [source] = sources; const staging = join(root, "modules"); const { version, admitted, run, parsed } = @@ -127,12 +145,14 @@ export const prepareWakaruExecution = async (context: RecoveryWorkspace) => { parsed, execution: run, ...(options?.signal === undefined ? {} : { signal: options.signal }), + resources, }); for (const selected of sources) { const current = await fingerprintRecoveryFile( selected.path, RECOVERY_LIMITS.inputBytes, options?.signal, + resources, ); if (current.sha256 !== selected.sha256 || current.bytes !== selected.bytes) throw new AnalysisOutputError( @@ -146,6 +166,7 @@ export const prepareWakaruExecution = async (context: RecoveryWorkspace) => { engine.command, RECOVERY_LIMITS.outputBytes, options?.signal, + resources, ) ).sha256 !== engine.sha256 ) diff --git a/src/javascript/recovery/WakaruProvider.ts b/src/javascript/recovery/WakaruProvider.ts index e258126630..626dd2b1cb 100644 --- a/src/javascript/recovery/WakaruProvider.ts +++ b/src/javascript/recovery/WakaruProvider.ts @@ -6,6 +6,8 @@ import type { AnalysisExecution, ExecutionOptions, } from "../../application/AnalysisProvider.js"; +import { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js"; +import { AnalysisCleanupLease } from "../../artifacts/AnalysisCleanupLease.js"; import { SafeOutputTree } from "../../artifacts/SafeOutputTree.js"; import { SafeOutputTreeCreationFailure } from "../../artifacts/SafeOutputTreeCreationFailure.js"; import { AnalysisError } from "../../domain/analysisErrorBase.js"; @@ -39,16 +41,22 @@ import { RECOVERY_LIMITS } from "./WakaruRelease.js"; const OPERATION = "recover_javascript_sources"; +interface PendingRecoveryCleanup { + root: string | undefined; + process: ProviderProcessSupervisor | undefined; + tree: SafeOutputTree | undefined; +} + /** Serialized recovery adapter retaining failed process, workspace and publication cleanup. */ export class WakaruProvider implements JavaScriptRecoveryPort { + readonly #resources = new ArtifactResourceScope(); #tail: Promise = Promise.resolve(); #closing = false; #closePromise: Promise | undefined; #pendingCleanup: | { - root: string | undefined; - process: ProviderProcessSupervisor | undefined; - tree: SafeOutputTree | undefined; + readonly resources: PendingRecoveryCleanup; + readonly lease: AnalysisCleanupLease; } | undefined; @@ -107,7 +115,11 @@ export class WakaruProvider implements JavaScriptRecoveryPort { const deadline = performance.now() + RECOVERY_LIMITS.timeoutMs; try { checkRecoveryCancellation(options?.signal); - const engine = await resolveWakaruCommand(this.environment); + const engine = await resolveWakaruCommand( + this.environment, + this.#resources, + options?.signal, + ); root = await mkdtemp(join(tmpdir(), "rea-wakaru-")); const inputs = join(root, "inputs"); const staging = join(root, "modules"); @@ -117,6 +129,7 @@ export class WakaruProvider implements JavaScriptRecoveryPort { [input.path, ...(input.additional_paths ?? [])], inputs, options?.signal, + this.#resources, ); try { tree = await SafeOutputTree.create(input.output_directory); @@ -131,6 +144,7 @@ export class WakaruProvider implements JavaScriptRecoveryPort { execution = await prepareWakaruExecution({ root, tree, + resources: this.#resources, input, sources, engine, @@ -150,7 +164,8 @@ export class WakaruProvider implements JavaScriptRecoveryPort { if (failed !== undefined) { const failure = failed.cause instanceof ArtifactReaderFailure && - failed.cause.reason === "cancelled" + failed.cause.reason === "cancelled" && + failed.cause.cleanup === undefined ? new AnalysisCancelledError(OPERATION, { cause: failed.cause }) : failed.cause; const residuals: string[] = []; @@ -174,7 +189,9 @@ export class WakaruProvider implements JavaScriptRecoveryPort { } if (cleanupOwner !== undefined) residuals.push(cleanupOwner.launch.ownership?.runId ?? "wakaru-worker"); - if (root !== undefined && cleanupOwner === undefined) { + const filesPending = this.#resources.hasPending; + if (filesPending) residuals.push(...this.#resources.cleanupResources); + if (root !== undefined && cleanupOwner === undefined && !filesPending) { const failedRoot = root; try { await rm(failedRoot, { recursive: true, force: true }); @@ -190,47 +207,75 @@ export class WakaruProvider implements JavaScriptRecoveryPort { if ( cleanupOwner !== undefined || failedTree !== undefined || - root !== undefined + root !== undefined || + filesPending ) { - this.#pendingCleanup = { + const resources: PendingRecoveryCleanup = { root, process: cleanupOwner, tree: failedTree, }; + const files = this.#resources; + const lease = new AnalysisCleanupLease( + () => this.#releasePending(resources), + { + get cleanupResources() { + return [ + ...files.cleanupResources, + ...(resources.process === undefined + ? [] + : [ + resources.process.launch.ownership?.runId ?? + "wakaru-worker", + ]), + ...(resources.tree === undefined + ? [] + : [resources.tree.outputRoot]), + ...(resources.root === undefined ? [] : [resources.root]), + ]; + }, + }, + ); + this.#pendingCleanup = { resources, lease }; if (root !== undefined) residuals.push(root); root = undefined; } - if (residuals.length > 0) - return err( - new ProviderCleanupError( - "wakaru", - residuals, - { - ...(failure instanceof WakaruCleanupFailure - ? failure.diagnostics - : {}), - previous_error: - failure instanceof WakaruCleanupFailure - ? (failure.diagnostics?.previous_error ?? null) - : recoveryFailureMessage(failure), - cleanup_failures: cleanupFailures, - }, - { - operation: OPERATION, - cause: failure, - ...(tree?.published && execution !== undefined - ? { - partialObservation: { - kind: "javascript-recovery" as const, - result: javascriptRecoveryResultSchema.parse( - execution.result, - ), - }, - } - : {}), - }, - ), + if (residuals.length > 0) { + const error = new ProviderCleanupError( + "wakaru", + residuals, + { + ...(failure instanceof WakaruCleanupFailure + ? failure.diagnostics + : {}), + previous_error: + failure instanceof WakaruCleanupFailure + ? (failure.diagnostics?.previous_error ?? null) + : recoveryFailureMessage(failure), + cleanup_failures: cleanupFailures, + }, + { + operation: OPERATION, + cause: failure, + ...(tree?.published && execution !== undefined + ? { + partialObservation: { + kind: "javascript-recovery" as const, + result: javascriptRecoveryResultSchema.parse( + execution.result, + ), + }, + } + : {}), + }, ); + const lease = this.#pendingCleanup?.lease; + if (lease !== undefined) { + lease.bindCleanupError(error); + error.bindCleanupResourceOwner(lease); + } + return err(error); + } if (tree?.published === true && execution !== undefined) return ok(execution); return err( @@ -266,49 +311,36 @@ export class WakaruProvider implements JavaScriptRecoveryPort { async #retire(): Promise { const pending = this.#pendingCleanup; if (pending === undefined) return undefined; + try { + await pending.lease.release(); + this.#pendingCleanup = undefined; + return undefined; + } catch (cause: unknown) { + const error = new ProviderCleanupError( + "wakaru", + pending.lease.cleanupResources ?? [], + { reason: recoveryFailureMessage(cause) }, + { operation: OPERATION, cause }, + ).bindCleanupResourceOwner(pending.lease); + pending.lease.bindCleanupError(error); + return error; + } + } + + async #releasePending(pending: PendingRecoveryCleanup): Promise { if (pending.process !== undefined) { const stopped = await pending.process.stop(); - if (stopped.status === "incomplete") - return new ProviderCleanupError( - "wakaru", - [ - pending.process.launch.ownership?.runId ?? "wakaru-worker", - ...(pending.root === undefined ? [] : [pending.root]), - ], - { reason: stopped.reason }, - { operation: OPERATION }, - ); + if (stopped.status === "incomplete") throw new Error(stopped.reason); pending.process = undefined; } + await this.#resources.retryCleanup(); if (pending.tree !== undefined) { - const tree = pending.tree; - try { - await tree.rollback(); - pending.tree = undefined; - } catch (cause: unknown) { - return new ProviderCleanupError( - "wakaru", - [tree.outputRoot], - { reason: recoveryFailureMessage(cause) }, - { operation: OPERATION, cause }, - ); - } + await pending.tree.rollback(); + pending.tree = undefined; } if (pending.root !== undefined) { - const root = pending.root; - try { - await rm(root, { recursive: true, force: true }); - pending.root = undefined; - } catch (cause: unknown) { - return new ProviderCleanupError( - "wakaru", - [root], - { reason: recoveryFailureMessage(cause) }, - { operation: OPERATION, cause }, - ); - } + await rm(pending.root, { recursive: true, force: true }); + pending.root = undefined; } - this.#pendingCleanup = undefined; - return undefined; } } diff --git a/src/javascript/recovery/WakaruPublication.ts b/src/javascript/recovery/WakaruPublication.ts index fc17486813..a54a0be3dc 100644 --- a/src/javascript/recovery/WakaruPublication.ts +++ b/src/javascript/recovery/WakaruPublication.ts @@ -1,6 +1,7 @@ import { join } from "node:path"; import { Readable } from "node:stream"; import type { SafeOutputTree } from "../../artifacts/SafeOutputTree.js"; +import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js"; import { AnalysisOutputError } from "../../domain/analysisErrorCore.js"; import { javascriptRecoveryResultSchema, @@ -37,6 +38,7 @@ type PublicationContext = { admitted: Exclude; parsed: Awaited>; execution: Awaited>; + resources?: ArtifactResourceScope; signal?: AbortSignal; }; type PublishedFile = JavaScriptRecoveryResult["manifest"]; @@ -46,7 +48,7 @@ type RecoveryWriter = (path: string, bytes: Buffer) => Promise; export const publishWakaruArtifacts = async ( context: PublicationContext, ): Promise => { - const { tree, sources, parsed, staging, signal } = context; + const { tree, sources, parsed, staging, signal, resources } = context; const write = async (path: string, bytes: Buffer) => { const sha256 = recoveryDigest(bytes); const file = await tree.write( @@ -68,6 +70,7 @@ export const publishWakaruArtifacts = async ( source.snapshot_path, source.bytes, signal, + resources, ); if (recoveryDigest(snapshot) !== source.sha256) throw new AnalysisOutputError( @@ -94,6 +97,7 @@ export const publishWakaruArtifacts = async ( join(staging, "provenance.json"), RECOVERY_LIMITS.reportBytes, signal, + resources, ), ); const details = { @@ -161,13 +165,14 @@ const publishModule = async ( module: PublicationContext["parsed"]["report"]["modules"][number], write: RecoveryWriter, ): Promise => { - const { staging, sources, parsed, signal } = context; + const { staging, sources, parsed, signal, resources } = context; const artifact = await write( `modules/${module.filename}`, await readRecoveryFile( join(staging, module.filename), RECOVERY_LIMITS.outputBytes, signal, + resources, ), ); const mapPath = join(staging, `${module.filename}.map`); @@ -177,6 +182,7 @@ const publishModule = async ( mapPath, RECOVERY_LIMITS.outputBytes, signal, + resources, ).catch((cause: unknown) => { if (cause instanceof Error && "code" in cause && cause.code === "ENOENT") return null; diff --git a/src/javascript/recovery/WakaruReport.ts b/src/javascript/recovery/WakaruReport.ts index b9a828e37a..99fc44ff2b 100644 --- a/src/javascript/recovery/WakaruReport.ts +++ b/src/javascript/recovery/WakaruReport.ts @@ -1,4 +1,6 @@ import { z } from "zod"; +import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js"; +import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js"; import { normalizeArtifactPath, ArtifactPathRegistry, @@ -57,6 +59,8 @@ type ReportContext = { provenancePath: string; snapshots: ReadonlyMap; files: readonly { relativePath: string }[]; + resources?: ArtifactResourceScope; + signal?: AbortSignal; }; /** @@ -66,7 +70,7 @@ type ReportContext = { * checks below remain exact. */ export const parseWakaruReports = async (context: ReportContext) => { - const { execution, provenancePath } = context; + const { execution, provenancePath, resources, signal } = context; try { if (execution.exit_code !== 0 && execution.exit_code !== 1) @@ -76,7 +80,9 @@ export const parseWakaruReports = async (context: ReportContext) => { const report = reportSchema.parse(JSON.parse(execution.stdout)); const provenance = provenanceSchema.parse( JSON.parse( - (await readWakaruProvenanceReport(provenancePath)).toString("utf8"), + ( + await readWakaruProvenanceReport(provenancePath, signal, resources) + ).toString("utf8"), ), ); if (report.total !== report.modules.length || report.failed > report.total) @@ -101,6 +107,11 @@ export const parseWakaruReports = async (context: ReportContext) => { : ("complete" as const), }; } catch (cause: unknown) { + if ( + cause instanceof ArtifactReaderFailure && + (cause.cleanup !== undefined || cause.reason === "cancelled") + ) + throw cause; throw new ProviderAdapterError("wakaru", OPERATION, { cause, ...wakaruReportGuidance(cause), @@ -118,9 +129,18 @@ export const parseWakaruReports = async (context: ReportContext) => { const PROVENANCE_REPORT_ALLOWANCE = `Wakaru provenance report exceeds the ${String(RECOVERY_LIMITS.reportBytes)}-byte report allowance`; /** Report-size refusals name the fixed allowance; other read failures stay unchanged. */ -const readWakaruProvenanceReport = async (path: string): Promise => { +const readWakaruProvenanceReport = async ( + path: string, + signal: AbortSignal | undefined, + resources: ArtifactResourceScope | undefined, +): Promise => { try { - return await readRecoveryFile(path, RECOVERY_LIMITS.reportBytes); + return await readRecoveryFile( + path, + RECOVERY_LIMITS.reportBytes, + signal, + resources, + ); } catch (cause: unknown) { if (!isRecoveryFileSizeError(cause, RECOVERY_LIMITS.reportBytes)) throw cause; diff --git a/tests/boundary/javascript-recovery/wakaruFileOwnership.test.ts b/tests/boundary/javascript-recovery/wakaruFileOwnership.test.ts new file mode 100644 index 0000000000..7933622f69 --- /dev/null +++ b/tests/boundary/javascript-recovery/wakaruFileOwnership.test.ts @@ -0,0 +1,206 @@ +import { createHash } from "node:crypto"; +import { access, readFile, type FileHandle } from "node:fs/promises"; +import { basename, join } from "node:path"; +import { afterEach, expect, it, onTestFinished, vi } from "vitest"; +import { + ArtifactFileCancelled, + ArtifactFileCleanupFailure, +} from "../../../src/artifacts/ArtifactFileOperation.js"; +import { ProviderCleanupError } from "../../../src/domain/providerCleanupError.js"; +import { AnalysisCancelledError } from "../../../src/domain/analysisErrorCore.js"; +import { + assertRecoveryCleanup, + recoveryFixture, +} from "../../fixtures/javascript-recovery/provider.js"; + +const fileClose = vi.hoisted(() => ({ + selected: (_path: string): boolean => false, + denied: false, + deniedRemoval: undefined as string | undefined, + cancel: undefined as AbortController | undefined, + handles: [] as FileHandle[], +})); + +vi.mock("node:fs/promises", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + rm: async (...args: Parameters) => { + if (args[0] === fileClose.deniedRemoval) + throw Object.assign( + new Error("controlled recovery workspace removal failure"), + { code: "EACCES" }, + ); + return actual.rm(...args); + }, + open: async (...args: Parameters) => { + const handle = await actual.open(...args); + if (typeof args[0] === "string" && fileClose.selected(args[0])) { + const close = handle.close.bind(handle); + fileClose.handles.push(handle); + onTestFinished(async () => { + if (handle.fd >= 0) await close(); + }); + handle.close = async () => { + fileClose.cancel?.abort(); + if (fileClose.denied) + throw new Error("controlled Wakaru file close failure"); + await close(); + }; + } + return handle; + }, + }; +}); + +afterEach(() => { + fileClose.selected = () => false; + fileClose.denied = false; + fileClose.deniedRemoval = undefined; + fileClose.cancel = undefined; + fileClose.handles = []; +}); + +const linuxX64 = it.skipIf( + process.platform !== "linux" || process.arch !== "x64", +); + +linuxX64.each([ + "engine", + "input", + "provenance", + "module", + "source-map", +] as const)( + "retains %s file cleanup, blocks new recovery and resolves old errors after exact-owner retry", + async (stage) => { + const fixture = await recoveryFixture(); + onTestFinished(async () => { + fileClose.denied = false; + fileClose.deniedRemoval = undefined; + await fixture.service.close(); + }); + fileClose.selected = (path) => + stage === "engine" + ? path === join(fixture.root, "fixture-wakaru") + : stage === "input" + ? path === fixture.path + : stage === "provenance" + ? path.endsWith("/modules/provenance.json") + : stage === "module" + ? path.endsWith("/modules/module.js") + : path.endsWith("/modules/module.js.map"); + fileClose.denied = true; + const input = { path: fixture.path, output_directory: fixture.output }; + const outcome = await fixture.service.recover(input); + expect(outcome.ok, JSON.stringify(outcome)).toBe(false); + if (outcome.ok) return; + expect(outcome.error).toBeInstanceOf(ProviderCleanupError); + expect(outcome.error.cleanupIncomplete).toBe(true); + expect(outcome.error.cause).toBeInstanceOf(ArtifactFileCleanupFailure); + expect(fileClose.handles).toHaveLength(1); + const handle = fileClose.handles[0]!; + expect(handle.fd).toBeGreaterThanOrEqual(0); + const resources = outcome.error.cleanupResources; + const workspace = resources.find((path) => + basename(path).startsWith("rea-wakaru-"), + ); + if (stage !== "engine") { + expect(workspace).toBeDefined(); + await expect(access(workspace!)).resolves.toBeUndefined(); + } + const launches = fixture.launches.length; + const blocked = await fixture.service.recover(input); + expect(blocked.ok).toBe(false); + if (blocked.ok) return; + expect(blocked.error).toBeInstanceOf(ProviderCleanupError); + expect(blocked.error.cleanupIncomplete).toBe(true); + expect(fileClose.handles).toHaveLength(1); + expect(fixture.launches).toHaveLength(launches); + await expect(fixture.service.close()).rejects.toBeInstanceOf( + ProviderCleanupError, + ); + expect(handle.fd).toBeGreaterThanOrEqual(0); + fileClose.denied = false; + if (stage === "input") { + fileClose.deniedRemoval = workspace; + await expect(fixture.service.close()).rejects.toBeInstanceOf( + ProviderCleanupError, + ); + expect(handle.fd).toBe(-1); + expect(outcome.error.cleanupIncomplete).toBe(true); + expect(outcome.error.cleanupResources).toEqual([workspace]); + expect(blocked.error.cleanupIncomplete).toBe(true); + fileClose.deniedRemoval = undefined; + } + await fixture.service.close(); + expect(handle.fd).toBe(-1); + expect(outcome.error.cleanupIncomplete).toBe(false); + expect(outcome.error.cleanupResources).toEqual([]); + expect(blocked.error.cleanupIncomplete).toBe(false); + if (workspace !== undefined) + await expect(access(workspace)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(access(fixture.output)).rejects.toMatchObject({ + code: "ENOENT", + }); + await assertRecoveryCleanup(fixture.launches); + }, +); + +linuxX64.each([ + ["engine", false], + ["engine", true], + ["input", false], + ["input", true], +] as const)( + "preserves %s observation and cancellation arriving during final close (close denied: %s)", + async (stage, denied) => { + const fixture = await recoveryFixture(); + onTestFinished(async () => { + fileClose.denied = false; + fileClose.deniedRemoval = undefined; + await fixture.service.close(); + }); + const selected = + stage === "engine" ? join(fixture.root, "fixture-wakaru") : fixture.path; + const bytes = await readFile(selected); + fileClose.selected = (path) => path === selected; + fileClose.denied = denied; + const controller = new AbortController(); + fileClose.cancel = controller; + const result = await fixture.service.recover( + { path: fixture.path, output_directory: fixture.output }, + { signal: controller.signal }, + ); + expect(result.ok).toBe(false); + if (result.ok) return; + expect(result.error).toBeInstanceOf( + denied ? ProviderCleanupError : AnalysisCancelledError, + ); + const failure = result.error.cause; + const physical = + failure instanceof ArtifactFileCleanupFailure + ? failure.fileCleanupFailure + : failure instanceof ArtifactFileCancelled + ? failure.fileCancellation + : undefined; + expect(physical?.outcome).toMatchObject({ + kind: "completed", + value: { + bytes: bytes.length, + sha256: createHash("sha256").update(bytes).digest("hex"), + }, + }); + expect(fixture.launches).toHaveLength(0); + if (denied) { + expect(result.error.cleanupIncomplete).toBe(true); + fileClose.denied = false; + fileClose.deniedRemoval = undefined; + await fixture.service.close(); + expect(result.error.cleanupIncomplete).toBe(false); + } + await expect(access(fixture.output)).rejects.toMatchObject({ + code: "ENOENT", + }); + }, +);