diff --git a/docs/plans/2026-09-27-monitor-unexamined-run.md b/docs/plans/2026-09-27-monitor-unexamined-run.md new file mode 100644 index 000000000..8c090fad6 --- /dev/null +++ b/docs/plans/2026-09-27-monitor-unexamined-run.md @@ -0,0 +1,56 @@ +# A run the monitor store never examined is not empty (#1453) + +## Evidence +- Found while verifying #1411 (q115), with a probe: store B indexes the monitor folder at startup. A second store under another run id then creates `runs/run-a` and writes `incidents.json` + `operations.json`. B's next `maintain()` deletes `run-a` (ENOENT afterwards). +- Cause, `MonitorHistoryStore.maintain()`: the expired-run pass deletes every run folder not in `index` / `incidentRuns` / `unindexedRuns`. Those maps are filled only by startup indexing, so a run that appeared later is "not known", which the pass reads as "empty". +- Two app processes can share one data folder: `--packaging-smoke` skips the single-instance lock. +- It is the same "never seen means empty" shape the worker rule forbids (q109, q115). + +## Change +- `examinedRuns`: the runs startup indexing actually looked at. +- Retention deletes a run as empty only if it was examined. An unexamined run is unknown and waits for the next start to index it. +- `clear()` resets the set with the rest. + +## Not changed (residual) +- The capacity budget (`pruneRuns`) may still remove an unexamined run, as it already may for `unindexedRuns`. That is the documented policy: the 128 MiB ceiling wins over unknown runs. It orders an unexamined run as oldest, since it has no indexed points. +- ~~Two live stores can still examine each other's run while it is empty at startup.~~ Closed by review b's `touchedSince` (see below). + +## Test (real files) +`MonitorHistoryStore.test.ts`: run-a appears after store B indexed. +- It survives two of B's maintenance passes. This is red before the fix (ENOENT on the first pass). +- A restarted store examines it and keeps its in-retention incident. +- Past retention, its incident file is deleted, and its (then empty, aged) folder on a later pass, so the protection is not permanent. + +## Review a (round 1), fixed +- **An unreadable or untrusted incident file:** at indexing, an `incidents.json` that exists but cannot be read, parsed or trusted now makes the run UNKNOWN (`unindexedRuns`). It used to read as `[]`, so an examined run was deleted. +- **`examinedRuns` is forgotten when the run is deleted** (retention or capacity), so a name another store recreates with fresh data is unexamined again and kept. +- **Tests:** real files. An incident file at mode 000 during indexing survives maintenance once readable; a run recreated after its retention deletion survives. Both were red before. (Their mutation gates were shadowed by review b's `touchedSince` until review c's real-clock rewrite; see below.) + +## Review b (round 1), fixed +- **A tier `stat` failure other than ENOENT** now marks the run unknown instead of skipping the tier as absent. +- **A tier file with unparseable lines** marks the run unknown. It was indexed with no points, deleted as fully expired, and then its run was deleted. +- **Retention keeps any run with a file touched within the retention window** (`touchedSince`; any list or stat failure counts as touched). A run examined while empty can be filled later by the other store, which is residual 2 above, now closed. One side effect: once an expired run's last file is removed, its fresh folder mtime keeps the empty folder for one more window. +- **Tests:** real files, one per finding (ELOOP tier link, unparseable tier line, filled after examination). Each was red before. Mutations: "unparsed ignored" and "no touched check" fail on their own; the stat and touched guards back each other up on the ELOOP case (removing both fails). + +## Review a round 2: residual (manager decision) +A second store's write can land between retention's final `touchedSince()` check and its recursive `rm()`: a check-then-act race between two uncoordinated processes. It needs two app processes sharing one data folder (possible only under `--packaging-smoke`, which skips the single-instance lock) and a write inside that sub-millisecond window. Closing it needs a cross-process lock on the monitor folder. A rename-to-tombstone-then-recheck scheme narrows it but brings restore-collision cases of its own. That is left out under the PR freeze; B6 decides whether to accept this residual or require the lock. + +## Review c (round 1), fixed (tests and docs) +- **The problem:** every new test used a 1970-scale fake clock, so `touchedSince` saw every fixture as freshly touched and shadowed the other guards. Removing the `examinedRuns` guard itself (#1453's fix), the unknown-incidents guard, or the forget-on-delete survived the suite. +- **The fix:** the tests run on the real clock, and each fixture's files and folder are aged past retention with `utimes`, so only the guard a test names can keep the run. A cleanup-direction assertion is added: an examined run whose data expired loses its folder on a later pass. +- **Mutations, each killed on its own:** the `examinedRuns` guard dropped (2 red); `examinedRuns` never populated (2 red); examined kept after delete; unknown incidents unprotected; unparsed lines ignored; no touched check (2 red). Only the ELOOP case has two guards (indexing and `touchedSince`), as noted in its test. +- **Docs:** residual 2 is closed; "deleted as before" is corrected (the incident file goes first, the emptied folder on a later pass). + + +## Review b round 2, fixed +- **The problem:** the index is only a snapshot of another live store's run. Expiring or compacting a foreign tier file, or rewriting its incidents, from that snapshot deleted what the other store wrote afterwards, including content appended after indexing. +- **The fix:** each foreign tier and incident file's size and mtime are recorded at indexing (`foreignFiles`). Every expiry, compaction and incident rewrite of a foreign file first checks the file is unchanged. A changed or vanished file makes the run unknown (`unindexedRuns`) instead. After the store's own rewrite of a foreign file, the new fingerprint is recorded. The store's own run needs no check, since only it writes there. +- **Tests (two live stores on one folder, real files and clock):** + - a foreign tier expired from a stale index; + - content appended to a foreign tier after indexing (compaction); + - a fresh incident added after indexing (incident rewrite). + With the check disabled, the first two fail; removing only the incident-loop check fails the third. +- **Residual unchanged:** the sub-millisecond window between the check and the act (the review a round 2 cross-process residual). + +## Review b round 3, fixed +The global incident limit also rewrote a foreign incident file from the cached rows. The unchanged-file check now lives inside `writeRunIncidents`, so retention and the limit both pass it. Pinned by "never enforces the incident limit on another live store's changed file". Removing the check fails that test and the incident-retention test. diff --git a/src/main/performance/MonitorHistoryStore.listingFailure.test.ts b/src/main/performance/MonitorHistoryStore.listingFailure.test.ts index 8082bff9a..5902766ec 100644 --- a/src/main/performance/MonitorHistoryStore.listingFailure.test.ts +++ b/src/main/performance/MonitorHistoryStore.listingFailure.test.ts @@ -1,4 +1,4 @@ -import { mkdir, mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises' +import { mkdir, mkdtemp, readFile, readdir, rm, utimes, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, expect, it, vi } from 'vitest' @@ -47,11 +47,19 @@ it('keeps a run whose listing failed at startup, with its set-aside bytes intact const refused = JSON.stringify({ version: 2, incidents: ['evidence'] }) await writeFile(join(runOld, aside), refused) + // Real clock, and the run aged past retention (#1455 review c, B6 check + // 2110): maintenance keeps any run touched within the window, so with a + // 1970 snapshot the unindexed mark was never what kept this run. + const now = Date.now() + const old = new Date(now - 30 * 24 * 60 * 60_000) + await utimes(join(runOld, aside), old, old) + await utimes(runOld, old, old) + failOnce.path = runOld const store = new MonitorHistoryStore(root, 'run-now') await store.settled() expect(failOnce.path).toBeNull() - store.record(snapshot(90_000), null, [], 0, 0) + store.record(snapshot(now), null, [], 0, 0) await store.settled() expect(await readdir(runOld)).toContain(aside) diff --git a/src/main/performance/MonitorHistoryStore.test.ts b/src/main/performance/MonitorHistoryStore.test.ts index 68b13c0c3..88ccdbb2b 100644 --- a/src/main/performance/MonitorHistoryStore.test.ts +++ b/src/main/performance/MonitorHistoryStore.test.ts @@ -1,4 +1,4 @@ -import { appendFile, chmod, mkdir, mkdtemp, readFile, readdir, rm, stat, writeFile } from 'node:fs/promises' +import { appendFile, chmod, lstat, mkdir, mkdtemp, readFile, readdir, rm, stat, symlink, utimes, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, describe, expect, it, vi } from 'vitest' @@ -8,6 +8,20 @@ import { MonitorHistoryStore } from './MonitorHistoryStore.js' const roots: string[] = [] afterEach(async () => { await Promise.all(roots.splice(0).map(root => rm(root, { recursive: true, force: true }))) }) +// Retention keeps any run with a file touched within the window (touchedSince, +// #1455 review b), measured against the maintenance time, which is the +// snapshot's sampledAt. A retention test that wants ONE named guard to be the +// only thing keeping a run must therefore run on the real clock and age that +// run's files AND folder past the window; with a 1970-scale snapshot every +// fixture looks freshly touched and the named guard is shadowed (#1455 +// review c, B6 check 2110). +const DAY = 24 * 60 * 60_000 +const aged = async (dir: string, now: number): Promise => { + const old = new Date(now - 30 * DAY) + for (const name of await readdir(dir)) await utimes(join(dir, name), old, old) + await utimes(dir, old, old) +} + const snapshot = (at: number): MonitorWorkerSnapshot => ({ schemaVersion: 1, sampledAt: at, main: { at, cpuPercent: 2, rss: 1024, heapUsed: 256, heapLimit: 2048, loopMeanMs: 20, loopP99Ms: 22, loopMaxMs: 25, sleepGap: false }, @@ -195,14 +209,17 @@ describe('bounded local performance history', () => { const refused = JSON.stringify({ version: 2, incidents: [incident] }) await mkdir(runA, { recursive: true }) await writeFile(join(runA, 'incidents.json'), refused) + const now = Date.now() const first = new MonitorHistoryStore(root, 'run-a') await first.settled() - first.record(snapshot(11_000), null, [{ ...incident, id: 2, at: 11_000 }], 0, 1) + first.record(snapshot(now), null, [{ ...incident, id: 2, at: now }], 0, 1) await first.settled() + // Aged, so only refusedAsideRuns can keep run-a (see `aged`). + await aged(runA, now) const later = new MonitorHistoryStore(root, 'run-b') await later.settled() - later.record(snapshot(11_000 + 8 * 24 * 60 * 60_000), null, [], 0, 0) + later.record(snapshot(now + 8 * DAY), null, [], 0, 0) await later.settled() const aside = (await readdir(runA)).filter(name => name.startsWith('incidents.refused-')) @@ -285,16 +302,44 @@ describe('bounded local performance history', () => { await mkdir(refused, { recursive: true }) await writeFile(join(foreignOnly, 'incidents.json'), JSON.stringify([{ ...incident, rule: 'rule-from-a-newer-build' }])) await writeFile(join(refused, 'incidents.json'), JSON.stringify({ version: 2 })) + // Aged, so only foreignIncidents / refusedIncidentRuns can keep each run. + const now = Date.now() + await aged(foreignOnly, now) + await aged(refused, now) const store = new MonitorHistoryStore(root, 'run-now') await store.settled() - store.record(snapshot(90_000), null, [], 0, 0) + store.record(snapshot(now), null, [], 0, 0) await store.settled() await expect(stat(foreignOnly)).resolves.toBeTruthy() await expect(stat(refused)).resolves.toBeTruthy() }) + // B6 check 2110: the carried-rows path across two stores. Another run's + // file mixes a readable row that has expired with a row from a newer build. + // Retention rewrites the file (it is unchanged since indexing, so + // foreignChanged lets it) and must carry the unrecognised row, and the run + // must survive, although it is aged past the window. + it('carries an unrecognised row through retention of an aged, unchanged foreign file, and keeps its run', async () => { + const root = await mkdtemp(join(tmpdir(), 'agent-code-monitor-')) + roots.push(root) + const runA = join(root, 'runs', 'run-a') + await mkdir(runA, { recursive: true }) + const now = Date.now() + const carried = { ...incident, id: 7, at: now - 20 * DAY, rule: 'rule-from-a-newer-build' } + await writeFile(join(runA, 'incidents.json'), JSON.stringify([{ ...incident, id: 1, at: now - 20 * DAY }, carried])) + await aged(runA, now) + + const store = new MonitorHistoryStore(root, 'run-b') + await store.settled() + store.record(snapshot(now), null, [], 0, 0) + await store.settled() + + expect(JSON.parse(await readFile(join(runA, 'incidents.json'), 'utf8'))).toEqual([carried]) + await expect(stat(runA)).resolves.toBeTruthy() + }) + // Review of #1411 (a): with the run's file full of carried rows, a new // readable incident has no room. Keeping the carried rows is right; hiding // that the new one was not kept is not. @@ -344,3 +389,238 @@ describe('bounded local performance history', () => { expect((await store.query(10 * 60_000, 16 * 60_000, undefined, 7)).resolution).toBe('1s') }) }) + +// #1453 (q115 "unknown is never empty"): retention deleted any run folder +// its in-memory index did not know. A run created AFTER this store indexed, +// by a second store sharing the folder (`--packaging-smoke` skips the +// single-instance lock), was never examined, so it looked empty and was +// deleted on the next maintenance pass. +// +// WHY the real clock and aged fixtures (#1455 review c): retention also keeps +// any run touched within the retention window (touchedSince, review b). With a +// 1970-scale fake clock every fixture looked freshly touched, so that guard +// shadowed every other one and their mutations survived. Here each fixture's +// files AND folder are aged past retention, so only the guard a test names +// can keep the run. +describe('a run this store never examined', () => { + const setup = async () => { + const root = await mkdtemp(join(tmpdir(), 'monitor-unexamined-')) + roots.push(root) + const foreign = join(root, 'runs', 'run-a') + return { root, foreign } + } + + it('is kept by retention until a later start examines it, and its folder is removed once expired', async () => { + const { root, foreign } = await setup() + let now = Date.now() + const first = new MonitorHistoryStore(root, 'run-b', () => now) + await first.settled() + // What the other store writes once it starts, with old mtimes (a copied + // or restored folder): only the examinedRuns guard can keep it. + await mkdir(foreign, { recursive: true }) + await writeFile(join(foreign, 'incidents.json'), JSON.stringify([{ ...incident, at: now }])) + await writeFile(join(foreign, 'operations.json'), '[]') + await aged(foreign, now) + first.record(snapshot(now), null, [], 0, 0) + await first.settled() + expect(JSON.parse(await readFile(join(foreign, 'incidents.json'), 'utf8'))).toHaveLength(1) + + // A later start examines run-a; its incident is still within retention. + const restarted = new MonitorHistoryStore(root, 'run-c', () => now) + restarted.record(snapshot(now), null, [], 0, 0) + await restarted.settled() + expect(JSON.parse(await readFile(join(foreign, 'incidents.json'), 'utf8'))).toHaveLength(1) + + // Past retention, the examined run's incident goes, and then its folder: + // the protection is not permanent (review c pins this direction). + now += 8 * DAY + restarted.record(snapshot(now), null, [], 0, 0) + await restarted.settled() + await expect(readFile(join(foreign, 'incidents.json'), 'utf8')).rejects.toMatchObject({ code: 'ENOENT' }) + now += 2 * 60_000 + restarted.record(snapshot(now), null, [], 0, 0) + await restarted.settled() + await expect(readdir(foreign)).rejects.toMatchObject({ code: 'ENOENT' }) + }) + + // #1455 review a (1): an incident file that cannot be read at indexing is + // UNKNOWN; the run was examined but its contents are not known. + it('keeps a run whose incidents could not be read at indexing', async () => { + const { root, foreign } = await setup() + const now = Date.now() + await mkdir(foreign, { recursive: true }) + await writeFile(join(foreign, 'incidents.json'), JSON.stringify([{ ...incident, at: now - 30 * DAY }])) + await aged(foreign, now) + await chmod(join(foreign, 'incidents.json'), 0o000) + const store = new MonitorHistoryStore(root, 'run-b', () => now) + try { + await store.settled() + } finally { + await chmod(join(foreign, 'incidents.json'), 0o600) + } + await aged(foreign, now) + store.record(snapshot(now), null, [], 0, 0) + await store.settled() + expect(JSON.parse(await readFile(join(foreign, 'incidents.json'), 'utf8'))).toHaveLength(1) + }) + + // #1455 review a (2): `examinedRuns` names a run, not what this store saw. + // Recreated by another store after its deletion, the run is unexamined. + it('does not treat a run recreated after its deletion as examined', async () => { + const { root, foreign } = await setup() + let now = Date.now() + await mkdir(foreign, { recursive: true }) + await writeFile(join(foreign, 'incidents.json'), JSON.stringify([{ ...incident, at: now - 30 * DAY }])) + await aged(foreign, now) + const store = new MonitorHistoryStore(root, 'run-b', () => now) + store.record(snapshot(now), null, [], 0, 0) + await store.settled() + // The expired incident file went; the emptied folder's fresh mtime keeps + // it one more pass, so age it and let retention remove it. + await aged(foreign, now) + now += 2 * 60_000 + store.record(snapshot(now), null, [], 0, 0) + await store.settled() + await expect(readdir(foreign)).rejects.toMatchObject({ code: 'ENOENT' }) + // The other store recreates run-a (old mtimes: only the forgotten + // examination can keep it). + await mkdir(foreign, { recursive: true }) + await writeFile(join(foreign, 'incidents.json'), JSON.stringify([{ ...incident, at: now }])) + await aged(foreign, now) + now += 2 * 60_000 + store.record(snapshot(now), null, [], 0, 0) + await store.settled() + expect(JSON.parse(await readFile(join(foreign, 'incidents.json'), 'utf8'))).toHaveLength(1) + }) + + // #1455 review b (1): a tier file whose stat fails with anything but ENOENT + // (ELOOP from a self-referencing link) makes the run unknown. Two guards hold + // here: indexing marks the run unknown, and touchedSince counts an + // unstat-able file as touched; removing one alone survives. + it('keeps a run whose tier file cannot be stat-ed at indexing', async () => { + const { root, foreign } = await setup() + const now = Date.now() + await mkdir(foreign, { recursive: true }) + await symlink(join(foreign, '1m.jsonl'), join(foreign, '1m.jsonl')) + const store = new MonitorHistoryStore(root, 'run-b', () => now) + store.record(snapshot(now), null, [], 0, 0) + await store.settled() + expect((await lstat(join(foreign, '1m.jsonl'))).isSymbolicLink()).toBe(true) + }) + + // #1455 review b (2): content the store cannot parse is unknown, not + // expired. Aged, so only the "unparsed" guard can keep it. + it('keeps a run whose tier file holds content it cannot parse', async () => { + const { root, foreign } = await setup() + const now = Date.now() + await mkdir(foreign, { recursive: true }) + await writeFile(join(foreign, '1m.jsonl'), '{"private":"unparseable-point"}\n') + await aged(foreign, now) + const store = new MonitorHistoryStore(root, 'run-b', () => now) + store.record(snapshot(now), null, [], 0, 0) + await store.settled() + expect(await readFile(join(foreign, '1m.jsonl'), 'utf8')).toContain('unparseable-point') + }) + + // #1455 review b (3): a run examined while EMPTY can be filled later by the + // other store; its fresh files keep it (touchedSince). + it('keeps a run that was empty when examined and filled afterwards', async () => { + const { root, foreign } = await setup() + const now = Date.now() + await mkdir(foreign, { recursive: true }) + await aged(foreign, now) + const store = new MonitorHistoryStore(root, 'run-b', () => now) + await store.settled() + await writeFile(join(foreign, 'incidents.json'), JSON.stringify([{ ...incident, at: now }])) + store.record(snapshot(now), null, [], 0, 0) + await store.settled() + expect(JSON.parse(await readFile(join(foreign, 'incidents.json'), 'utf8'))).toHaveLength(1) + }) + + // #1455 review b round 2 (1): two LIVE stores. B indexed A's tier file when + // it held only an old point; A then appended a current point. B's retention + // expired the file from its stale index and deleted A's fresh data. A + // foreign file that changed since indexing now makes the run unknown. + it('never expires another live store\'s tier file from a stale index', async () => { + const { root } = await setup() + const now = Date.now() + let aNow = now - 2 * 60 * 60_000 + const a = new MonitorHistoryStore(root, 'run-a', () => aNow) + a.record(snapshot(aNow), null, [], 0, 0) + await a.flush() + const b = new MonitorHistoryStore(root, 'run-b', () => now) + await b.settled() + aNow = now + a.record(snapshot(aNow), null, [], 0, 0) + await a.flush() + const tier = join(root, 'runs', 'run-a', '1s.jsonl') + const before = await readFile(tier, 'utf8') + b.record(snapshot(now), null, [], 0, 0) + await b.settled() + expect(await readFile(tier, 'utf8')).toBe(before) + }) + + // #1455 review b round 2 (2): content appended to a foreign tier after + // indexing (here, a record B cannot parse) must not be discarded by B's + // compaction of its stale view of that file. + it('never compacts away content appended to a foreign tier after indexing', async () => { + const { root, foreign } = await setup() + const now = Date.now() + // A real point line from the store's own writer, re-dated: one expired and + // one current, so B's view of the file is due for compaction. + const source = new MonitorHistoryStore(join(root, 'source'), 'run-src', () => now) + source.record(snapshot(now), null, [], 0, 0) + await source.flush() + const line = (await readFile(join(root, 'source', 'runs', 'run-src', '1s.jsonl'), 'utf8')).trim().split('\n')[0]! + const at = (value: number) => JSON.stringify({ ...JSON.parse(line), at: value }) + await mkdir(foreign, { recursive: true }) + const tier = join(foreign, '1s.jsonl') + await writeFile(tier, `${at(now - 20 * 60_000)}\n${at(now)}\n`) + const b = new MonitorHistoryStore(root, 'run-b', () => now) + await b.settled() + await appendFile(tier, '{"private":"appended-after-indexing"}\n') + b.record(snapshot(now), null, [], 0, 0) + await b.settled() + expect(await readFile(tier, 'utf8')).toContain('appended-after-indexing') + }) + + // #1455 review b round 2 (1, incidents): B cached run-a's only incident, + // which is expiring; the other store then added a fresh one. B's retention + // rewrote the file from its cache and removed the fresh incident. + it('never rewrites another live store\'s incidents from a stale index', async () => { + const { root, foreign } = await setup() + const now = Date.now() + await mkdir(foreign, { recursive: true }) + const file = join(foreign, 'incidents.json') + const expiring = { ...incident, at: now - 8 * DAY } + await writeFile(file, JSON.stringify([expiring])) + const b = new MonitorHistoryStore(root, 'run-b', () => now) + await b.settled() + await writeFile(file, JSON.stringify([expiring, { ...incident, id: 2, at: now }])) + b.record(snapshot(now), null, [], 0, 0) + await b.settled() + expect(JSON.parse(await readFile(file, 'utf8'))).toHaveLength(2) + }) + + // #1455 review b round 3: the GLOBAL incident limit (fifty across runs) + // evicted from B's cached copy of A's incidents and rewrote A's file, + // removing the incident A had just added. + it('never enforces the incident limit on another live store\'s changed file', async () => { + const { root } = await setup() + const now = Date.now() + const many = (ids: number[]) => ids.map(id => ({ ...incident, id, at: now - (100 - id) * 1000 })) + const a = new MonitorHistoryStore(root, 'run-a', () => now) + a.record(snapshot(now), null, many(Array.from({ length: 50 }, (_, i) => i + 1)), 0, 0) + await a.flush() + const b = new MonitorHistoryStore(root, 'run-b', () => now) + await b.settled() + a.record(snapshot(now), null, many(Array.from({ length: 50 }, (_, i) => i + 2)), 0, 1) + await a.flush() + const file = join(root, 'runs', 'run-a', 'incidents.json') + expect((JSON.parse(await readFile(file, 'utf8')) as Array<{ id: number }>).map(row => row.id)).toContain(51) + b.record(snapshot(now), null, [{ ...incident, id: 900, at: now }], 0, 1) + await b.settled() + expect((JSON.parse(await readFile(file, 'utf8')) as Array<{ id: number }>).map(row => row.id)).toContain(51) + }) +}) + diff --git a/src/main/performance/MonitorHistoryStore.ts b/src/main/performance/MonitorHistoryStore.ts index b1b96b36e..10082db42 100644 --- a/src/main/performance/MonitorHistoryStore.ts +++ b/src/main/performance/MonitorHistoryStore.ts @@ -106,6 +106,21 @@ export class MonitorHistoryStore { // Runs with a file that could not be indexed. Retention must treat them as // unknown, not empty; only the capacity budget may still remove them. private unindexedRuns = new Set() + // Runs startup indexing actually looked at (#1453). Retention may delete a + // run as empty only if it was examined. A run that appeared later (a second + // store sharing this folder: `--packaging-smoke` skips the single-instance + // lock) is UNKNOWN, not empty, and waits for the next start to index it. + // Only the capacity budget may still remove it, as with unindexedRuns. + private examinedRuns = new Set() + // Size and mtime of each OTHER run's tier and incident file as this store + // indexed it (#1455 review b round 2). Another live store sharing the folder + // may still be writing that run, so the index is only a snapshot of it: + // expiring, compacting or rewriting a foreign file from that snapshot + // deleted what the other store wrote afterwards. Every such action first + // checks the file is unchanged; a changed (or vanished) file makes the run + // unknown for retention instead. The store's own run needs no check: only + // this store writes it. + private foreignFiles = new Map() private rollups: Record = { '1s': new TierRollup('1s'), '10s': new TierRollup('10s'), '1m': new TierRollup('1m') } // Coalesced work. Incidents and operations are whole-value replacements, so // only the newest value matters; the old promise chain queued one rewrite @@ -331,7 +346,7 @@ export class MonitorHistoryStore { await mkdir(this.runDir, { recursive: true }) this.index.clear(); this.incidentRuns.clear(); this.foreignIncidents.clear(); this.refusedIncidentRuns.clear(); this.refusedAsideRuns.clear(); this.repairedTails.clear(); this.indexed = true this.bytes = 0; this.shortened = false; this.degraded = false - this.operationFingerprint = ''; this.unindexedRuns.clear() + this.operationFingerprint = ''; this.unindexedRuns.clear(); this.examinedRuns.clear(); this.foreignFiles.clear() this.lastMaintenanceAt = -Infinity } catch { this.degraded = true } return this.status() @@ -385,6 +400,7 @@ export class MonitorHistoryStore { try { await this.cleanupTemps() for (const run of await this.runNames()) { + this.examinedRuns.add(run) // A failed listing is UNKNOWN, not empty (steering q115): read as "no // set-aside file", a prior run holding only one had no marker left and // the next maintenance deleted it with the refused bytes. Such a run is @@ -406,15 +422,28 @@ export class MonitorHistoryStore { // Repair before indexing: a torn final append from a crashed helper // is expected crash residue, not corruption worth a degraded state. await this.repairTail(file) - const size = await stat(file).then(value => value.size, () => null) - if (size === null) continue + // Only a missing tier is absent (#1455 review b): any other stat + // failure throws into the catch below, which marks the run unknown. + const fileStat = await stat(file).then(value => value, (error: NodeJS.ErrnoException) => { + if (error.code === 'ENOENT') return null + throw error + }) + if (fileStat === null) continue + const size = fileStat.size + if (run !== this.runId) this.foreignFiles.set(file, `${fileStat.size}:${fileStat.mtimeMs}`) const entry: FileStat = { run, resolution, points: 0, bytes: size, oldestAt: null, newestAt: null } const failure = { failed: false } + let unparsed = false for await (const line of this.lines(file, failure)) { const point = this.parseLine(line) if (point) this.notePoint(entry, point) + else unparsed = true } if (failure.failed) throw new Error('index-read-failed') + // Content it cannot parse is UNKNOWN, not expired (#1455 review b): + // indexed as a file with no points, retention deleted it as fully + // expired and then the run. Unindexed, retention leaves both alone. + if (unparsed) throw new Error('index-unparseable') this.index.set(file, entry) } catch { this.degraded = true @@ -423,6 +452,9 @@ export class MonitorHistoryStore { } const file = join(this.root, RUNS_DIR, run, 'incidents.json') const stored = await this.readIncidentFile(file, run) + // Fingerprint another run's incident file as indexed (#1455 review b + // round 2): later rewrites check it is unchanged (foreignChanged). + if (run !== this.runId) await this.noteForeign(file) // Every retained run is repaired, not only the current one. A crash or // force-quit in ANY earlier run left its last capture as "capturing" // forever, and only a helper restart within the same run fixed it. @@ -578,11 +610,13 @@ export class MonitorHistoryStore { private async maintain(now: number): Promise { for (const [file, entry] of [...this.index]) { const cutoff = now - RETENTION[entry.resolution] + if (await this.foreignChanged(file, entry.run)) continue if (entry.newestAt === null || entry.newestAt < cutoff) { // Fully expired: delete instead of rewriting an empty file forever. await rm(file, { force: true }) this.index.delete(file) this.repairedTails.delete(file) + this.foreignFiles.delete(file) continue } if (entry.oldestAt === null || entry.oldestAt >= cutoff) continue @@ -592,7 +626,10 @@ export class MonitorHistoryStore { // roughly uniform in time, so the expired share of the span estimates // the expired share of the file without reading it. const expired = (cutoff - entry.oldestAt) / Math.max(1, entry.newestAt - entry.oldestAt) - if (expired >= 0.25) await this.compact(file, entry, cutoff) + if (expired >= 0.25) { + await this.compact(file, entry, cutoff) + if (entry.run !== this.runId) await this.noteForeign(file) + } } const incidentCutoff = now - RETENTION['1m'] for (const [run, rows] of [...this.incidentRuns]) { @@ -603,8 +640,15 @@ export class MonitorHistoryStore { // with no remaining points or incidents holds only an unattributable // operations snapshot, so it is retention-expired, not capacity-pruned. if (this.indexed) for (const run of await this.runNames()) { - if (run === this.runId || this.incidentRuns.has(run) || this.foreignIncidents.has(run) || this.refusedIncidentRuns.has(run) || this.refusedAsideRuns.has(run) || this.unindexedRuns.has(run) || [...this.index.values()].some(entry => entry.run === run)) continue + if (run === this.runId || !this.examinedRuns.has(run) || this.incidentRuns.has(run) || this.foreignIncidents.has(run) || this.refusedIncidentRuns.has(run) || this.refusedAsideRuns.has(run) || this.unindexedRuns.has(run) || [...this.index.values()].some(entry => entry.run === run)) continue + // Examined while empty is not "still empty" (#1455 review b): another + // store sharing the folder can fill the run after this one indexed it. + // Keep any run with a file touched within the retention window. + if (await this.touchedSince(run, now - RETENTION['1m'])) continue await rm(join(this.root, RUNS_DIR, run), { recursive: true, force: true }) + // Examined means THIS contents (#1455 review a): once deleted, the name + // may come back with fresh data from another store, unexamined. + this.examinedRuns.delete(run) } this.bytes = await this.diskBytes() await this.pruneRuns(DATA_BUDGET) @@ -636,8 +680,16 @@ export class MonitorHistoryStore { this.index.set(file, next) } + /** + * Replace (or remove) a run's incident file with `rows`. For ANOTHER run the + * file must be unchanged since this store last saw it (#1455 review b rounds + * 2 and 3): both retention and the global incident limit rewrite foreign + * files from the cached rows, and a changed file holds incidents the other + * live store wrote afterwards. The check lives here so no caller can skip it. + */ private async writeRunIncidents(run: string, rows: MonitorIncident[]): Promise { const file = join(this.root, RUNS_DIR, run, 'incidents.json') + if (await this.foreignChanged(file, run)) return if (!rows.length && !this.foreignIncidents.has(run)) { await rm(file, { force: true }) this.incidentRuns.delete(run) @@ -645,6 +697,7 @@ export class MonitorHistoryStore { if (rows.length) this.incidentRuns.set(run, rows) else this.incidentRuns.delete(run) } + if (run !== this.runId) await this.noteForeign(file) } /** Fifty incidents across all retained runs, newest first. */ @@ -821,6 +874,39 @@ export class MonitorHistoryStore { } } + /** Record a foreign file's current size and mtime (after indexing it, or + * after this store rewrote it). A file that is gone is forgotten. */ + private async noteForeign(file: string): Promise { + const current = await stat(file).catch(() => null) + if (current) this.foreignFiles.set(file, `${current.size}:${current.mtimeMs}`) + else this.foreignFiles.delete(file) + } + + /** True, and the run marked unknown, when a foreign file changed since this + * store last saw it (or cannot be stat-ed). Always false for the own run. */ + private async foreignChanged(file: string, run: string): Promise { + if (run === this.runId) return false + const seen = this.foreignFiles.get(file) + const current = await stat(file).then(value => `${value.size}:${value.mtimeMs}`, () => null) + if (seen !== undefined && current === seen) return false + this.unindexedRuns.add(run) + return true + } + + /** Whether any file in the run changed at or after `since`. Unknown (any + * failure to list or stat) counts as touched, so retention keeps the run. */ + private async touchedSince(run: string, since: number): Promise { + const dir = join(this.root, RUNS_DIR, run) + try { + for (const file of await readdir(dir, { withFileTypes: true })) { + if ((await stat(join(dir, file.name))).mtimeMs >= since) return true + } + return (await stat(dir)).mtimeMs >= since + } catch { + return true + } + } + private async runBytes(run: string): Promise { const dir = join(this.root, RUNS_DIR, run) let total = 0 @@ -851,6 +937,7 @@ export class MonitorHistoryStore { this.refusedIncidentRuns.delete(run) this.refusedAsideRuns.delete(run) this.unindexedRuns.delete(run) + this.examinedRuns.delete(run) total = Math.max(0, total - size); this.shortened = true } this.bytes = total