diff --git a/.oxlintrc.json b/.oxlintrc.json index 08597e0f5a..886194eda5 100644 --- a/.oxlintrc.json +++ b/.oxlintrc.json @@ -107,6 +107,7 @@ "rea/provider-failure-classifier": "error", "rea/typed-resource-refusal": "error", "rea/no-babel-visitor-keys": "error", + "rea/no-raw-file-handle-close": "error", "typescript/no-misused-promises": "error" } }, @@ -121,6 +122,7 @@ "rea/provider-failure-classifier": "off", "rea/typed-resource-refusal": "off", "rea/no-babel-visitor-keys": "off", + "rea/no-raw-file-handle-close": "off", "max-lines": "off", "max-lines-per-function": [ "error", @@ -133,6 +135,10 @@ "typescript/no-misused-promises": "warn" } }, + { + "files": ["src/filesystem/Owned*.ts"], + "rules": { "rea/no-raw-file-handle-close": "off" } + }, { "files": ["src/domain/javascript/javascriptSemanticTraversal.ts"], "rules": { "rea/no-babel-visitor-keys": "off" } diff --git a/scripts/lint/index.mjs b/scripts/lint/index.mjs index bc9c565776..0de438f53a 100644 --- a/scripts/lint/index.mjs +++ b/scripts/lint/index.mjs @@ -9,6 +9,7 @@ import noLocaleCompare from "./rules/no-locale-compare.mjs"; import noRawCdpSession from "./rules/no-raw-cdp-session.mjs"; import noBabelVisitorKeys from "./rules/no-babel-visitor-keys.mjs"; import projectWaitDeadline from "./rules/project-wait-deadline.mjs"; +import noRawFileHandleClose from "./rules/no-raw-file-handle-close.mjs"; import providerFailureClassifier from "./rules/provider-failure-classifier.mjs"; import typedResourceRefusal from "./rules/typed-resource-refusal.mjs"; @@ -25,6 +26,7 @@ export default { "no-raw-cdp-session": noRawCdpSession, "no-babel-visitor-keys": noBabelVisitorKeys, "project-wait-deadline": projectWaitDeadline, + "no-raw-file-handle-close": noRawFileHandleClose, "provider-failure-classifier": providerFailureClassifier, "typed-resource-refusal": typedResourceRefusal, }, diff --git a/scripts/lint/rules/no-raw-file-handle-close.d.mts b/scripts/lint/rules/no-raw-file-handle-close.d.mts new file mode 100644 index 0000000000..86d61729e2 --- /dev/null +++ b/scripts/lint/rules/no-raw-file-handle-close.d.mts @@ -0,0 +1,4 @@ +import type { RuleTester } from "oxlint/plugins-dev"; + +declare const rule: Parameters[1]; +export default rule; diff --git a/scripts/lint/rules/no-raw-file-handle-close.mjs b/scripts/lint/rules/no-raw-file-handle-close.mjs new file mode 100644 index 0000000000..5f7f1a5133 --- /dev/null +++ b/scripts/lint/rules/no-raw-file-handle-close.mjs @@ -0,0 +1,244 @@ +import { findBinding } from "../bindings.mjs"; + +const FS_PROMISES = new Set(["node:fs/promises", "fs/promises"]); +const FS = new Set(["node:fs", "fs"]); + +const unwrap = (node) => { + let current = node; + while ( + current?.type === "AwaitExpression" || + current?.type === "TSAsExpression" || + current?.type === "TSNonNullExpression" || + current?.type === "TSSatisfiesExpression" || + current?.type === "ParenthesizedExpression" || + current?.type === "ChainExpression" + ) + current = current.argument ?? current.expression; + return current; +}; + +/** The module and imported name an identifier binds to, when it is an import. */ +const importedFrom = (sourceCode, identifier) => { + const definition = findBinding(sourceCode, identifier)?.defs?.[0]; + if (definition?.type !== "ImportBinding") return undefined; + const declaration = definition.parent; + const specifier = definition.node; + return { + source: declaration?.source?.value, + imported: + specifier.type === "ImportSpecifier" + ? (specifier.imported.name ?? specifier.imported.value) + : "*", + }; +}; + +/** `name(...)` or `namespace.name(...)` imported from one of `modules`. */ +const callsImport = (sourceCode, call, modules, names) => { + if (call?.type !== "CallExpression") return false; + const callee = unwrap(call.callee); + if (callee?.type === "Identifier") { + const binding = importedFrom(sourceCode, callee); + return ( + binding !== undefined && + modules.has(binding.source) && + names.has(binding.imported) + ); + } + if ( + callee?.type === "MemberExpression" && + !callee.computed && + names.has(callee.property.name) && + callee.object.type === "Identifier" + ) { + const binding = importedFrom(sourceCode, callee.object); + return ( + binding !== undefined && + binding.imported === "*" && + modules.has(binding.source) + ); + } + return false; +}; + +const OPEN = new Set(["open"]); +const DESCRIPTOR_CLOSE = new Set(["close", "closeSync"]); + +/** Whether an expression evaluates to a FileHandle from fs/promises open(). */ +const opensFileHandle = (sourceCode, expression) => { + const value = unwrap(expression); + if (callsImport(sourceCode, value, FS_PROMISES, OPEN)) return true; + // open(...).catch(...) still resolves to the opened handle. + return ( + value?.type === "CallExpression" && + value.callee.type === "MemberExpression" && + !value.callee.computed && + value.callee.property.name === "catch" && + callsImport(sourceCode, unwrap(value.callee.object), FS_PROMISES, OPEN) + ); +}; + +/** Whether a type annotation names fs/promises' FileHandle, alone or in a union. */ +const namesFileHandle = (sourceCode, annotation) => { + const type = annotation?.typeAnnotation ?? annotation; + if (type?.type === "TSUnionType") + return type.types.some((member) => namesFileHandle(sourceCode, member)); + if (type?.type !== "TSTypeReference") return false; + const name = type.typeName; + if (name?.type === "Identifier") { + const binding = importedFrom(sourceCode, name); + return ( + binding !== undefined && + FS_PROMISES.has(binding.source) && + binding.imported === "FileHandle" + ); + } + // fsp.FileHandle through a namespace import. + if (name?.type === "TSQualifiedName" && name.right.name === "FileHandle") { + const binding = + name.left.type === "Identifier" + ? importedFrom(sourceCode, name.left) + : undefined; + return ( + binding !== undefined && + binding.imported === "*" && + FS_PROMISES.has(binding.source) + ); + } + return false; +}; + +/** + * The class body whose instance (or, when `isStatic`, whose constructor) the + * `this` at `node` refers to. Arrow functions borrow the enclosing `this`; + * ordinary functions and object methods bind their own, which is no class. + */ +const thisClass = (node) => { + for (let current = node.parent; current; current = current.parent) { + if (current.type === "ArrowFunctionExpression") continue; + if ( + current.type === "FunctionExpression" || + current.type === "FunctionDeclaration" + ) { + const method = current.parent; + return method?.type === "MethodDefinition" && + method.value === current && + method.parent?.type === "ClassBody" + ? { body: method.parent, isStatic: method.static === true } + : undefined; + } + if ( + (current.type === "PropertyDefinition" || + current.type === "AccessorProperty") && + current.parent?.type === "ClassBody" + ) + return { body: current.parent, isStatic: current.static === true }; + if (current.type === "StaticBlock" && current.parent?.type === "ClassBody") + return { body: current.parent, isStatic: true }; + } + return undefined; +}; + +/** A class field declared as a FileHandle, for `this.field.close()`. */ +const fieldHoldsFileHandle = (sourceCode, member) => { + if ( + member.type !== "MemberExpression" || + member.computed || + member.object.type !== "ThisExpression" + ) + return false; + const property = member.property; + const owner = thisClass(member); + if (owner === undefined) return false; + return owner.body.body.some((element) => { + // Constructor parameter properties are public or TS-private names, never `#name`. + if ( + !owner.isStatic && + property.type === "Identifier" && + element.type === "MethodDefinition" && + element.kind === "constructor" + ) + return element.value.params.some((parameter) => { + if (parameter.type !== "TSParameterProperty") return false; + const value = parameter.parameter; + const id = value.type === "AssignmentPattern" ? value.left : value; + return ( + id.type === "Identifier" && + id.name === property.name && + namesFileHandle(sourceCode, id.typeAnnotation) + ); + }); + return ( + element.type === "PropertyDefinition" && + (element.static === true) === owner.isStatic && + element.key.type === property.type && + element.key.name === property.name && + (namesFileHandle(sourceCode, element.typeAnnotation) || + opensFileHandle(sourceCode, element.value)) + ); + }); +}; + +/** Whether `identifier` is a variable declared or ever assigned as a FileHandle. */ +const holdsFileHandle = (sourceCode, identifier) => { + const variable = findBinding(sourceCode, identifier); + if (variable === undefined) return false; + for (const definition of variable.defs ?? []) { + if (namesFileHandle(sourceCode, definition.name?.typeAnnotation)) + return true; + if ( + definition.node?.type === "VariableDeclarator" && + opensFileHandle(sourceCode, definition.node.init) + ) + return true; + } + for (const reference of variable.references ?? []) { + const parent = reference.identifier?.parent; + if ( + parent?.type === "AssignmentExpression" && + parent.left === reference.identifier && + opensFileHandle(sourceCode, parent.right) + ) + return true; + } + return false; +}; + +export default { + meta: { + type: "problem", + docs: { + description: + "Close file handles and descriptors only through the src/filesystem/Owned* owners.", + }, + schema: [], + messages: { + raw: "Close file handles and descriptors through OwnedFileHandle or withOwnedFileHandle (src/filesystem), which retain a failed close instead of losing it or masking the operation's own failure.", + }, + }, + create(context) { + const { sourceCode } = context; + return { + CallExpression(node) { + if (callsImport(sourceCode, node, FS, DESCRIPTOR_CLOSE)) { + context.report({ node, messageId: "raw" }); + return; + } + const callee = node.callee; + if ( + callee.type !== "MemberExpression" || + callee.computed || + callee.property.name !== "close" + ) + return; + const target = unwrap(callee.object); + if ( + (target?.type === "Identifier" && + holdsFileHandle(sourceCode, target)) || + fieldHoldsFileHandle(sourceCode, target) || + opensFileHandle(sourceCode, target) + ) + context.report({ node, messageId: "raw" }); + }, + }; + }, +}; diff --git a/src/artifacts/ArtifactResourceScope.test.ts b/src/artifacts/ArtifactResourceScope.test.ts index 5f6678c2b6..46beb1feeb 100644 --- a/src/artifacts/ArtifactResourceScope.test.ts +++ b/src/artifacts/ArtifactResourceScope.test.ts @@ -715,3 +715,111 @@ it.each(["artifact", "domain"] as const)( } }, ); + +it("tracks diagnostic work outside admission and joins it before a cleanup retry", async () => { + const scope = new ArtifactResourceScope(); + let earlierReleased = false; + scope.retain({ + kind: "lease", + resource: "earlier", + lease: { + release: async () => { + if (!earlierReleased) throw new Error("earlier close denied"); + }, + }, + }); + const started = deferred(); + const finish = deferred(); + let lateReleases = 0; + // A pending owner neither delays nor fails tracked work. + const sample = scope.track(async () => { + started.resolve(); + await finish.promise; + scope.retain({ + kind: "lease", + resource: "sampled", + lease: { + release: async () => { + lateReleases += 1; + }, + }, + }); + return "sample"; + }); + try { + await awaitAcknowledgement(started.promise, sample, "tracked work"); + earlierReleased = true; + let retried = false; + const retry = scope.retryCleanup().then(() => { + retried = true; + }); + await Promise.resolve(); + expect(retried).toBe(false); + finish.resolve(); + await expect(sample).resolves.toBe("sample"); + await retry; + // The drain joined the tracked work, so the owner it retained was retried. + expect(lateReleases).toBe(1); + expect(scope.cleanupResources).toEqual([]); + } finally { + earlierReleased = true; + finish.resolve(); + await scope.close(); + } +}); + +it("joins diagnostic work admitted while a cleanup retry releases an earlier owner", async () => { + const scope = new ArtifactResourceScope(); + const releasing = deferred(); + const allowRelease = deferred(); + scope.retain({ + kind: "lease", + resource: "earlier", + lease: { + release: async () => { + releasing.resolve(); + await allowRelease.promise; + }, + }, + }); + let lateReleases = 0; + const started = deferred(); + const finish = deferred(); + let retried = false; + const retry = scope.retryCleanup().then(() => { + retried = true; + }); + let sample: Promise | undefined; + try { + await awaitAcknowledgement(releasing.promise, retry, "earlier release"); + // Admitted after the drain already found no active work. + sample = scope.track(async () => { + started.resolve(); + await finish.promise; + scope.retain({ + kind: "lease", + resource: "sampled", + lease: { + release: async () => { + lateReleases += 1; + }, + }, + }); + return "sample"; + }); + await awaitAcknowledgement(started.promise, sample, "tracked work"); + allowRelease.resolve(); + await new Promise((resolve) => setImmediate(resolve)); + expect(retried).toBe(false); + finish.resolve(); + await expect(sample).resolves.toBe("sample"); + await retry; + expect(lateReleases).toBe(1); + expect(scope.hasPending).toBe(false); + } finally { + allowRelease.resolve(); + finish.resolve(); + await Promise.allSettled(sample === undefined ? [retry] : [retry, sample]); + await scope.close(); + } +}); diff --git a/src/artifacts/ArtifactResourceScope.ts b/src/artifacts/ArtifactResourceScope.ts index 723022c372..8c8e53f798 100644 --- a/src/artifacts/ArtifactResourceScope.ts +++ b/src/artifacts/ArtifactResourceScope.ts @@ -136,13 +136,33 @@ export class ArtifactResourceScope { await this.retryCleanup(); continue; } - const running = Promise.resolve().then(operation); - this.#active.add(running); - try { - return await running; - } finally { - this.#active.delete(running); - } + return this.#track(operation); + } + } + + /** + * Run optional work, such as diagnostic sampling, that may retain owners but + * must not wait for or fail on earlier pending cleanup. Cleanup drains still + * join it, so an owner it retains is not left behind by a finished drain. + */ + track(operation: () => Promise): Promise { + if (this.#closed) + return Promise.reject( + new ArtifactReaderFailure( + "unavailable", + "Artifact resource scope is closed", + ), + ); + return this.#track(operation); + } + + async #track(operation: () => Promise): Promise { + const running = Promise.resolve().then(operation); + this.#active.add(running); + try { + return await running; + } finally { + this.#active.delete(running); } } @@ -307,20 +327,29 @@ export class ArtifactResourceScope { } async #drainAndRetry(): Promise { - while (this.#active.size > 0) await Promise.allSettled(this.#active); - const failures: { - readonly owner: ArtifactResourceOwner; - readonly cause: unknown; - }[] = []; - for (const owner of this.#pending) { - try { - await this.#lease(owner).release(); - this.#pending.delete(owner); - } catch (cause: unknown) { - this.#observeCleanupFailure(owner, cause); - failures.push({ owner, cause }); + const failed = new Map(); + // Tracked work may start while an earlier owner is being released, and may + // retain owners of its own; repeat until no work is active and every + // retained owner has had one attempt in this drain. + while (true) { + while (this.#active.size > 0) await Promise.allSettled(this.#active); + for (const owner of this.#pending) { + if (failed.has(owner)) continue; + try { + await this.#lease(owner).release(); + this.#pending.delete(owner); + } catch (cause: unknown) { + this.#observeCleanupFailure(owner, cause); + failed.set(owner, cause); + } } + if ( + this.#active.size === 0 && + [...this.#pending].every((owner) => failed.has(owner)) + ) + break; } + const failures = [...failed].map(([owner, cause]) => ({ owner, cause })); if (failures.length === 0) return; const observations = failures.map(({ owner, cause }) => ArtifactReaderFailure.cleanupObservation( diff --git a/src/desktop/AgentEnvComputerHelper.cleanup.test.ts b/src/desktop/AgentEnvComputerHelper.cleanup.test.ts new file mode 100644 index 0000000000..9faf7e559e --- /dev/null +++ b/src/desktop/AgentEnvComputerHelper.cleanup.test.ts @@ -0,0 +1,56 @@ +import { writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { expect, it, vi } from "vitest"; + +import { ArtifactFileCleanupFailure } from "../artifacts/ArtifactFileOperation.js"; +import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js"; +import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; +import { createTestTempDirectory } from "../../tests/fixtures/temporaryDirectory.js"; +import { agentEnvInterpreter } from "./AgentEnvComputerHelper.js"; + +it("retains a failed runner descriptor close for retry by the caller", async () => { + const directory = await createTestTempDirectory("rea-agentenv-close-"); + const path = join(directory, "agent-env"); + await writeFile(path, "#!/usr/bin/python3\n"); + const resources = new ArtifactResourceScope(); + const close = vi + .spyOn(OwnedFileHandle.prototype, "close") + .mockRejectedValueOnce(new Error("close denied")); + try { + let failure: unknown; + try { + await agentEnvInterpreter( + { REA_AGENT_ENV_COMMAND: path }, + "desktop_build_doctor", + resources, + ); + } catch (cause: unknown) { + failure = cause; + } + expect(failure).toBeInstanceOf(ArtifactFileCleanupFailure); + expect(resources.cleanupResources).toContain(path); + await resources.retryCleanup(); + expect(resources.cleanupResources).toEqual([]); + // The reported failure shares the retained lease, so the retry resolves it. + expect( + failure instanceof ArtifactFileCleanupFailure && + failure.cleanupLease.released, + ).toBe(true); + close.mockRejectedValueOnce(new Error("close denied without scope")); + let unscoped: unknown; + try { + await agentEnvInterpreter( + { REA_AGENT_ENV_COMMAND: path }, + "check_build_task", + ); + } catch (cause: unknown) { + unscoped = cause; + } + expect(unscoped).toBeInstanceOf(ArtifactFileCleanupFailure); + if (unscoped instanceof ArtifactFileCleanupFailure) + await unscoped.cleanupLease.release(); + } finally { + close.mockRestore(); + await resources.retryCleanup(); + } +}); diff --git a/src/desktop/AgentEnvComputerHelper.ts b/src/desktop/AgentEnvComputerHelper.ts index b16f3688c6..8e3330c8c2 100644 --- a/src/desktop/AgentEnvComputerHelper.ts +++ b/src/desktop/AgentEnvComputerHelper.ts @@ -17,6 +17,11 @@ import { spawnOwnedProviderProcess, type SpawnedOwnedProviderProcess, } from "../process/ProviderProcess.js"; +import { + ArtifactFileCleanupFailure, + withArtifactFileHandle, +} from "../artifacts/ArtifactFileOperation.js"; +import type { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js"; const execFileAsync = promisify(execFile); @@ -67,15 +72,27 @@ const executable = async (path: string): Promise => () => false, ); -const readHead = async (path: string): Promise => { - const file = await open(path, "r"); - try { - const { buffer, bytesRead } = await file.read(Buffer.alloc(512), 0, 512, 0); - return buffer.subarray(0, bytesRead).toString("utf8"); - } finally { - await file.close(); - } -}; +const readHead = async ( + path: string, + resources?: ArtifactResourceScope, +): Promise => + // The artifact file owner binds one cleanup lease to both the retained owner + // and the failure it reports, so a later successful retry resolves both. + withArtifactFileHandle( + path, + await open(path, "r"), + resources, + undefined, + async (file) => { + const { buffer, bytesRead } = await file.read( + Buffer.alloc(512), + 0, + 512, + 0, + ); + return buffer.subarray(0, bytesRead).toString("utf8"); + }, + ); /** * The Python the agentenv runner runs with, so the helper can import @@ -85,6 +102,7 @@ const readHead = async (path: string): Promise => { export const agentEnvInterpreter = async ( environment: Readonly, operation: string, + resources?: ArtifactResourceScope, ): Promise => { const command = environment.REA_AGENT_ENV_COMMAND; if (command === undefined || !isAbsolute(command)) @@ -96,8 +114,9 @@ export const agentEnvInterpreter = async ( let head: string; try { program = await realpath(command); - head = await readHead(program); + head = await readHead(program, resources); } catch (cause: unknown) { + if (cause instanceof ArtifactFileCleanupFailure) throw cause; throw unavailable( operation, `REA_AGENT_ENV_COMMAND ${command} cannot be read.`, diff --git a/src/desktop/DesktopBuildDoctorFacts.test.ts b/src/desktop/DesktopBuildDoctorFacts.test.ts index 6a59433e12..f716538266 100644 --- a/src/desktop/DesktopBuildDoctorFacts.test.ts +++ b/src/desktop/DesktopBuildDoctorFacts.test.ts @@ -12,13 +12,18 @@ import { import { join } from "node:path"; import { promisify } from "node:util"; -import { expect, it } from "vitest"; +import { expect, it, vi } from "vitest"; import { createTestTempDirectory } from "../../tests/fixtures/temporaryDirectory.js"; import { runDoctor, systemDoctorHost } from "../application/Doctor.js"; import { readDesktopBuildFacts } from "./DesktopBuildDoctorFacts.js"; +import { ArtifactFileCleanupFailure } from "../artifacts/ArtifactFileOperation.js"; import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js"; +import type { ClientRegistrationStatus } from "../application/ClientRegistrationStatus.js"; +import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; import { AnalysisCancelledError } from "../domain/analysisErrorCore.js"; +import { ProviderCleanupError } from "../domain/providerCleanupError.js"; +import { awaitAcknowledgement } from "../../tests/fixtures/lifecycleAcknowledgement.js"; import { PRODUCT_IDENTITY } from "../identity.js"; import { defaultBuildComputerArchive, @@ -794,3 +799,116 @@ it("recognizes a gzip runtime archive with a different top folder", async () => status: "current", }); }); + +it.skipIf(process.platform === "win32")( + "settles every client inspection before reporting a retained runner close failure", + async () => { + const root = await createTestTempDirectory("rea-doctor-peer-settle-"); + const failing = join(root, "agent-env-a"); + const slow = join(root, "agent-env-b"); + const archive = join(root, "runtime.tar.gz"); + await program(failing, "exit 0"); + await program(slow, "exit 0"); + await writeFile(archive, "archive"); + const resources = new ArtifactResourceScope(); + let releaseSlow!: () => void; + const slowReleased = new Promise((resolve) => { + releaseSlow = resolve; + }); + let reachSlow!: () => void; + const slowReached = new Promise((resolve) => { + reachSlow = resolve; + }); + let failClose!: () => void; + const closeFailed = new Promise((resolve) => { + failClose = resolve; + }); + const close = vi + .spyOn(OwnedFileHandle.prototype, "close") + .mockImplementationOnce(async () => { + failClose(); + throw new Error("close denied"); + }); + const recorded: Record> = { + a: { + REA_AGENT_ENV_COMMAND: failing, + REA_BUILD_COMPUTER_REA_ARCHIVE: archive, + }, + b: { REA_AGENT_ENV_COMMAND: slow }, + }; + const status = (client: string, differing: readonly string[]) => + ({ + client, + command: [process.execPath, "rea.mjs", "mcp"], + provider_settings: { differing, unrecorded: [], unresolved: [] }, + }) as unknown as ClientRegistrationStatus; + let outcome: Promise = Promise.resolve(); + try { + let settled = false; + outcome = readDesktopBuildFacts({ + environment: { HOME: root, PATH: "" }, + platform: process.platform, + resources, + signal: new AbortController().signal, + run: async (command) => { + if (command === slow) { + reachSlow(); + await slowReleased; + } + const stdout = "agent-env 0.9.1431\n"; + return { + stdout, + stderr: "", + stdoutBytes: Buffer.byteLength(stdout), + stderrBytes: 0, + }; + }, + clientIds: ["a", "b"], + registrations: async (_clients, _settings, _owned, observe) => { + for (const [client, values] of Object.entries(recorded)) + for (const [name, value] of Object.entries(values)) + observe?.(client, name, value); + return [ + status("a", ["REA_BUILD_COMPUTER_REA_ARCHIVE"]), + status("b", []), + ]; + }, + }).then( + () => { + settled = true; + return undefined; + }, + (cause: unknown) => { + settled = true; + return cause; + }, + ); + await awaitAcknowledgement( + Promise.all([slowReached, closeFailed]).then(() => undefined), + outcome, + "slow client probe and failed runner close", + ); + for (let turn = 0; turn < 5; turn += 1) + await new Promise((resolve) => setImmediate(resolve)); + // The failed close must not report while client b is still inspecting. + expect(settled).toBe(false); + releaseSlow(); + // The doctor reports a typed cleanup failure naming the retained runner + // file, so the CLI and MCP error documents keep it and its recovery. + const failure = await outcome; + expect(failure).toBeInstanceOf(ProviderCleanupError); + const cleanup = failure as ProviderCleanupError; + expect(cleanup.operation).toBe("doctor"); + expect(cleanup.cause).toBeInstanceOf(ArtifactFileCleanupFailure); + expect(resources.cleanupResources).toHaveLength(1); + expect(cleanup.cleanupResources).toEqual(resources.cleanupResources); + } finally { + releaseSlow(); + // Join the inspection before restoring the close it observes. + await outcome; + close.mockRestore(); + await resources.retryCleanup(); + await resources.close(); + } + }, +); diff --git a/src/desktop/DesktopBuildDoctorFacts.ts b/src/desktop/DesktopBuildDoctorFacts.ts index 8b0f94a321..b671e946c4 100644 --- a/src/desktop/DesktopBuildDoctorFacts.ts +++ b/src/desktop/DesktopBuildDoctorFacts.ts @@ -24,6 +24,12 @@ import { PRODUCT_IDENTITY } from "../identity.js"; import { execFileOutputFailure } from "../process/ExecFileOutput.js"; import { describeSystemPs, resolveSystemPs } from "../process/systemPs.js"; import { agentEnvInterpreter } from "./AgentEnvComputerHelper.js"; +import { ArtifactFileCleanupFailure } from "../artifacts/ArtifactFileOperation.js"; +import { + AnalysisCleanupLeaseGroup, + type AnalysisCleanupLease, +} from "../artifacts/AnalysisCleanupLease.js"; +import { ProviderCleanupError } from "../domain/providerCleanupError.js"; import { agentEnvReportedVersion, defaultBuildComputerArchive, @@ -90,7 +96,12 @@ export const readDesktopBuildFacts = async ( ? agentEnvInterpreter( { ...input.environment, REA_AGENT_ENV_COMMAND: agentEnv.path }, "desktop_build_doctor", - ).catch(() => "python3") + input.resources, + ).catch((cause: unknown) => { + if (cause instanceof ArtifactFileCleanupFailure) + throw runnerCleanupFailure(cause); + return "python3"; + }) : "python3", ); const chosenCommand = ( @@ -367,7 +378,9 @@ const registrationFacts = async ( ); const registered = statuses.filter(({ command }) => command.length > 0); const considered = selected ?? registered.map(({ client }) => client); - return Promise.all( + // Every admitted client inspection settles before a failure is reported, so + // no file a peer opened outlives the doctor operation that owns it. + const outcomes = await Promise.allSettled( considered.map(async (client) => { const status = registered.find((entry) => entry.client === client); const values = status === undefined ? undefined : recorded.get(client); @@ -440,7 +453,12 @@ const registrationFacts = async ( : agentEnvInterpreter( { ...environment, REA_AGENT_ENV_COMMAND: effectiveRunner }, "desktop_build_doctor", - ).catch(() => undefined), + input.resources, + ).catch((cause: unknown) => { + if (cause instanceof ArtifactFileCleanupFailure) + throw runnerCleanupFailure(cause); + return undefined; + }), ); if (archive.status === "unknown") unchecked.push(name); else if (archive.status !== "current") compared.push(name); @@ -457,6 +475,11 @@ const registrationFacts = async ( }; }), ); + const failure = outcomes.find((outcome) => outcome.status === "rejected"); + if (failure !== undefined) throw failure.reason; + return outcomes.flatMap((outcome) => + outcome.status === "fulfilled" ? [outcome.value] : [], + ); }; const solveEndpoint = async ( @@ -523,6 +546,38 @@ const solveEndpoint = async ( : (selected ?? "agentenv model configuration"); }; +/** + * A doctor file whose close failed stays owned by `input.resources`, which the + * doctor host retries on close. Report it as the doctor's typed cleanup + * failure, bound to that owner's lease, so the error document keeps the + * retained file and its recovery. + */ +const doctorFileCleanupFailure = ( + lease: AnalysisCleanupLease, + resources: readonly string[], + reason: string, + cause: unknown, +): ProviderCleanupError => { + const failure = new ProviderCleanupError( + "rea", + resources, + { reason }, + { operation: "doctor", cause }, + ); + new AnalysisCleanupLeaseGroup(failure, [lease]); + return failure; +}; + +const runnerCleanupFailure = ( + cause: ArtifactFileCleanupFailure, +): ProviderCleanupError => + doctorFileCleanupFailure( + cause.cleanupLease, + cause.cleanup?.resources ?? cause.cleanupLease.cleanupResources ?? [], + cause.cleanup?.reason ?? cause.message, + cause, + ); + const canReadModelConfig = async ( path: string, input: DesktopBuildFactsInput, @@ -545,7 +600,12 @@ const canReadModelConfig = async ( } catch (cause: unknown) { if (cause instanceof RetainedRegularFileCleanupFailure) { input.resources.retain(cause.cleanupOwner); - throw cause; + throw doctorFileCleanupFailure( + cause.cleanupLease, + [cause.path], + cause.message, + cause, + ); } if (cause instanceof AnalysisCancelledError) throw cause; if (input.signal.aborted) throw cancelledBy("doctor", input.signal); diff --git a/src/ghidra/GhidraClient.ts b/src/ghidra/GhidraClient.ts index 30604e1db8..ae8e981ab7 100644 --- a/src/ghidra/GhidraClient.ts +++ b/src/ghidra/GhidraClient.ts @@ -1667,8 +1667,23 @@ export class GhidraClient { }); } + /** + * Optional diagnostic sampling, outside the scope's admission gate: a + * pending owner must not mask the provider failure being explained. The + * scope still tracks the read, so session cleanup joins it and reports a + * failed close it retains. + */ #readMemoryCgroup(): Promise { - return (this.#options.readMemoryCgroup ?? readLinuxCgroupMemoryState)(); + return this.#fileResources.track(() => + (this.#options.readMemoryCgroup ?? readLinuxCgroupMemoryState)({ + retainFailedClose: (path, handle) => + this.#fileResources.retain({ + kind: "file-handle", + resource: path, + handle, + }), + }), + ); } /** diff --git a/src/ghidra/GhidraClientTypes.ts b/src/ghidra/GhidraClientTypes.ts index 404382864b..26cf22dcc4 100644 --- a/src/ghidra/GhidraClientTypes.ts +++ b/src/ghidra/GhidraClientTypes.ts @@ -3,6 +3,7 @@ import type { GhidraCheckpointSelection } from "./GhidraCheckpoint.js"; import type { BridgeMaxRequestBytes } from "../config/bridgeMaxRequestBytes.js"; import type { Result } from "../domain/result.js"; import type { LinuxCgroupMemoryState } from "../process/LinuxCgroupMemory.js"; +import type { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; import type { GhidraLauncher } from "./GhidraLauncher.js"; import type { GhidraSessionError } from "./GhidraSessionError.js"; import type { GhidraSessionInfo } from "./GhidraSessionValues.js"; @@ -56,8 +57,13 @@ export interface GhidraClientOptions { readonly readStartupTimeoutMs?: () => number; readonly onDiagnostic?: (event: GhidraDiagnostic) => void; readonly logger?: Logger; - /** Observes REA's memory cgroup around a run; defaults to the Linux cgroup v2 reader. */ - readonly readMemoryCgroup?: () => Promise; + /** + * Observes REA's memory cgroup around a run; defaults to the Linux cgroup v2 + * reader. A file whose close fails is handed to `retainFailedClose`. + */ + readonly readMemoryCgroup?: (options: { + readonly retainFailedClose: (path: string, owner: OwnedFileHandle) => void; + }) => Promise; /** Effective request ceiling; defaults to the configured bridge ceiling. */ readonly bridgeMaxRequestBytes?: BridgeMaxRequestBytes; } diff --git a/src/javascript/analysis/JavaScriptAnalysisTransfer.ts b/src/javascript/analysis/JavaScriptAnalysisTransfer.ts index 5bd990be47..80e2befdb8 100644 --- a/src/javascript/analysis/JavaScriptAnalysisTransfer.ts +++ b/src/javascript/analysis/JavaScriptAnalysisTransfer.ts @@ -24,6 +24,8 @@ import { javaScriptWorkerModuleRecordSchema, javaScriptWorkerProjectionHeaderSchema, } from "./JavaScriptAnalysisWorkerSchemas.js"; +import { OwnedFileHandle } from "../../filesystem/OwnedFileHandle.js"; +import { withOwnedFileHandle } from "../../filesystem/OwnedRegularFile.js"; // Budgets bound transfer representation, independently of source length. const FRAME_BYTES = JAVASCRIPT_ANALYSIS_TRANSFER_RECORD_BYTES; @@ -195,7 +197,8 @@ export const writeJavaScriptAnalysisTransfer = async ( transferBytes = TRANSFER_BYTES, ): Promise => { assertTransferByteBudget(transferBytes); - const output = await open(path, "wx", 0o600); + const owner = new OwnedFileHandle(await open(path, "wx", 0o600)); + const output = owner.handle; const digest = createHash("sha256"); let bytes = 0; let records = 0; @@ -237,7 +240,7 @@ export const writeJavaScriptAnalysisTransfer = async ( batchBytes += size; if (batchBytes >= BATCH_BYTES) await flush(); }; - try { + return withOwnedFileHandle(path, owner, {}, async () => { await append({ kind: "projection-header", value: { @@ -270,9 +273,7 @@ export const writeJavaScriptAnalysisTransfer = async ( graph_bytes: graphBytes, graph_records: graphRecords, }; - } finally { - await output.close(); - } + }); }; /** Decode producer records into realm-local, source-bound facts; sealing follows in the consumer. */ diff --git a/src/process/LinuxCgroupMemory.test.ts b/src/process/LinuxCgroupMemory.test.ts index e69862beb8..78c3818ea0 100644 --- a/src/process/LinuxCgroupMemory.test.ts +++ b/src/process/LinuxCgroupMemory.test.ts @@ -1,9 +1,51 @@ import { mkdir, writeFile } from "node:fs/promises"; import { join } from "node:path"; -import { expect, it } from "vitest"; +import { expect, it, vi } from "vitest"; import { createTestTempDirectory } from "../../tests/fixtures/temporaryDirectory.js"; import { readLinuxCgroupMemoryState } from "./LinuxCgroupMemory.js"; +import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; +import { RegularFileCleanupFailure } from "../filesystem/OwnedRegularFile.js"; + +it("retains a failed kernel file close for the caller to retry", async () => { + const directory = await createTestTempDirectory("rea-cgroup-retained-"); + const path = join(directory, "cgroup"); + await writeFile(path, "0::/jobs/rea\n"); + const failure = new Error("close denied"); + const close = vi + .spyOn(OwnedFileHandle.prototype, "close") + .mockRejectedValueOnce(failure); + let retained: OwnedFileHandle | undefined; + try { + await expect( + readLinuxCgroupMemoryState({ + platform: "linux", + procCgroupPath: path, + retainFailedClose: (_path, owner) => { + retained = owner; + }, + }), + ).resolves.toBeUndefined(); + expect(retained).toBeDefined(); + await retained?.close(); + close.mockRejectedValueOnce(failure); + let unretained: unknown; + try { + await readLinuxCgroupMemoryState({ + platform: "linux", + procCgroupPath: path, + }); + } catch (cause: unknown) { + unretained = cause; + } + expect(unretained).toBeInstanceOf(RegularFileCleanupFailure); + if (unretained instanceof RegularFileCleanupFailure) + await unretained.owner.close(); + } finally { + close.mockRestore(); + await retained?.close(); + } +}); const events = (oomKill: number) => `low 0\nhigh 0\nmax 772\noom 3\noom_kill ${String(oomKill)}\noom_group_kill 0\n`; diff --git a/src/process/LinuxCgroupMemory.ts b/src/process/LinuxCgroupMemory.ts index e8a780c70a..6224b0ec2c 100644 --- a/src/process/LinuxCgroupMemory.ts +++ b/src/process/LinuxCgroupMemory.ts @@ -1,5 +1,10 @@ import { open } from "node:fs/promises"; import { posix } from "node:path"; +import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; +import { + RegularFileCleanupFailure, + withOwnedFileHandle, +} from "../filesystem/OwnedRegularFile.js"; /** Memory facts for the cgroup v2 group that holds this process. */ export interface LinuxCgroupMemoryState { @@ -131,11 +136,14 @@ export const readLinuxCgroupMemoryState = async ( readonly platform?: NodeJS.Platform; readonly procCgroupPath?: string; readonly procMountInfoPath?: string; + readonly retainFailedClose?: (path: string, owner: OwnedFileHandle) => void; } = {}, ): Promise => { if ((options.platform ?? process.platform) !== "linux") return undefined; const membership = await readKernelFile( options.procCgroupPath ?? "/proc/self/cgroup", + undefined, + options.retainFailedClose, ); const path = membership === undefined ? undefined : parseLinuxCgroupV2Path(membership); @@ -143,17 +151,26 @@ export const readLinuxCgroupMemoryState = async ( const mountInfo = await readKernelFile( options.procMountInfoPath ?? "/proc/self/mountinfo", MOUNT_INFO_BYTES, + options.retainFailedClose, ); const group = mountInfo === undefined ? undefined : linuxCgroupDirectory(parseLinuxCgroup2Mounts(mountInfo), path); if (group === undefined) return undefined; - const events = await readKernelFile(posix.join(group, "memory.events")); + const events = await readKernelFile( + posix.join(group, "memory.events"), + undefined, + options.retainFailedClose, + ); const oomKills = events === undefined ? undefined : parseLinuxCgroupOomKills(events); if (oomKills === undefined) return undefined; - const limit = await readKernelFile(posix.join(group, "memory.max")); + const limit = await readKernelFile( + posix.join(group, "memory.max"), + undefined, + options.retainFailedClose, + ); return { path, oomKills, @@ -169,28 +186,38 @@ export const readLinuxCgroupMemoryState = async ( const readKernelFile = async ( path: string, maxBytes = KERNEL_FILE_BYTES, + retainFailedClose?: (path: string, owner: OwnedFileHandle) => void, ): Promise => { try { - const handle = await open(path, "r"); - try { - const buffer = Buffer.alloc(maxBytes + 1); - let length = 0; - for (;;) { - const { bytesRead } = await handle.read( - buffer, - length, - buffer.length - length, - null, - ); - if (bytesRead === 0) break; - length += bytesRead; - if (length > maxBytes) return undefined; - } - return buffer.subarray(0, length).toString("utf8"); - } finally { - await handle.close(); - } - } catch { + return await withOwnedFileHandle( + path, + new OwnedFileHandle(await open(path, "r")), + retainFailedClose === undefined + ? {} + : { retainFailedClose: (owner) => retainFailedClose(path, owner) }, + async (handle) => { + const buffer = Buffer.alloc(maxBytes + 1); + let length = 0; + for (;;) { + const { bytesRead } = await handle.read( + buffer, + length, + buffer.length - length, + null, + ); + if (bytesRead === 0) break; + length += bytesRead; + if (length > maxBytes) return undefined; + } + return buffer.subarray(0, length).toString("utf8"); + }, + ); + } catch (cause: unknown) { + if ( + cause instanceof RegularFileCleanupFailure && + cause.retention.kind !== "callback_returned" + ) + throw cause; return undefined; } }; diff --git a/tests/boundary/providers/ghidra/ghidraOomRetention.test.ts b/tests/boundary/providers/ghidra/ghidraOomRetention.test.ts index 2149af8c48..bd44ace4a3 100644 --- a/tests/boundary/providers/ghidra/ghidraOomRetention.test.ts +++ b/tests/boundary/providers/ghidra/ghidraOomRetention.test.ts @@ -1,5 +1,5 @@ import { createHash } from "node:crypto"; -import { writeFile } from "node:fs/promises"; +import { writeFile, type FileHandle } from "node:fs/promises"; import { join } from "node:path"; import { describe, expect, it, onTestFinished } from "vitest"; @@ -7,6 +7,7 @@ import { parseConfig } from "../../../../src/config/parseConfig.js"; import { analysisErrorProjectionSchema } from "../../../../src/contracts/errorSchemas.js"; import { projectAnalysisError } from "../../../../src/domain/analysisErrorProjection.js"; import { elf } from "../../../../src/domain/binaryTarget.fixture.js"; +import { OwnedFileHandle } from "../../../../src/filesystem/OwnedFileHandle.js"; import type { BinaryTarget } from "../../../../src/domain/binaryTargetTypes.js"; import { ok } from "../../../../src/domain/result.js"; import { GhidraClient } from "../../../../src/ghidra/GhidraClient.js"; @@ -242,6 +243,35 @@ describe.skipIf(process.platform === "win32")( expect(projected.remediation.action).not.toContain("rea doctor"); }); + it("keeps the cgroup evidence while a sampled file's failed close is retained", async () => { + let released = false; + const stuck = new OwnedFileHandle({ + fd: 1_000_000, + close: async () => { + if (!released) throw new Error("close denied"); + }, + } as unknown as FileHandle); + let reads = 0; + const projected = await failedOperation("sigkill_tools", { + readMemoryCgroup: async ({ retainFailedClose }) => { + // The post-kill sample leaves an owner pending for session cleanup. + if (reads === 1) retainFailedClose("/proc/self/cgroup", stuck); + return { path: "/", oomKills: reads++, memoryMax: 3_221_225_472 }; + }, + }); + released = true; + expect(reads).toBe(2); + expect(projected).toMatchObject({ + code: "resource_constraint", + details: { + diagnostics: { + exit_signal: "SIGKILL", + memory_cgroup: { oom_kill_before: 0, oom_kill_after: 1 }, + }, + }, + }); + }); + it("names the signal when no cgroup attributes the kill", async () => { const projected = await failedOperation("sigkill_tools"); expect(projected.code).toBe("execution_failure"); diff --git a/tests/evaluation/noRawFileHandleCloseRule.test.ts b/tests/evaluation/noRawFileHandleCloseRule.test.ts new file mode 100644 index 0000000000..21caa76c9e --- /dev/null +++ b/tests/evaluation/noRawFileHandleCloseRule.test.ts @@ -0,0 +1,82 @@ +import { describe, it } from "vitest"; +import { RuleTester } from "oxlint/plugins-dev"; +import rule from "../../scripts/lint/rules/no-raw-file-handle-close.mjs"; + +RuleTester.describe = describe; +RuleTester.it = it; +new RuleTester({ languageOptions: { parserOptions: { lang: "ts" } } }).run( + "no-raw-file-handle-close", + rule, + { + valid: [ + // Owned closes, unrelated close methods and shadowed names. + 'import { open } from "node:fs/promises"; const owner = new OwnedFileHandle(await open("x")); await owner.close();', + "server.close(); transport.close();", + 'import { open } from "./local.js"; const h = await open("x"); await h.close();', + 'const open = async (p: string) => ({ close() {} }); const h = await open("x"); h.close();', + 'import type { FileHandle } from "./local.js"; const f = (h: FileHandle) => h.close();', + "class C { readonly x: { close(): void }; d() { this.x.close(); } }", + 'import type { FileHandle } from "node:fs/promises"; class C { static h: FileHandle; h = { close() {} }; d() { this.h.close(); } }', + 'import { close } from "./local.js"; close(1);', + // `#h` and a constructor parameter property `h` are distinct fields. + 'import type { FileHandle } from "node:fs/promises"; class C { #h = { close() {} }; constructor(private h: FileHandle) {} d() { this.#h.close(); } }', + // Object methods and ordinary functions bind their own `this`. + 'import type { FileHandle } from "node:fs/promises"; class C { h!: FileHandle; d() { const other = { h: { close() {} }, d() { this.h.close(); } }; other.d(); } }', + 'import type { FileHandle } from "node:fs/promises"; class C { h!: FileHandle; d() { return function (this: { h: { close(): void } }) { this.h.close(); }; } }', + ], + invalid: [ + { + code: 'import type { FileHandle } from "node:fs/promises"; class C { constructor(private h: FileHandle) {} d() { this.h.close(); } }', + errors: [{ messageId: "raw" }], + }, + { + code: 'import type { FileHandle } from "node:fs/promises"; class C { static h: FileHandle; h = { close() {} }; static d() { this.h.close(); } }', + errors: [{ messageId: "raw" }], + }, + { + code: 'import type { FileHandle } from "node:fs/promises"; class C { static h: FileHandle; static { this.h.close(); } }', + errors: [{ messageId: "raw" }], + }, + { + code: 'import type { FileHandle } from "node:fs/promises"; class C { h!: FileHandle; d() { return () => this.h.close(); } }', + errors: [{ messageId: "raw" }], + }, + { + code: 'import { open } from "node:fs/promises"; const h = await open("x"); await h.close();', + errors: [{ messageId: "raw" }], + }, + { + code: 'import * as fsp from "node:fs/promises"; let h; h = await fsp.open("x"); await h.close();', + errors: [{ messageId: "raw" }], + }, + { + code: 'import { open } from "fs/promises"; await (await open("x")).close();', + errors: [{ messageId: "raw" }], + }, + { + code: 'import { open } from "node:fs/promises"; const h = await open("x").catch(() => { throw new Error(); }); await h.close();', + errors: [{ messageId: "raw" }], + }, + { + code: 'import type { FileHandle } from "node:fs/promises"; const f = async (h: FileHandle | undefined) => h?.close();', + errors: [{ messageId: "raw" }], + }, + { + code: 'import type * as fsp from "node:fs/promises"; const f = (h: fsp.FileHandle) => h.close();', + errors: [{ messageId: "raw" }], + }, + { + code: 'import type { FileHandle } from "node:fs/promises"; class C { #h: FileHandle; async d() { await this.#h.close(); } }', + errors: [{ messageId: "raw" }], + }, + { + code: 'import { closeSync } from "node:fs"; closeSync(3);', + errors: [{ messageId: "raw" }], + }, + { + code: 'import * as fs from "fs"; fs.close(3, () => {});', + errors: [{ messageId: "raw" }], + }, + ], + }, +);