From ab396d5c701fd7a0401a714994f3af0500dc3272 Mon Sep 17 00:00:00 2001 From: HypForge Date: Sun, 18 Oct 2026 02:21:22 +0000 Subject: [PATCH] Fix #2053: hold loadSpooledBodies' shared read through the owner's unlink The shared `reading` entry was dropped as soon as the read settled, before the owning caller unlinked an unparseable body. A caller that arrived during the unlink read the file again, got ENOENT, and counted the body as missing instead of unparseable. The entry now counts its users and the last one out drops it in a finally, on every exit path, so it covers the unlink and never outlives the reads in flight. Co-Authored-By: Claude Opus 5.5 --- .../claude/src/telemetry/bodies.js | 131 +++++++++--------- .../claude-telemetry-unparseable-body.test.js | 97 ++++++++++++- 2 files changed, 162 insertions(+), 66 deletions(-) diff --git a/hypaware-core/plugins-workspace/claude/src/telemetry/bodies.js b/hypaware-core/plugins-workspace/claude/src/telemetry/bodies.js index cad49f5ef..464385d19 100644 --- a/hypaware-core/plugins-workspace/claude/src/telemetry/bodies.js +++ b/hypaware-core/plugins-workspace/claude/src/telemetry/bodies.js @@ -67,14 +67,14 @@ const removing = new Set() * same bytes or the exact same rejection, decided once, before `removing` * ever needs to arbitrate anything. * - * What it does not do is make the whole path race-free. An entry is dropped - * as soon as its read settles, which is before the first caller's `unlink` - * is even issued, so a caller that arrives inside that gap finds no entry, - * reads for itself, and can still see ENOENT and count `missing`. That - * caller never overlapped the read, and the gap it has to land in is the - * `unlink`'s own duration rather than the whole read's. Entries leave the - * map once the read settles, so it never outgrows the reads in flight. - * @type {Map>} + * An entry stays until every caller holding it has finished classifying, and + * for an unparseable body that includes the owner's `unlink`, so a caller that + * arrives while the delete is still in flight shares the bytes instead of + * reading a file that is about to go (issue #2053). Each entry counts its + * `users`; the last one out drops it, on every exit path, so the map never + * outgrows the reads in flight. A caller that arrives after the last one has + * left never overlapped the read, and reads for itself. + * @type {Map, users: number }>} */ const reading = new Map() @@ -124,8 +124,9 @@ export async function loadSpooledBodies(events, opts) { // dedup on its own: a ref that does not parse never lands there, so a batch // naming one twice reads it again after this call's own `unlink` took it off // the disk, and counts `missing` for the very ref the same result calls - // `unparseable`. The shared `reading` entry does not cover it either: that - // is dropped when the read settles, which is before the `unlink` is issued. + // `unparseable`. The shared `reading` entry does not cover it either: this + // call leaves it once the first occurrence is classified, so it is gone by + // the time a later occurrence in the same batch asks for it. /** @type {Set} */ const seen = new Set() @@ -142,69 +143,69 @@ export async function loadSpooledBodies(events, opts) { seen.add(ref) // Claimed synchronously, before any `await`: whichever overlapping call // for this `file` runs first creates and stores the read; a second call - // arriving before it resolves finds the same promise already in the map + // arriving before the last holder leaves finds the same entry in the map // and reuses it instead of racing its own `readFile` against the first // call's eventual `unlink`. - let pending = reading.get(file) - if (!pending) { - pending = fs.readFile(file) - reading.set(file, pending) - // A rejection (a missing file is the common case) is still handled by - // the caller below via its own `await pending`; this derived promise - // exists only for the cleanup side effect, so its own rejection is - // swallowed to avoid an unhandledRejection for a failure already - // being handled. - pending.finally(() => { - if (reading.get(file) === pending) reading.delete(file) - }).catch(() => {}) + let entry = reading.get(file) + if (!entry) { + entry = { read: fs.readFile(file), users: 0 } + reading.set(file, entry) } - /** @type {Buffer} */ - let raw + entry.users += 1 + const pending = entry.read try { - raw = await pending - } catch { - // Already projected, already evicted, or never written: the - // content is recoverable from the transcript either way. - // @ref LLP 0253#eviction-degrades [implements]: an evicted body is not an - // error; the failure mode is "captured later, with less detail" - missing += 1 - continue - } - const body = parseMaybeJson(raw.toString('utf8')) - if (!isPlainObject(body)) { - unparseable += 1 - // An overlapping read already owns this removal (see `removing`): it - // reports the bytes, or reports nothing if the file would not go. - if (removing.has(file)) continue - removing.add(file) + /** @type {Buffer} */ + let raw try { - // `unlink`, not `fs.rm(..., { force: true })`: a forced remove RESOLVES - // for a path that is already gone, and the bytes below are only ours to - // report if this call is the one that took the file off the disk. - await fs.unlink(file) - // Sized only once the file is gone: one that is still there (EPERM, a - // read-only spool) is still spooled, so subtracting its bytes would - // publish a spool smaller than the one the next sweep finds. - // @ref LLP 0257#status-and-health [implements]: S16 - `spool_bytes` is - // the spool's current size, so a deletion only the reader can see has - // to be sized for the caller - unparseableBytes += raw.length + raw = await pending } catch { - // Already gone, or not removable at all: either way this batch has no - // byte movement to report, and the content is recoverable from - // transcript backfill. - } finally { - removing.delete(file) + // Already projected, already evicted, or never written: the + // content is recoverable from the transcript either way. + // @ref LLP 0253#eviction-degrades [implements]: an evicted body is not an + // error; the failure mode is "captured later, with less detail" + missing += 1 + continue } - continue + const body = parseMaybeJson(raw.toString('utf8')) + if (!isPlainObject(body)) { + unparseable += 1 + // An overlapping read already owns this removal (see `removing`): it + // reports the bytes, or reports nothing if the file would not go. + if (removing.has(file)) continue + removing.add(file) + try { + // `unlink`, not `fs.rm(..., { force: true })`: a forced remove RESOLVES + // for a path that is already gone, and the bytes below are only ours to + // report if this call is the one that took the file off the disk. + await fs.unlink(file) + // Sized only once the file is gone: one that is still there (EPERM, a + // read-only spool) is still spooled, so subtracting its bytes would + // publish a spool smaller than the one the next sweep finds. + // @ref LLP 0257#status-and-health [implements]: S16 - `spool_bytes` is + // the spool's current size, so a deletion only the reader can see has + // to be sized for the caller + unparseableBytes += raw.length + } catch { + // Already gone, or not removable at all: either way this batch has no + // byte movement to report, and the content is recoverable from + // transcript backfill. + } finally { + removing.delete(file) + } + continue + } + bodies.set(ref, { + kind: event.name === 'api_request_body' ? 'request' : 'response', + file, + body, + }) + consumedFiles.push(file) + consumedBytes += raw.length + } finally { + // Every exit leaves here, the owner's only after its `unlink` settled. + entry.users -= 1 + if (entry.users === 0 && reading.get(file) === entry) reading.delete(file) } - bodies.set(ref, { - kind: event.name === 'api_request_body' ? 'request' : 'response', - file, - body, - }) - consumedFiles.push(file) - consumedBytes += raw.length } return { bodies, consumedFiles, consumedBytes, missing, unparseable, unparseableBytes, refused } diff --git a/test/plugins/claude-telemetry-unparseable-body.test.js b/test/plugins/claude-telemetry-unparseable-body.test.js index e10955bcc..2e56da5ca 100644 --- a/test/plugins/claude-telemetry-unparseable-body.test.js +++ b/test/plugins/claude-telemetry-unparseable-body.test.js @@ -283,7 +283,7 @@ test('overlapping loadSpooledBodies calls for one body_ref issue exactly one rea // `bodies`: the second event read the file again after the first event had // deleted it, saw ENOENT, and counted `missing`, so one result called one ref // both `unparseable` and `missing`. The shared `reading` entry does not -// cover it: it is dropped when the read settles, before the first `unlink`. +// cover it: the call leaves it once the first event is classified. test('one batch naming an unparseable body_ref twice counts it once, never missing', async () => { const dir = await fsp.mkdtemp(path.join(os.tmpdir(), 'hyp-claude-unparseable-dup-')) const realReadFile = fsp.readFile @@ -316,3 +316,98 @@ test('one batch naming an unparseable body_ref twice counts it once, never missi await fsp.rm(dir, { recursive: true, force: true }) } }) + +/** + * @param {string} file + */ +function bodyEvents(file) { + return /** @type {any} */ ([{ + name: 'api_request_body', + timestamp: '2026-08-17T19:31:00.000Z', + attributes: { body_ref: file, request_id: REQUEST_ID }, + }]) +} + +// Issue #2053. The second caller starts while the first is inside its `unlink`, +// which is after the shared read settled. Ordering is forced with gates rather +// than ticks: `unlink` holds until the second call has started, and any fresh +// `readFile` holds until the deletion is done, so a caller that does not share +// the first read always reaches the disk after the file is gone. +test('a caller that arrives during the owner\'s unlink shares the read and never counts missing', async (t) => { + const dir = await fsp.mkdtemp(path.join(os.tmpdir(), 'hyp-claude-unparseable-unlink-')) + const unlinking = Promise.withResolvers() + const release = Promise.withResolvers() + const unlinked = Promise.withResolvers() + try { + const content = 'not json at all' + const file = path.join(dir, 'broken.request.json') + await fsp.writeFile(file, content, 'utf8') + const readFile = fsp.readFile + const unlink = fsp.unlink + let reads = 0 + t.mock.method(fsp, 'readFile', async (/** @type {string} */ target) => { + if (target === file && ++reads > 1) await unlinked.promise + return readFile(target) + }) + t.mock.method(fsp, 'unlink', async (/** @type {string} */ target) => { + if (target !== file) return unlink(target) + unlinking.resolve(undefined) + await release.promise + try { + return await unlink(target) + } finally { + unlinked.resolve(undefined) + } + }) + const first = loadSpooledBodies(bodyEvents(file), { spoolDir: dir }) + await unlinking.promise + const second = loadSpooledBodies(bodyEvents(file), { spoolDir: dir }) + release.resolve(undefined) + const both = await Promise.all([first, second]) + assert.equal(both[0].unparseable + both[1].unparseable, 2, 'both callers classify it unparseable') + assert.equal(both[0].missing + both[1].missing, 0, 'neither caller counts it missing') + assert.equal(both[0].unparseableBytes + both[1].unparseableBytes, content.length, 'bytes reported once') + assert.equal(reads, 1, 'the second caller shares the first read') + await assert.rejects(fsp.stat(file), { code: 'ENOENT' }) + + // The shared entry is gone once both settled: a valid body written to the + // same path afterwards is read fresh and projects, not the stale bytes. + await fsp.writeFile(file, JSON.stringify({ messages: [] }), 'utf8') + const after = await loadSpooledBodies(bodyEvents(file), { spoolDir: dir }) + assert.equal(after.bodies.size, 1, 'a fresh read, not a leaked entry') + assert.equal(after.unparseable, 0) + assert.equal(reads, 2) + } finally { + release.resolve(undefined) + unlinked.resolve(undefined) + await fsp.rm(dir, { recursive: true, force: true }) + } +}) + +// The other exits (a missing file, a body that projects) leave the shared +// entry too, once every overlapping caller is done with it. +test('the shared read is released after overlapping callers on the missing and projected paths', async () => { + const dir = await fsp.mkdtemp(path.join(os.tmpdir(), 'hyp-claude-shared-read-release-')) + try { + const file = path.join(dir, 'body.request.json') + const missing = await Promise.all([ + loadSpooledBodies(bodyEvents(file), { spoolDir: dir }), + loadSpooledBodies(bodyEvents(file), { spoolDir: dir }), + ]) + assert.equal(missing[0].missing + missing[1].missing, 2) + + await fsp.writeFile(file, JSON.stringify({ model: 'first' }), 'utf8') + const projected = await Promise.all([ + loadSpooledBodies(bodyEvents(file), { spoolDir: dir }), + loadSpooledBodies(bodyEvents(file), { spoolDir: dir }), + ]) + assert.equal(projected[0].bodies.size + projected[1].bodies.size, 2, 'a released rejection, not a cached ENOENT') + + await fsp.writeFile(file, JSON.stringify({ model: 'second' }), 'utf8') + const again = await loadSpooledBodies(bodyEvents(file), { spoolDir: dir }) + const body = /** @type {any} */ (again.bodies.get(file))?.body + assert.equal(body?.model, 'second', 'a released read, not the earlier bytes') + } finally { + await fsp.rm(dir, { recursive: true, force: true }) + } +})