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 }) + } +})