From 4a041ae7532588eaf9973584e517c9bce995445b Mon Sep 17 00:00:00 2001 From: gensokyo-dev <342580529+gensokyo-dev@users.noreply.github.com> Date: Sat, 24 Oct 2026 15:49:41 +0800 Subject: [PATCH] perf(artifacts): bound stable reads to one snapshot buffer (#6966) * perf(artifacts): read stable files into one bounded buffer * perf(artifacts): fill stable snapshots through direct descriptor reads * docs(perf): qualify stable snapshot memory with real ELF and BEAM tasks * test(artifacts): count descriptor close retries from admission * test(artifacts): capture the retained descriptor after asynchronous admission * docs(perf): record integrated qualification and cleanup fixture correction --- .../stable-artifact-costs.py | 99 ++ .../stable-artifact-elf-oracle.py | 108 ++ .../stable-artifact-profile.mjs | 81 + .../stable-artifact-reads.json | 1430 +++++++++++++++++ .../stable-artifact-reads.md | 212 +++ .../stable-artifact-reads.mjs | 156 ++ src/artifacts/readStableArtifact.ts | 46 +- .../filesystem/stableArtifact.test.ts | 282 ++++ .../artifactReadCleanupRetry.test.ts | 15 +- 9 files changed, 2413 insertions(+), 16 deletions(-) create mode 100644 experiments/analysis-efficiency/stable-artifact-costs.py create mode 100644 experiments/analysis-efficiency/stable-artifact-elf-oracle.py create mode 100644 experiments/analysis-efficiency/stable-artifact-profile.mjs create mode 100644 experiments/analysis-efficiency/stable-artifact-reads.json create mode 100644 experiments/analysis-efficiency/stable-artifact-reads.md create mode 100644 experiments/analysis-efficiency/stable-artifact-reads.mjs diff --git a/experiments/analysis-efficiency/stable-artifact-costs.py b/experiments/analysis-efficiency/stable-artifact-costs.py new file mode 100644 index 0000000000..2b527e5760 --- /dev/null +++ b/experiments/analysis-efficiency/stable-artifact-costs.py @@ -0,0 +1,99 @@ +"""Verify fixed real ELF/BEAM tasks and their whole-task resource/token costs.""" +import argparse +from pathlib import Path +import hashlib +import json +import statistics +import tiktoken + +parser = argparse.ArgumentParser(description=__doc__) +parser.add_argument('--receipts', type=Path, required=True) +parser.add_argument('--output', type=Path, required=True) +options = parser.parse_args() +receipts = options.receipts +cohort = receipts / 'cohort' +index = json.loads((cohort / 'index.json').read_text()) +assert len(index) == 18 and all((record['exit'] == 0 for record in index)) +assert len({record['name'] for record in index}) == 18 +encoding = tiktoken.get_encoding('cl100k_base') +compact = lambda d: json.dumps(d, ensure_ascii=False, separators=(',', ':'), allow_nan=False) +count = lambda s: len(encoding.encode(s, disallowed_special=())) +digest = lambda task: hashlib.sha256(task.read_bytes()).hexdigest() +rows = [] +for item in index: + task = cohort / item['name'] + record = json.loads((task / 'report.json').read_text()) + catalog = json.loads((task / 'catalog.json').read_text()) + transcript = json.loads((task / 'transcript.json').read_text()) + scope = next((json.loads(l) for l in reversed((cohort / (item['name'] + '.log')).read_text().splitlines()) if l.startswith('{"resource_scope":'))) + observed = scope['payload_resource_observation'] + events = observed['resource_events'] + assert record['oracle_sha256'] == digest(receipts / f"{item['case']}-oracle.json") + assert item['exit'] == 0 and record['status'] == 'passed' and record['complete_source_export_verified'] and (not record['cleanup_failures']) and (not record['receipt_failures']) + assert scope['abort'] is None and scope['payload_resource_observation_verified'] and (observed['payload_exit_code'] == 0) and observed['payload_cleanup_verified'] + assert all((events['memory.events'][k] == 0 for k in ['high', 'max', 'oom', 'oom_kill'])) and events['pids.events']['max'] == 0 + oracle = json.loads((receipts / f"{item['case']}-oracle.json").read_text()) + assert [e['name'] for e in transcript] == [oracle['operation'], 'export_evidence_bundle'] + assert transcript[0]['input'] == oracle['input'] and record['question'] == oracle['question'] + assert all((not e['reply'].get('isError', False) for e in transcript)) + directory_tokens = count(compact([{k: t[k] for k in ['name', 'description', 'inputSchema']} for t in catalog['tools']])) + argument_tokens = sum((count(compact({'name': e['name'], 'arguments': e['input']})) for e in transcript)) + reply_tokens = sum((count(c['text']) for e in transcript for c in e['reply']['content'] if c['type'] == 'text')) + question_tokens = count(record['question']) + instruction_tokens = count(record['server_instructions']) + rows.append({ + **item, + 'catalog_tools': len(catalog['tools']), + 'catalog_sha256': digest(task / 'catalog.json'), + 'parent_evidence_id': record['parent_evidence_id'], + 'source_report_sha256': record['source_report_sha256'], + 'runtime_identity_sha256': record['runtime_identity_sha256'], + 'wall_seconds': scope['elapsed_seconds'], + 'cpu_seconds': observed['cpu_stat']['usage_usec'] / 1000000.0, + 'kernel_family_peak_bytes': observed['memory_peak_bytes'], + 'inspection_ms': record['calls'][0]['elapsed_ms'], + 'discovery_ms': record['phases']['discovery_ms'], + 'token_estimate': { + 'question': question_tokens, + 'instructions': instruction_tokens, + 'directory_once': directory_tokens, + 'arguments': argument_tokens, + 'text_replies': reply_tokens, + 'whole_task': question_tokens + instruction_tokens + directory_tokens + argument_tokens + reply_tokens, + 'raw_catalog': count(compact(catalog)), + }, + 'receipt_sha256': {n: digest(task / n) for n in ['report.json', 'transcript.json', 'catalog.json', 'source.evidence.json']}, + 'payload_exit_events': events, + 'payload_cleanup_verified': True, + 'whole_scope_events_final': scope['resource_events_final'], + }) +summary = {} +for case in ['llvm', 'ls', 'beam']: + case_samples = [record for record in rows if record['case'] == case] + for k in ['catalog_sha256', 'source_report_sha256', 'parent_evidence_id']: + assert len({record[k] for record in case_samples}) == 1, (case, k) + summary[case] = {} + for lane in ['before', 'after']: + lane_samples = [record for record in case_samples if record['lane'] == lane] + assert len(lane_samples) == 3 + assert {record['repeat'] for record in lane_samples} == {1, 2, 3} + metrics = {k: [record[k] for record in lane_samples] for k in ['wall_seconds', 'cpu_seconds', 'kernel_family_peak_bytes', 'inspection_ms', 'discovery_ms']} + metrics['whole_task_tokens'] = [record['token_estimate']['whole_task'] for record in lane_samples] + summary[case][lane] = {k: {'median': statistics.median(v), 'min': min(v), 'max': max(v)} for k, v in metrics.items()} +identities = {lane: json.loads((receipts / f'{lane}-runtime-identity.json').read_text()) for lane in ['before', 'after']} +assert all((identity['source_clean'] for identity in identities.values())) +for row in rows: + assert row['runtime_identity_sha256'] == digest(receipts / f"{row['lane']}-runtime-identity.json") +result = { + 'tiktoken': tiktoken.__version__, + 'encoding': encoding.name, + 'actual_model_usage': None, + 'revisions': {l: i['source_revision'] for l, i in identities.items()}, + 'identity_receipts': {l: digest(receipts / f'{l}-runtime-identity.json') for l in identities}, + 'oracles': {c: {'sha256': digest(receipts / f'{c}-oracle.json'), 'artifact': json.loads((receipts / f'{c}-oracle.json').read_text())['expected']['artifact']} for c in summary}, + 'summary': summary, + 'recordings': rows, + 'measurement': 'Three alternating fresh production stdio SDK sessions per lane/case; exact discovered toolDefinition/default validator, fresh HOME, Node compile cache disabled, OS caches unflushed. Each task discovers complete catalog, inspects bytes and exports/compares full source Evidence; all oracle facts checked. Whole process-family CPU and kernel peak include client/server, runtime fingerprint verification, serialization, export and cleanup. Oracle generation and tokenization outside measured tasks. One-CPU quota, 3GiB early stop/4GiB hard/no swap; payload exit/cleanup/events verified, final whole-scope counters unavailable. Token serialization estimate includes question/instructions/all input name+description+schema once/all arguments/all text replies, raw catalog separately. Host framing, replay, model inference quality and billing unmeasured.', +} +options.output.write_text(json.dumps(result, indent=2) + '\n') +print(json.dumps(summary, indent=2)) diff --git a/experiments/analysis-efficiency/stable-artifact-elf-oracle.py b/experiments/analysis-efficiency/stable-artifact-elf-oracle.py new file mode 100644 index 0000000000..f6bdf32c6e --- /dev/null +++ b/experiments/analysis-efficiency/stable-artifact-elf-oracle.py @@ -0,0 +1,108 @@ +"""Independent ELF64 little-endian note oracle for the fixed Linux corpus.""" +import argparse +import hashlib +import json +from pathlib import Path +import struct + + +def inspect(path): + before = path.stat() + data = path.read_bytes() + header = struct.unpack_from('<16sHHIQQQIHHHHHH', data) + ident, kind, machine, version, _, phoff, shoff, _, _, phsize, phnum, shsize, shnum, names_index = header + assert ident[:7] == b'\x7fELF\x02\x01\x01' and version == 1 and kind == 3 + assert phsize == 56 and shsize == 64 and 0 < phnum < 65535 and 0 < shnum < 65535 + sections = [struct.unpack_from(' [ + name, + { type: "string" }, + ]), + ), +}); +for (const value of Object.values(values)) assert.equal(value, resolve(value)); +const sha256 = (value) => createHash("sha256").update(value).digest("hex"); +const oracleBytes = await readFile(values.oracle); +const oracle = JSON.parse(oracleBytes); +assert.ok( + ["inspect_elf_notes", "inspect_erlang_beam"].includes(oracle.operation), +); +const identityBytes = await readFile(values.identity); +const identity = JSON.parse(identityBytes); +await mkdir(values.output, { mode: 0o700 }); +const account = await mkdtemp(join(tmpdir(), "rea-stable-task-")); +const client = new Client({ name: "stable-file-complete-task", version: "1" }); +const transport = new StdioClientTransport({ + command: process.execPath, + args: [join(values.runtime, "scripts/rea.mjs"), "mcp"], + maxBufferSize: 10485760, + env: { + HOME: account, + TMPDIR: account, + PATH: process.env.PATH, + REA_LOG_LEVEL: "silent", + NODE_OPTIONS: "--max-old-space-size=2048", + NODE_DISABLE_COMPILE_CACHE: "1", + UV_THREADPOOL_SIZE: "2", + }, + stderr: "pipe", +}); +const serverLog = createWriteStream(join(values.output, "server.log")); +const serverLogFinished = finished(serverLog); +serverLogFinished.catch(() => {}); +transport.stderr?.pipe(serverLog); +const report = { + status: "running", + question: oracle.question, + request: oracle.input, + calls: [], + phases: {}, + client: + "Production stdio SDK, exact discovered toolDefinition and default validator", + runtime_identity_sha256: sha256(identityBytes), + oracle_sha256: sha256(oracleBytes), + driver_sha256: sha256(await readFile(new URL(import.meta.url))), +}; +const transcript = []; +let definitions; +const call = createRetainedQueryCaller( + { + callTool(params, options) { + return client.callTool(params, { + ...options, + toolDefinition: definitions.get(params.name), + }); + }, + }, + (receipt, event) => { + report.calls.push(receipt); + transcript.push(event); + }, +); +const observe = async (name, args, phase) => { + const result = await call(name, args, phase); + assert.notEqual(result.isError, true, JSON.stringify(result)); + assert.ok(result.structuredContent); + return result.structuredContent; +}; +const verifyRuntime = async () => { + assert.equal(sha256(await readFile(values.oracle)), sha256(oracleBytes)); + assert.equal(sha256(await readFile(values.identity)), sha256(identityBytes)); + assert.equal(sha256(await readFile(process.execPath)), identity.node_sha256); + assert.equal( + sha256(await readFile(new URL(import.meta.url))), + report.driver_sha256, + ); + for (const item of identity.files) + assert.equal(sha256(await readFile(item.path)), item.sha256, item.path); +}; +try { + await verifyRuntime(); + const start = performance.now(); + await client.connect(transport); + report.server_instructions = client.getInstructions() ?? ""; + const catalog = await client.listTools(); + definitions = new Map(catalog.tools.map((tool) => [tool.name, tool])); + await writeFile(join(values.output, "catalog.json"), JSON.stringify(catalog)); + report.phases.discovery_ms = performance.now() - start; + const parent = await observe(oracle.operation, oracle.input, "inspect"); + for (const [key, value] of Object.entries(oracle.expected)) + assert.deepEqual(parent.normalized_result[key], value, key); + report.parent_evidence_id = parent.evidence_id; + report.source_report_sha256 = sha256( + JSON.stringify(parent.normalized_result), + ); + report.artifact = parent.normalized_result.artifact; + await observe( + "export_evidence_bundle", + { path: join(values.output, "source.evidence.json") }, + "export", + ); + const bundle = JSON.parse( + await readFile(join(values.output, "source.evidence.json")), + ); + assert.deepEqual( + bundle.records.find((item) => item.evidence_id === parent.evidence_id), + parent, + ); + report.complete_source_export_verified = true; + report.status = "passed"; +} catch (cause) { + report.status = "failed"; + report.error = String(cause); +} finally { + try { + await verifyRuntime(); + } catch (cause) { + report.status = "failed"; + report.identity_failure = String(cause); + } + await finalizeRetainedQueryReceipt({ + client, + transport, + account, + serverLog, + serverLogFinished, + report, + transcript, + output: values.output, + }); + if ( + report.status !== "passed" || + report.cleanup_failures.length || + report.receipt_failures.length + ) + process.exitCode = 1; +} diff --git a/src/artifacts/readStableArtifact.ts b/src/artifacts/readStableArtifact.ts index 183e5162b8..c67c41eaa1 100644 --- a/src/artifacts/readStableArtifact.ts +++ b/src/artifacts/readStableArtifact.ts @@ -81,7 +81,8 @@ export const readStableArtifact = async ( "path", `Expected a regular file without a symlink: ${path}`, ); - if (before.size > BigInt(maximumBytes)) + const maximumSize = BigInt(maximumBytes); + if (before.size > maximumSize) throw new ArtifactReaderFailure( "limit", `Selected artifact exceeds ${String(maximumBytes)} bytes: ${path}`, @@ -109,28 +110,46 @@ export const readStableArtifact = async ( "integrity", `Artifact changed before open: ${path}`, ); - const chunks: Buffer[] = []; + assertArtifactReadActive(signal); + // A raced descriptor size must not allocate beyond the selected budget. + // These private bytes are published only after the exact count and both + // file identities below validate, so an incomplete read exposes no tail. + const bytes = Buffer.allocUnsafe( + Number(initial.size > maximumSize ? maximumSize : initial.size), + ); + const overflow = Buffer.allocUnsafe(65536); + const hash = createHash("sha256"); let count = 0; - for await (const chunk of file.createReadStream({ - autoClose: false, - highWaterMark: 65536, - })) { + for (;;) { assertArtifactReadActive(signal); - const value: unknown = chunk; - if (!Buffer.isBuffer(value)) + const inside = count < bytes.length; + const target = inside ? bytes : overflow; + const offset = inside ? count : 0; + const wanted = Math.min(65536, target.length - offset); + const { bytesRead } = await file.read(target, offset, wanted, count); + // EOF still passes through the original final metadata/cancellation + // checks. A rejected physical read remains primary over cancellation. + if (bytesRead === 0) break; + assertArtifactReadActive(signal); + if ( + !Number.isSafeInteger(bytesRead) || + bytesRead < 0 || + bytesRead > wanted + ) throw new ArtifactReaderFailure( "format", - `Artifact stream did not return bytes: ${path}`, + `Artifact descriptor returned an invalid byte count: ${path}`, ); - const bytes = value; - count += bytes.length; + count += bytesRead; if (count > maximumBytes) throw new ArtifactReaderFailure( "limit", `Selected artifact grew beyond its byte limit: ${path}`, { boundary: "file-size" }, ); - chunks.push(bytes); + // Read growth into bounded scratch space while preserving physical + // read failures and byte-budget refusal ahead of final integrity checks. + hash.update(target.subarray(offset, offset + bytesRead)); } const after = await file.stat({ bigint: true }); const current = await afterValidation(path, "during read", () => @@ -145,10 +164,9 @@ export const readStableArtifact = async ( `Artifact changed during read: ${path}`, ); assertArtifactReadActive(signal); - const bytes = Buffer.concat(chunks, count); return { bytes, - sha256: createHash("sha256").update(bytes).digest("hex"), + sha256: hash.digest("hex"), }; }, ); diff --git a/tests/boundary/filesystem/stableArtifact.test.ts b/tests/boundary/filesystem/stableArtifact.test.ts index 0ce9fcd64f..d3616d111d 100644 --- a/tests/boundary/filesystem/stableArtifact.test.ts +++ b/tests/boundary/filesystem/stableArtifact.test.ts @@ -1,5 +1,6 @@ import { lstat, + appendFile, open, rename, symlink, @@ -47,6 +48,247 @@ it("reads exact binary and empty bytes and rejects symlink/size/cancellation bou }); }); +it("preserves exact binary bytes and digest across physical chunks and a partial tail", async () => { + const root = await createTestTempDirectory("rea-stable-chunks-"); + const path = join(root, "source.bin"); + const bytes = Buffer.from( + Array.from({ length: 3 * 65536 + 7 }, (_, i) => i % 251), + ); + await writeFile(path, bytes); + const snapshot = await readStableArtifact(path, bytes.length); + expect(snapshot.sha256).toBe( + createHash("sha256").update(bytes).digest("hex"), + ); + expect(snapshot.bytes.equals(bytes)).toBe(true); +}); + +it("joins short descriptor reads into the exact snapshot without gaps", async () => { + const root = await createTestTempDirectory("rea-stable-short-read-"); + const path = join(root, "source.bin"); + const bytes = Buffer.from(Array.from({ length: 1031 }, (_, i) => i % 251)); + await writeFile(path, bytes); + const snapshot = await readStableArtifact(path, bytes.length, undefined, { + lstat: (selected) => lstat(selected, { bigint: true }), + open: async (...args) => { + const handle = await open(...args); + const read = handle.read.bind(handle); + Object.defineProperty(handle, "read", { + value: async ( + buffer: Buffer, + offset: number, + length: number, + position: number, + ) => read(buffer, offset, Math.min(length, 17), position), + }); + return handle; + }, + }); + expect(snapshot.bytes.equals(bytes)).toBe(true); + expect(snapshot.sha256).toBe( + createHash("sha256").update(bytes).digest("hex"), + ); +}); + +it.each([-1, Number.NaN, "oversized"] as const)( + "refuses malformed descriptor byte counts: %s", + async (count) => { + const root = await createTestTempDirectory("rea-stable-read-count-"); + const path = join(root, "source.bin"); + await writeFile(path, "base"); + await expect( + readStableArtifact(path, 64, undefined, { + lstat: (selected) => lstat(selected, { bigint: true }), + open: async (...args) => { + const handle = await open(...args); + Object.defineProperty(handle, "read", { + value: async (buffer: Buffer, _offset: number, length: number) => ({ + bytesRead: count === "oversized" ? length + 1 : count, + buffer, + }), + }); + return handle; + }, + }), + ).rejects.toMatchObject({ reason: "format" }); + }, +); + +it.each(["complete", "fail"] as const)( + "joins an admitted descriptor read before cancellation finalization: %s", + async (outcome) => { + const root = await createTestTempDirectory("rea-stable-read-cancel-"); + const path = join(root, "source.bin"); + const bytes = Buffer.from("base"); + await writeFile(path, bytes); + const controller = new AbortController(); + const reason = { request: "cancelled while descriptor read was pending" }; + const physical = Object.assign(new Error("pending physical read failed"), { + code: "EIO", + }); + let enter!: () => void; + const entered = new Promise((resolve) => { + enter = resolve; + }); + let release!: () => void; + let reject!: (cause: unknown) => void; + const blocked = new Promise((resolve, fail) => { + release = resolve; + reject = fail; + }); + let descriptor: Awaited> | undefined; + const reading = readStableArtifact(path, 64, controller.signal, { + lstat: (selected) => lstat(selected, { bigint: true }), + open: async (...args) => { + const handle = await open(...args); + descriptor = handle; + Object.defineProperty(handle, "read", { + value: async (buffer: Buffer, offset: number) => { + enter(); + await blocked; + buffer.set(bytes, offset); + return { bytesRead: bytes.length, buffer }; + }, + }); + return handle; + }, + }); + let settled = false; + const result = reading + .catch((cause: unknown) => cause) + .finally(() => { + settled = true; + }); + try { + await entered; + controller.abort(reason); + await new Promise((resolve) => setImmediate(resolve)); + expect(settled).toBe(false); + expect(descriptor?.fd).toBeGreaterThanOrEqual(0); + if (outcome === "fail") reject(physical); + else release(); + const failure = await result; + if (outcome === "fail") expect(failure).toBe(physical); + else + expect(failure).toMatchObject({ reason: "cancelled", cause: reason }); + expect(artifactFileCompletion(failure)).toBeUndefined(); + expect(descriptor?.fd).toBe(-1); + } finally { + release(); + await result; + } + }, +); + +it.each(["integrity", "limit", "io"] as const)( + "preserves %s precedence after growth beyond the initial snapshot size", + async (outcome) => { + const root = await createTestTempDirectory("rea-stable-growth-"); + const path = join(root, "source.bin"); + await writeFile(path, "base"); + const extra = Buffer.from("expanded"); + const physical = Object.assign(new Error("late physical read failure"), { + code: "EIO", + }); + const reading = readStableArtifact( + path, + outcome === "limit" ? 12 : 1024, + undefined, + { + lstat: (selected) => lstat(selected, { bigint: true }), + open: async (...args) => { + const handle = await open(...args); + const read = handle.read.bind(handle); + let appended = 0; + Object.defineProperty(handle, "read", { + value: async ( + buffer: Buffer, + offset: number, + length: number, + position: number, + ) => { + if (appended < 2 && position >= 4 + appended * extra.length) { + await appendFile(path, extra); + appended += 1; + } + if (outcome === "io" && position >= 4 + 2 * extra.length) + throw physical; + return read(buffer, offset, length, position); + }, + }); + return handle; + }, + }, + ); + if (outcome === "io") await expect(reading).rejects.toBe(physical); + else + await expect(reading).rejects.toMatchObject({ + reason: outcome, + ...(outcome === "limit" ? { boundary: "file-size" } : {}), + }); + }, +); + +it("bounds allocation when the opened descriptor exceeds the selected byte budget", async () => { + const root = await createTestTempDirectory("rea-stable-admission-race-"); + const path = join(root, "source.bin"); + await writeFile(path, "base"); + await expect( + readStableArtifact(path, 64, undefined, { + lstat: (selected) => lstat(selected, { bigint: true }), + open: async (...args) => { + const handle = await open(...args); + const stat = handle.stat.bind(handle); + Object.defineProperty(handle, "stat", { + value: async () => + Object.assign(await stat({ bigint: true }), { size: 1n << 62n }), + }); + return handle; + }, + }), + ).rejects.toMatchObject({ reason: "integrity" }); +}); + +it("does not start a read when cancellation arrives during descriptor admission", async () => { + const root = await createTestTempDirectory("rea-stable-admission-cancel-"); + const path = join(root, "source.bin"); + await writeFile(path, "base"); + const controller = new AbortController(); + const reason = { request: "cancelled while admitting descriptor" }; + let reads = 0; + const reading = readStableArtifact(path, 64, controller.signal, { + lstat: (selected) => lstat(selected, { bigint: true }), + open: async (...args) => { + const handle = await open(...args); + const stat = handle.stat.bind(handle); + const read = handle.read.bind(handle); + Object.defineProperty(handle, "stat", { + value: async () => { + const facts = await stat({ bigint: true }); + controller.abort(reason); + return facts; + }, + }); + Object.defineProperty(handle, "read", { + value: async ( + buffer: Buffer, + offset: number, + length: number, + position: number, + ) => { + reads += 1; + return read(buffer, offset, length, position); + }, + }); + return handle; + }, + }); + await expect(reading).rejects.toMatchObject({ + reason: "cancelled", + cause: reason, + }); + expect(reads).toBe(0); +}); + const checkChange = async ( stage: "before-open" | "after-read" | "symlink-replacement", ): Promise => { @@ -119,6 +361,46 @@ const heldDescriptor = async () => { }; }; +it.each(["scoped", "transferred"])( + "keeps incomplete private bytes unavailable through %s close failure", + async (ownership) => { + const file = await heldDescriptor(); + const read = file.handle.read.bind(file.handle); + Object.defineProperty(file.handle, "read", { + value: async ( + buffer: Buffer, + offset: number, + length: number, + position: number, + ) => + position >= 2 + ? { bytesRead: 0, buffer } + : read(buffer, offset, Math.min(length, 2 - position), position), + }); + const scope = + ownership === "scoped" ? new ArtifactResourceScope() : undefined; + const failure = await readStableArtifact( + file.path, + 1024, + undefined, + file.fileSystem, + scope, + ).catch((cause: unknown) => cause); + expect(failure).toBeInstanceOf(ArtifactFileCleanupFailure); + if (!(failure instanceof ArtifactFileCleanupFailure)) + throw new Error("Missing incomplete read owner"); + expect(failure.fileCleanupFailure.outcome).toMatchObject({ + kind: "failed", + cause: { reason: "integrity" }, + }); + expect(artifactFileCompletion(failure)).toBeUndefined(); + file.release(); + if (scope !== undefined) await scope.close(); + else await failure.cleanupLease.release(); + expect(file.handle.fd).toBe(-1); + }, +); + it.each(["scoped", "transferred"])( "retains exact completed bytes and descriptor cleanup under %s ownership", async (ownership) => { diff --git a/tests/boundary/providers/artifactReadCleanupRetry.test.ts b/tests/boundary/providers/artifactReadCleanupRetry.test.ts index ebaba673ab..1903ca368a 100644 --- a/tests/boundary/providers/artifactReadCleanupRetry.test.ts +++ b/tests/boundary/providers/artifactReadCleanupRetry.test.ts @@ -11,6 +11,12 @@ const cleanupState = vi.hoisted(() => ({ firstHandleCloseAttempts: 0, })); +const openedFirstHandle = (): FileHandle => { + const handle = cleanupState.firstHandle; + if (handle === undefined) throw new Error("Missing admitted descriptor"); + return handle; +}; + vi.mock("node:fs/promises", async (importOriginal) => { const fs = await importOriginal(); return { @@ -18,6 +24,9 @@ vi.mock("node:fs/promises", async (importOriginal) => { open: async (...args: Parameters) => { const handle = await fs.open(...args); const close = handle.close.bind(handle); + // Establish the owner at admission and count its first close as well as + // retries; stream teardown must not supply an implicit extra attempt. + cleanupState.firstHandle ??= handle; if (cleanupState.abortController !== undefined) setImmediate(() => { cleanupState.abortController?.abort(); @@ -26,8 +35,6 @@ vi.mock("node:fs/promises", async (importOriginal) => { handle.close = async () => { if (handle === cleanupState.firstHandle) cleanupState.firstHandleCloseAttempts += 1; - else if (cleanupState.firstHandle === undefined) - cleanupState.firstHandle = handle; if ( cleanupState.failFirstClose && handle === cleanupState.firstHandle @@ -84,10 +91,14 @@ it.each(["go", "pe"] as const)( expect(first.error.cause.errors[1]).toMatchObject({ cleanupResources: [path], }); + expect(cleanupState.firstHandleCloseAttempts).toBe(1); + const retainedHandle = openedFirstHandle(); + expect(retainedHandle.fd).toBeGreaterThanOrEqual(0); const retry = await provider.inspect(input); expect(retry.ok).toBe(true); expect(cleanupState.firstHandleCloseAttempts).toBe(2); + expect(retainedHandle.fd).toBe(-1); } finally { await provider.close(); cleanupState.failFirstClose = false;