mirror of
https://github.com/morluto/rea.git
synced 2026-10-26 17:35:11 +00:00
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
This commit is contained in:
1 parent
c2068b94d7
commit
4a041ae753
9 files changed
+2413
-16
No files matched your search
@@ -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))
|
||||
@@ -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('<IIQQQQIIQQ', data, shoff + i * shsize) for i in range(shnum)]
|
||||
names = sections[names_index]
|
||||
assert names[1] == 3 and not names[2] & 0x800
|
||||
strings = data[names[4]:names[4] + names[5]]
|
||||
tables = []
|
||||
for i in range(phnum):
|
||||
program = struct.unpack_from('<IIQQQQQQ', data, phoff + i * phsize)
|
||||
if program[0] == 4:
|
||||
assert program[7] in [4, 8]
|
||||
tables.append({'kind':'program','index':i,'name':None,'location':{'offset':program[2],'bytes':program[5]},'alignment':program[7]})
|
||||
for i, section in enumerate(sections):
|
||||
if section[1] == 7:
|
||||
assert section[8] in [4, 8] and not section[2] & 0x800
|
||||
name_end = strings.index(0, section[0])
|
||||
tables.append({'kind':'section','index':i,'name':strings[section[0]:name_end].decode('ascii'),'location':{'offset':section[4],'bytes':section[5]},'alignment':section[8]})
|
||||
entries = []
|
||||
type_names = {1:'gnu_abi_tag',2:'gnu_hwcap',3:'gnu_build_id',4:'gnu_gold_version',5:'gnu_property'}
|
||||
for table in tables:
|
||||
pos = table['location']['offset']
|
||||
end = pos + table['location']['bytes']
|
||||
alignment = table['alignment']
|
||||
while pos < end:
|
||||
namesz, descsz, tag = struct.unpack_from('<III', data, pos)
|
||||
descriptor_offset = ((12 + namesz + alignment - 1) // alignment) * alignment
|
||||
next_offset = ((descriptor_offset + descsz + alignment - 1) // alignment) * alignment
|
||||
assert pos + next_offset <= end and namesz == 4 and descsz <= 64
|
||||
owner = data[pos + 12:pos + 12 + namesz]
|
||||
assert owner == b'GNU\0'
|
||||
descriptor = data[pos + descriptor_offset:pos + descriptor_offset + descsz]
|
||||
gnu = {'build_id_hex':None,'abi_tag':None,'gold_version':None}
|
||||
if tag == 3:
|
||||
gnu['build_id_hex'] = descriptor.hex()
|
||||
elif tag == 4:
|
||||
gnu['gold_version'] = descriptor.split(b'\0', 1)[0].decode('ascii')
|
||||
elif tag == 1:
|
||||
os_id, major, minor, subminor = struct.unpack('<IIII', descriptor)
|
||||
assert os_id == 0
|
||||
gnu['abi_tag'] = {'os':os_id,'os_name':'linux','major':major,'minor':minor,'subminor':subminor}
|
||||
entries.append({
|
||||
'index': len(entries),
|
||||
'table_kind': table['kind'],
|
||||
'table_index': table['index'],
|
||||
'location': {'offset': pos, 'bytes': next_offset},
|
||||
'namesz': namesz,
|
||||
'descsz': descsz,
|
||||
'type': tag,
|
||||
'type_name': type_names.get(tag, 'unknown'),
|
||||
'owner': 'GNU',
|
||||
'owner_bytes_hex': owner.hex(),
|
||||
'descriptor': {
|
||||
'location': {'offset': pos + descriptor_offset, 'bytes': descsz},
|
||||
'bytes': descsz,
|
||||
'sha256': hashlib.sha256(descriptor).hexdigest(),
|
||||
'hex': descriptor.hex(),
|
||||
},
|
||||
'gnu': gnu,
|
||||
})
|
||||
pos += next_offset
|
||||
assert pos == end
|
||||
after = path.stat()
|
||||
assert (before.st_dev,before.st_ino,before.st_size,before.st_mtime_ns,before.st_ctime_ns) == (after.st_dev,after.st_ino,after.st_size,after.st_mtime_ns,after.st_ctime_ns)
|
||||
return {
|
||||
'artifact': {'path': str(path), 'sha256': hashlib.sha256(data).hexdigest(), 'bytes': len(data)},
|
||||
'format': 'elf64',
|
||||
'byte_order': 'little',
|
||||
'machine': machine,
|
||||
'type': kind,
|
||||
'type_name': 'shared',
|
||||
'tables': tables,
|
||||
'entries': entries,
|
||||
'coverage': {'status': 'complete', 'examined_tables': len(tables), 'examined_entries': len(entries)},
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--path', type=Path, required=True)
|
||||
parser.add_argument('--output', type=Path, required=True)
|
||||
args = parser.parse_args()
|
||||
assert args.path.is_absolute()
|
||||
expected = inspect(args.path)
|
||||
result = {
|
||||
'operation': 'inspect_elf_notes',
|
||||
'input': {'path': str(args.path), 'max_file_bytes': 268435456},
|
||||
'expected': expected,
|
||||
'question': 'Return every declared ELF note table and note, preserving program/section identity, file ranges, raw owner bytes, descriptor digests, GNU fields and complete source Evidence.',
|
||||
}
|
||||
args.output.write_text(json.dumps(result,indent=2)+'\n')
|
||||
print(json.dumps({'artifact':expected['artifact'],'tables':len(expected['tables']),'entries':len(expected['entries'])}))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,81 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { createHash } from "node:crypto";
|
||||
import { writeFile } from "node:fs/promises";
|
||||
import { pathToFileURL } from "node:url";
|
||||
|
||||
const [runtime, mode, path, expectedDigest, output] = process.argv.slice(2);
|
||||
const { readStableArtifact } = await import(
|
||||
pathToFileURL(`${runtime}/dist/artifacts/readStableArtifact.js`)
|
||||
);
|
||||
const report = {
|
||||
mode,
|
||||
path,
|
||||
status: "running",
|
||||
node: process.version,
|
||||
memory_before: process.memoryUsage(),
|
||||
};
|
||||
let service;
|
||||
const start = performance.now();
|
||||
try {
|
||||
if (mode === "read") {
|
||||
const snapshot = await readStableArtifact(path, 256 * 1024 * 1024);
|
||||
report.read_ms = performance.now() - start;
|
||||
assert.equal(snapshot.sha256, expectedDigest);
|
||||
assert.equal(
|
||||
createHash("sha256").update(snapshot.bytes).digest("hex"),
|
||||
expectedDigest,
|
||||
);
|
||||
report.bytes = snapshot.bytes.length;
|
||||
report.sha256 = snapshot.sha256;
|
||||
} else {
|
||||
assert.equal(mode, "elf");
|
||||
const { ElfNotesService } = await import(
|
||||
pathToFileURL(
|
||||
`${runtime}/dist/application/binaryDiagnostics/ElfNotesService.js`,
|
||||
)
|
||||
);
|
||||
const { ElfNotesProvider } = await import(
|
||||
pathToFileURL(`${runtime}/dist/native/elf/ElfNotesProvider.js`)
|
||||
);
|
||||
service = new ElfNotesService(new ElfNotesProvider());
|
||||
const inspected = await service.inspect({
|
||||
path,
|
||||
max_file_bytes: 256 * 1024 * 1024,
|
||||
});
|
||||
assert.equal(inspected.ok, true, JSON.stringify(inspected));
|
||||
report.evidence = inspected.value;
|
||||
assert.equal(
|
||||
report.evidence.normalized_result.artifact.sha256,
|
||||
expectedDigest,
|
||||
);
|
||||
}
|
||||
report.status = "passed";
|
||||
} catch (error) {
|
||||
report.status = "failed";
|
||||
report.error = String(error);
|
||||
process.exitCode = 1;
|
||||
} finally {
|
||||
try {
|
||||
await service?.close();
|
||||
} catch (error) {
|
||||
report.status = "failed";
|
||||
report.cleanup_error = String(error);
|
||||
process.exitCode = 1;
|
||||
}
|
||||
report.elapsed_ms = performance.now() - start;
|
||||
report.memory_after = process.memoryUsage();
|
||||
report.process_max_rss_kib = process.resourceUsage().maxRSS;
|
||||
await writeFile(output, JSON.stringify(report, null, 2) + "\n");
|
||||
console.log(
|
||||
JSON.stringify({
|
||||
mode,
|
||||
path,
|
||||
status: report.status,
|
||||
elapsed_ms: report.elapsed_ms,
|
||||
read_ms: report.read_ms,
|
||||
bytes: report.bytes,
|
||||
max_rss_kib: report.process_max_rss_kib,
|
||||
array_buffers: report.memory_after.arrayBuffers,
|
||||
}),
|
||||
);
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,212 @@
|
||||
# Stable artifacts: read directly into one private snapshot
|
||||
|
||||
Issue: [#6954](https://github.com/morluto/rea/issues/6954).
|
||||
|
||||
`readStableArtifact` retained every stream chunk and then concatenated a second
|
||||
full-file buffer. Its 37 production call sites share this filesystem boundary.
|
||||
The owner now allocates one snapshot, capped by the caller's byte budget before
|
||||
converting the descriptor's bigint size, and fills it with direct descriptor
|
||||
reads. One reusable 64 KiB scratch buffer probes EOF and reads raced growth.
|
||||
There is no per-chunk data allocation or second full-file concatenation.
|
||||
|
||||
The buffer stays private until the count, descriptor/path state and cancellation
|
||||
checks succeed. Growth still reads through to EOF, an I/O failure or byte-budget
|
||||
refusal before final integrity checks. Pending reads settle before close; the
|
||||
existing descriptor owner retains failed cleanup and completed observations.
|
||||
The streaming helper keeps its separate immutable-chunk contract. No cache,
|
||||
public schema, report selection, Evidence identity or provider authority changes.
|
||||
|
||||
## Fixed real comparison
|
||||
|
||||
Baseline: `d87e73f2db50be72d33a4b3c7199af35518b441c`.
|
||||
Measured implementation: `45307fb9534f464eee094ccddc4e295d1c0dd0fb`.
|
||||
Both source trees were clean when pinned. Node 24.22.1, npm 11.16.0 and
|
||||
production MCP SDK 2.3.1 were identical. Runtime receipts fingerprint 2,582
|
||||
compiled/catalog/entrypoint/dependency and driver files per lane; the driver
|
||||
checks them and the Node binary before and after each task.
|
||||
|
||||
Artifacts were inspected as data without executing the targets:
|
||||
|
||||
| Artifact | Bytes | SHA-256 |
|
||||
| ---------------------------------------------------------- | ----------: | ------------------------------------------------------------------ |
|
||||
| Linux LLVM 16, `/usr/lib/x86_64-linux-gnu/libLLVM-16.so.1` | 123,379,936 | `f62d254b7f2bf42df8c8b07d46ee3bb4c2cafeca436b2e6bc6ccbe4581f58f40` |
|
||||
| GNU ls, `/usr/bin/ls` | 151,344 | `cb30d69b24245bf2ecdc9e7f53bbad19159999970b6d82c0c00c7d32d9e37aa4` |
|
||||
| OTP 27 `erl_parse.beam` | 608,240 | `729cce2b1688a1613d1ee71f5d0d99a9430d163d08fbe1aa5148dbbe6494f458` |
|
||||
|
||||
The ELF question requests every note table/entry, source ranges and decoded
|
||||
GNU facts, with complete Evidence access. `stable-artifact-elf-oracle.py`
|
||||
independently maps ELF64 little-endian file bytes with Python `struct`, including
|
||||
both program/section aliases, names, alignment, descriptor bytes/digests and
|
||||
examined coverage. Its supported corpus is explicit, not a general ELF parser
|
||||
claim. GNU readelf 2.40 independently agrees with the GNU build IDs, ABI and gold
|
||||
version. LLVM has three tables/three notes, build ID
|
||||
`649be1a7cae92cf064a666f2a64ececdf264cb7b`, and `gold 1.16`.
|
||||
The BEAM control requests the complete inspection, checked against the retained
|
||||
independent OTP chunk/table oracle from the BEAM investigation. All oracle
|
||||
fields match; full normalized reports, catalog and parent Evidence identities
|
||||
are identical across lanes. Full CLI Evidence also matches the MCP export.
|
||||
|
||||
Each case has three alternating fresh production stdio sessions per lane,
|
||||
18 tasks total. Each discovers all 222 tools, calls the exact discovered
|
||||
`toolDefinition` with the default SDK validator, inspects bytes and exports the
|
||||
complete Evidence bundle. HOME/client/server are fresh; Node compile cache is
|
||||
disabled; OS caches are unflushed. This measures fresh sessions, not cold disks.
|
||||
|
||||
| Completed task, 3 samples/lane | Whole elapsed median, s | Family CPU median, s | Kernel family peak median, MiB | Whole-task estimated tokens |
|
||||
| ------------------------------ | ----------------------: | -------------------: | -----------------------------: | --------------------------: |
|
||||
| LLVM notes | 15.848 → 12.209 | 13.393 → 11.161 | 756.496 → 725.566 | 58,486 → 58,486 |
|
||||
| ls notes | 7.385 → 7.288 | 6.477 → 6.660 | 644.621 → 643.844 | 58,992 → 58,992 |
|
||||
| OTP 27 erl_parse complete | 8.420 → 7.401 | 7.063 → 6.653 | 649.535 → 654.520 | 73,754 → 73,754 |
|
||||
|
||||
LLVM kernel-peak ranges are 743.4–819.3 → 721.6–732.2 MiB, a median reduction
|
||||
of 30.9 MiB (4.1%). Elapsed ranges are 13.853–18.630 → 11.994–12.631 seconds;
|
||||
CPU ranges are 12.982–14.749 → 10.294–11.868 seconds. These are observed cohort
|
||||
results, not a general latency guarantee: discovery medians also changed
|
||||
2.578 → 1.987 seconds, host frequency/background activity was uncontrolled, and
|
||||
absolute times shifted between earlier experiments with identical baseline
|
||||
runtime fingerprints. The small and BEAM controls do not establish a general
|
||||
memory or speed benefit. BEAM peak memory is slightly higher.
|
||||
|
||||
The [machine-readable report](./stable-artifact-reads.json) includes full ranges,
|
||||
per-task phases, commands, digests and resource observations.
|
||||
`cl100k_base`/tiktoken 0.14.0 counts the question, instructions, all tool names,
|
||||
descriptions and input schemas once, arguments and every text reply including
|
||||
export. The input directory costs 56,685 tokens; raw catalog including output
|
||||
schemas separately costs 838,375. Actual model usage, host framing/replay,
|
||||
inference latency and answer quality are unmeasured. There is no token saving.
|
||||
|
||||
## Locating the memory reduction
|
||||
|
||||
A separate 18-probe alternating cohort uses actual `readStableArtifact` and
|
||||
`ElfNotesService` with three samples/lane/case. Returned bytes are hashed again
|
||||
against independent `sha256sum`; every ELF oracle field is also checked.
|
||||
The retained probe source is reproduced in `stable-artifact-profile.mjs`
|
||||
(formatting only differs from the captured source). It does not force GC.
|
||||
|
||||
| Probe | Process max RSS median, MiB | Whole elapsed median, s | Family CPU median, s | Kernel family peak median, MiB |
|
||||
| ---------------- | --------------------------: | ----------------------: | -------------------: | -----------------------------: |
|
||||
| LLVM reader | 295.4 → 174.4 | 1.247 → 0.843 | 0.974 → 0.577 | 262.6 → 141.5 |
|
||||
| LLVM ELF service | 313.1 → 192.9 | 1.750 → 1.145 | 1.419 → 0.802 | 274.1 → 153.4 |
|
||||
| ls reader | 54.3 → 54.1 | 0.438 → 0.437 | 0.204 → 0.206 | 21.6 → 22.7 |
|
||||
|
||||
Linux process max RSS is distinct from cgroup memory, which includes charged
|
||||
cache and process-family overhead. The approximately 121 MiB RSS reduction on
|
||||
the large file supports removing the extra full-file copy. It must not be
|
||||
substituted for the smaller whole-MCP saving. Imports, independent digest
|
||||
verification and cleanup are included in whole probe costs. These phase probes
|
||||
are supporting evidence, not public CLI/MCP qualification for every consumer.
|
||||
|
||||
An earlier stream-copy implementation (`cb7364732`) passed correctness but did
|
||||
not establish a whole-MCP memory benefit: LLVM median 807.4 → 841.0 MiB with
|
||||
overlapping ranges. Those separate exploratory receipts are retained and are
|
||||
not pooled with the final direct-read cohort or used for final benefit claims.
|
||||
|
||||
## Verification and resource limits
|
||||
|
||||
Executed checks:
|
||||
|
||||
- 151 focused owner/consumer tests passed in
|
||||
`tests/boundary/filesystem/stableArtifact.test.ts` and
|
||||
`tests/process-global/stableArtifactConsumers.test.ts`, including short reads,
|
||||
cross-buffer partial tails, malformed byte counts, pending-read cancellation,
|
||||
physical I/O error precedence, raced huge sizes, growth refusal and incomplete
|
||||
private buffers through failed close/cleanup retry.
|
||||
- All six serial static tasks (typecheck, lint/module boundaries, format, knip,
|
||||
metadata and compilation), plus architecture guards passed. Static family
|
||||
peak was 1.60 GiB.
|
||||
- Six real CLI calls and 18 real stdio MCP/source-export tasks passed every
|
||||
independent oracle fact and full report/Evidence comparisons.
|
||||
- All 18 supporting producer probes passed their outcome, payload-exit,
|
||||
cleanup, memory/PID-event and independent oracle checks. The committed cost
|
||||
verifier reproduced the complete MCP measurement JSON exactly.
|
||||
|
||||
Heavy work ran serially under a one-CPU quota, 3 GiB early stop, 4 GiB hard
|
||||
limit, no swap and 192-task ceiling. Measured tasks include client/server,
|
||||
fingerprint reads, serialization, export and cleanup; tokenization and oracle
|
||||
generation are outside measurements. All measured payloads exited 0 with
|
||||
verified owned cleanup and no guard stop/OOM/limit events. Final whole-scope
|
||||
counters after transient-scope removal are unavailable and are not verified.
|
||||
|
||||
Two cached baseline build preparations returned successful tasks but their
|
||||
payload-exit witnesses still showed an extra process. Later authoritative
|
||||
systemd observations showed both scopes inactive/dead with empty control
|
||||
groups; the cause remains unknown. They are retained preparation attempts, not
|
||||
benchmark samples or verified-cleanup successes. Runtime bytes were checked
|
||||
separately in every measured MCP task. An initial CLI harness attempt failed
|
||||
before task start because its fresh environment omitted the user systemd bus;
|
||||
the corrected transport environment was passed explicitly without persisting
|
||||
ambient environment values. Initial test/typecheck/formatter failures from the
|
||||
implementation iterations remain in the local receipt archive.
|
||||
|
||||
Coverage is real static inspection of these Linux artifacts. Windows/macOS,
|
||||
unusual filesystems and all 37 caller workflows have not been runtime-qualified
|
||||
by this experiment. Existing cross-provider source tests are complementary
|
||||
coverage, not real-provider verification. No OOM reproduction is claimed.
|
||||
|
||||
## CI follow-up and integration
|
||||
|
||||
The first CI run, `40071483901` at `41adc629f`, passed source tests
|
||||
(17,305 passed, 13 skipped), static checks, package/Windows builds and all nine
|
||||
selected real-workflow/SQLite compatibility jobs. Four built-test shards passed;
|
||||
shard 4 failed two Go/PE cleanup cases and three shards were cancelled, leaving
|
||||
that coverage unknown.
|
||||
|
||||
The failed fixture selected its first descriptor during close and omitted that
|
||||
first call from its counter. The old stream's extra close invocation masked
|
||||
this assumption; the original fixture passes on the fixed baseline and fails
|
||||
with direct reads. The fixture now captures ownership at open, counts every
|
||||
attempt and checks the retained descriptor itself: open after failed close,
|
||||
closed after exactly one retry. This strengthens lifecycle coverage rather than
|
||||
lowering the expected count. An intermediate assertion needed a typed getter
|
||||
for state populated by asynchronous callbacks; its failed typecheck receipt is
|
||||
retained. The measured production reader was unchanged.
|
||||
|
||||
After integrating main `98ec00f40`, qualification at `8e5bf787650e147aa1927c920501665df45dfe3a` passed
|
||||
all 153 tests in the original two files plus
|
||||
`tests/boundary/providers/artifactReadCleanupRetry.test.ts`, all six serial static
|
||||
tasks and architecture guards. Three fresh real MCP/source-export tasks and
|
||||
three CLI calls matched every oracle fact and the frozen complete parent
|
||||
Evidence. All seven guarded groups exited 0 with verified payload cleanup and
|
||||
no memory/PID limit events or OOM; static peak was 1.59 GiB. These integration
|
||||
checks use their own runtime identity and preserve the fixed performance pins
|
||||
above. Updated PR CI/review is separate from this local qualification.
|
||||
|
||||
## Reproduction
|
||||
|
||||
Build both pinned checkouts serially with their locked toolchain and matching
|
||||
dependencies, without upgrading unrelated host software. Use the artifact
|
||||
hashes above. The ELF oracle accepts `--path` and `--output`:
|
||||
|
||||
```sh
|
||||
python3 experiments/analysis-efficiency/stable-artifact-elf-oracle.py \
|
||||
--path /usr/lib/x86_64-linux-gnu/libLLVM-16.so.1 --output /absolute/oracle.json
|
||||
python3 experiments/analysis-efficiency/resource-run.py --observe-payload-exit -- \
|
||||
node experiments/analysis-efficiency/stable-artifact-reads.mjs \
|
||||
--runtime /absolute/pinned-checkout --oracle /absolute/task-oracle.json \
|
||||
--identity /absolute/runtime-identity.json --output /absolute/new-task-directory
|
||||
```
|
||||
|
||||
The task oracle has `{operation, input, expected, question}`; use
|
||||
`inspect_elf_notes` with `{path, max_file_bytes: 268435456}` and the independently
|
||||
generated expected object. The small control has the same options. The BEAM
|
||||
operation is `inspect_erlang_beam`, complete/default selection, path only.
|
||||
Identity JSON has `source_revision`, `source_clean`, `node_sha256` and
|
||||
`files: [{path, sha256}]`; fingerprint the full compiled runtime/catalogs,
|
||||
entrypoints, lock/toolchain, actual SDK JS and driver/helpers. The historical
|
||||
payload affinity was CPUs 0–1, initial observer CPU 15, with the limits above.
|
||||
|
||||
Raw receipt archive: `/tmp/rea-perf-stable-read-20261024/direct/` on the
|
||||
investigation host. It contains runtime/oracle receipts, `cohort/index.json`,
|
||||
per-task catalogs/transcripts/reports/source exports/guard logs, `cli/` outputs,
|
||||
`profiles/` raw source results and verified summary, and static/architecture
|
||||
logs. The parent directory retains earlier variants and failed preparations.
|
||||
Reproduce the published MCP aggregation with:
|
||||
|
||||
```sh
|
||||
python3 experiments/analysis-efficiency/stable-artifact-costs.py \
|
||||
--receipts /absolute/direct-receipt-directory --output /absolute/costs.json
|
||||
```
|
||||
|
||||
The verifier requires all 18 distinct successful samples, exact oracle/runtime
|
||||
receipts, full source exports and equal parent/report/catalog identities.
|
||||
A failed or guard-stopped attempt cannot enter the report.
|
||||
@@ -0,0 +1,156 @@
|
||||
/** Compare shared stable reads through real ELF/BEAM stdio tasks and full source export. */
|
||||
import assert from "node:assert/strict";
|
||||
import { createHash } from "node:crypto";
|
||||
import { createWriteStream } from "node:fs";
|
||||
import { mkdir, mkdtemp, readFile, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join, resolve } from "node:path";
|
||||
import { finished } from "node:stream/promises";
|
||||
import { parseArgs } from "node:util";
|
||||
import { Client } from "@modelcontextprotocol/client";
|
||||
import { StdioClientTransport } from "@modelcontextprotocol/client/stdio";
|
||||
import { createRetainedQueryCaller } from "./retained-query-calls.mjs";
|
||||
import { finalizeRetainedQueryReceipt } from "./retained-query-finalization.mjs";
|
||||
|
||||
const { values } = parseArgs({
|
||||
options: Object.fromEntries(
|
||||
["runtime", "oracle", "identity", "output"].map((name) => [
|
||||
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;
|
||||
}
|
||||
@@ -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"),
|
||||
};
|
||||
},
|
||||
);
|
||||
|
||||
@@ -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<void>((resolve) => {
|
||||
enter = resolve;
|
||||
});
|
||||
let release!: () => void;
|
||||
let reject!: (cause: unknown) => void;
|
||||
const blocked = new Promise<void>((resolve, fail) => {
|
||||
release = resolve;
|
||||
reject = fail;
|
||||
});
|
||||
let descriptor: Awaited<ReturnType<typeof open>> | 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<void>((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<void> => {
|
||||
@@ -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) => {
|
||||
|
||||
@@ -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<typeof import("node:fs/promises")>();
|
||||
return {
|
||||
@@ -18,6 +24,9 @@ vi.mock("node:fs/promises", async (importOriginal) => {
|
||||
open: async (...args: Parameters<typeof fs.open>) => {
|
||||
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;
|
||||
|
||||
Reference in new issue
Block a user