diff --git a/src/cache-refresh-lock.ts b/src/cache-refresh-lock.ts index a0120098..58faf281 100644 --- a/src/cache-refresh-lock.ts +++ b/src/cache-refresh-lock.ts @@ -1,4 +1,4 @@ -import { randomBytes } from 'crypto' +import { createHash, randomBytes } from 'crypto' import { existsSync } from 'fs' import { mkdir, open, readFile, stat, unlink, utimes, writeFile } from 'fs/promises' import { homedir } from 'os' @@ -81,6 +81,13 @@ async function retryWindowsMutation(operation: () => Promise, sleep: (ms: return false } +// The directory entry becomes visible before the awaited body write, so the +// file is briefly observable at zero bytes. Deliberately left as is: a corrupt +// body is only ever recovered once its mtime is older than staleMs, and this +// window is milliseconds wide on a file whose mtime is by definition now, so +// no observer can reach the age gate through it. Closing it would mean +// link()ing a temp file into place, which is not portable to filesystems +// without hard links. async function createExclusive(path: string, body: string): Promise<'created' | 'exists' | 'unavailable'> { try { const handle = await open(path, 'wx', 0o600) @@ -92,7 +99,24 @@ async function createExclusive(path: string, body: string): Promise<'created' | } } -type Observation = { record: LockRecord; mtimeMs: number } +// A null record is a body whose stat bracket agreed across the read and that +// still does not parse into a lock record: a corrupt leftover of 0 bytes, a +// truncation, or a wrong shape. The bracket is a heuristic, not proof that the +// read was whole — a same-size rewrite moves neither size nor (on a coarse +// filesystem) mtime — which is why nothing here treats a single read as +// authoritative. It owns nothing, but it is a real file with a +// real mtime, not an infrastructure failure — classifying it 'unavailable' +// routed every later refresh to the read-only path and froze ingestion. It +// carries no authority: it is only ever recovered through the unmodified +// staleness gate, exactly like an abandoned but well-formed lock. +// +// `digest` fingerprints the exact bytes. A corrupt body has no token, so +// token equality between two corrupt observations degenerates to +// `undefined === undefined`, and mtime granularity is coarse on some +// filesystems (measured on macOS: a 2s grid on FAT32, 10ms on exFAT, sub-ms on +// APFS — and on all three a same-size rewrite moves neither mtime nor size), so +// mtime is not a reliable change signal on its own. +type Observation = { record: LockRecord | null; mtimeMs: number; digest: string } type ObservationResult = Observation | 'missing' | 'changing' | 'unavailable' async function observe(path: string): Promise { @@ -100,6 +124,7 @@ async function observe(path: string): Promise { // written, and heartbeat rewrites briefly truncate it. Treat that bounded // transition as contention, not broken infrastructure. let sawChange = false + let corrupt: Observation | null = null for (let attempt = 0; attempt < 3; attempt++) { try { const before = await stat(path) @@ -110,10 +135,23 @@ async function observe(path: string): Promise { await delay(1) continue } - const parsed = JSON.parse(raw) as Partial - if (typeof parsed.pid === 'number' && typeof parsed.token === 'string' && typeof parsed.at === 'number') { - return { record: { pid: parsed.pid, token: parsed.token, at: parsed.at }, mtimeMs: after.mtimeMs } + const digest = createHash('sha1').update(raw).digest('hex') + // A body that is valid JSON of the wrong shape is corrupt like any other, + // including one written by a future version with a different record + // shape. That is safe precisely because staleness is never waived: a + // foreign version's LIVE lock keeps its mtime fresh through its own + // heartbeat, so it is never taken — both versions just degrade to the + // read-only path. Only an abandoned one is recovered, and a lock record + // is per-run state with nothing in it worth preserving. + let parsed: Partial | undefined + try { parsed = JSON.parse(raw) as Partial } catch { parsed = undefined } + if (parsed && typeof parsed.pid === 'number' && typeof parsed.token === 'string' && typeof parsed.at === 'number') { + return { record: { pid: parsed.pid, token: parsed.token, at: parsed.at }, mtimeMs: after.mtimeMs, digest } } + // Keep the most recent corrupt read. It is not evidence of stability on + // its own: tryTakeover re-observes under the guard and compares with + // sameObservation before acting, so stability is proven there, not here. + corrupt = { record: null, mtimeMs: after.mtimeMs, digest } } catch (err) { if (isMissingError(err)) return 'missing' const code = (err as NodeJS.ErrnoException | undefined)?.code @@ -121,11 +159,19 @@ async function observe(path: string): Promise { } await delay(1) } - return sawChange ? 'changing' : 'unavailable' + // Contention outranks corruption: a body seen mid-rewrite is a live owner's, + // and the caller must poll rather than treat it as recoverable. + if (sawChange) return 'changing' + return corrupt ?? 'unavailable' } function sameObservation(a: Observation, b: Observation): boolean { - return a.record.token === b.record.token && a.mtimeMs === b.mtimeMs + // A corrupt body and an owned one are never "the same observation", even + // though `a.record?.token === b.record?.token` cannot tell them apart once + // both sides are corrupt. Compare that boundary explicitly, then require the + // bytes themselves to match, so "unchanged" survives a coarse mtime. + if ((a.record === null) !== (b.record === null)) return false + return a.record?.token === b.record?.token && a.mtimeMs === b.mtimeMs && a.digest === b.digest } let singleFlightTail: Promise = Promise.resolve() @@ -209,7 +255,7 @@ export async function acquireCacheRefreshLock(options: RefreshLockOptions = {}): if (current === 'missing') return true if (current === 'changing') return false if (current === 'unavailable') return false - if (current.record.token !== token) return true + if (current.record?.token !== token) return true return retryWindowsMutation(() => unlink(lockPath), sleep) } finally { await retryWindowsMutation(() => unlink(takeoverPath), sleep) @@ -221,7 +267,7 @@ export async function acquireCacheRefreshLock(options: RefreshLockOptions = {}): if (guard !== 'created') return false try { const current = await observe(lockPath) - return current !== 'missing' && current !== 'changing' && current !== 'unavailable' && current.record.token === token + return current !== 'missing' && current !== 'changing' && current !== 'unavailable' && current.record?.token === token } finally { await retryWindowsMutation(() => unlink(takeoverPath), sleep) } @@ -238,7 +284,23 @@ export async function acquireCacheRefreshLock(options: RefreshLockOptions = {}): if (guard !== 'created') { heartbeatRunning = false; return } try { const current = await observe(lockPath) - if (current === 'missing' || current === 'changing' || current === 'unavailable' || current.record.token !== token) return + if (current === 'missing' || current === 'changing' || current === 'unavailable') return + // A corrupt body is NOT ours to rewrite, even though no parseable + // token contradicts us. Holding the takeover guard excludes the other + // guard-takers, but NOT createExclusive, which publishes a directory + // entry before its body — so an unparseable body may be a successor's + // lock a millisecond from being written, or a foreign version's whose + // record shape we cannot read. Stamping our token over it made this + // process an owner again after it had been legitimately replaced: + // verifyStillOwner then answered true for a displaced writer, and + // release()'s removeIfOwned deleted the live successor's lock. + // + // So a body we cannot prove is ours ends our ownership. The mtime + // stops advancing, the fence refuses to publish (the parse is + // discarded, which is the fail-safe direction), and a successor + // recovers the lock one staleMs later through the age gate. Losing a + // parse is the correct price for never having two owners. + if (current.record === null || current.record.token !== token) return await writeFile(lockPath, body(), { encoding: 'utf-8' }) const now = new Date(clock.wallNow()) await utimes(lockPath, now, now) @@ -325,6 +387,12 @@ export async function acquireCacheRefreshLock(options: RefreshLockOptions = {}): continue } + // A corrupt observation takes this path unchanged. Staleness is never + // waived for it: an abandoned corrupt lock is older than staleMs and is + // recovered here, while a corrupt body younger than that is waited out + // and left alone, because it may belong to a live owner whose heartbeat + // will repair it. Worst case we time out and serve the prior snapshot + // read-only for one staleMs window instead of freezing forever. const age = Math.max(0, clock.wallNow() - observation.mtimeMs) if (age > staleMs) { const takeover = await tryTakeover(observation) diff --git a/tests/cache-refresh-lock-corrupt-body.test.ts b/tests/cache-refresh-lock-corrupt-body.test.ts new file mode 100644 index 00000000..c4a77021 --- /dev/null +++ b/tests/cache-refresh-lock-corrupt-body.test.ts @@ -0,0 +1,121 @@ +import { spawn, type ChildProcess } from 'child_process' +import { existsSync } from 'fs' +import { mkdir, mkdtemp, readFile, rm, stat, writeFile } from 'fs/promises' +import { tmpdir } from 'os' +import { join } from 'path' +import { afterEach, describe, expect, it } from 'vitest' + +import { acquireCacheRefreshLock } from '../src/cache-refresh-lock.js' + +// Recovering a corrupt session-refresh.lock must never cost the two design +// commitments the lock exists for: it may not fail open into mutation, and it +// may not be stolen from a live heartbeating owner. An earlier fix waived the +// staleness gate for a corrupt body once the contender's wait expired, which +// handed two processes the lock at the same time in both directions below. +// Corruption is recovered only through the unmodified staleness gate. + +const roots: string[] = [] +afterEach(async () => { + while (roots.length) await rm(roots.pop()!, { recursive: true, force: true }) +}) + +async function tempCase(prefix: string): Promise<{ cacheDir: string; barriers: string; lockPath: string }> { + const root = await mkdtemp(join(tmpdir(), prefix)) + roots.push(root) + const cacheDir = join(root, 'cache') + const barriers = join(root, 'barriers') + await mkdir(cacheDir, { recursive: true }) + await mkdir(barriers, { recursive: true }) + return { cacheDir, barriers, lockPath: join(cacheDir, 'session-refresh.lock') } +} + +function spawnFixture(fixture: string, args: string[], env: NodeJS.ProcessEnv = {}): ChildProcess { + return spawn(process.execPath, ['--import', 'tsx', join(process.cwd(), 'tests/fixtures', fixture), ...args], { + cwd: process.cwd(), + stdio: ['ignore', 'ignore', 'inherit'], + env: { ...process.env, ...env }, + }) +} + +async function waitForFile(path: string, timeoutMs = 15_000): Promise { + const deadline = Date.now() + timeoutMs + while (!existsSync(path)) { + if (Date.now() > deadline) throw new Error(`timed out waiting for ${path}`) + await new Promise(resolve => { setTimeout(resolve, 5) }) + } +} + +const exited = (child: ChildProcess): Promise => new Promise(resolve => child.once('exit', resolve)) + +describe('warm refresh lock: a corrupt body never displaces a live owner', () => { + it('leaves the zero-byte lock alone while its creator is still inside createExclusive', async () => { + const { cacheDir, barriers, lockPath } = await tempCase('cb-refresh-corrupt-race-') + // UV_THREADPOOL_SIZE=1 plus the fixture's pbkdf2 churn stretches the gap + // between open(path,'wx') and the awaited body write, so the lock is + // genuinely observable at zero bytes with no external corruption at all. + const owner = spawnFixture('cache-refresh-slow-owner.ts', [cacheDir, barriers], { UV_THREADPOOL_SIZE: '1' }) + try { + const deadline = Date.now() + 20_000 + let sawZeroByteLock = false + while (Date.now() < deadline) { + const info = await stat(lockPath).catch(() => null) + if (info) { sawZeroByteLock = info.size === 0; break } + await new Promise(resolve => { setTimeout(resolve, 1) }) + } + expect(sawZeroByteLock, 'never caught the owner mid-createExclusive').toBe(true) + + const contender = await acquireCacheRefreshLock({ cacheDir, waitMs: 40, pollMs: 5, staleMs: 90_000 }) + if (contender.outcome === 'acquired') await contender.handle.release() + expect(contender.outcome).toBe('timed-out') + } finally { + await exited(owner) + } + // The owner kept the lock end to end, so its publication fence held. + expect(existsSync(join(barriers, 'owner.acquired'))).toBe(true) + expect(existsSync(join(barriers, 'owner.verify.true')), 'owner lost its own lock').toBe(true) + }, 60_000) + + it('leaves a live owner alone after its body is truncated, however long the wait', async () => { + const { cacheDir, barriers, lockPath } = await tempCase('cb-refresh-corrupt-live-') + const owner = spawnFixture('cache-refresh-corrupt-owner.ts', [cacheDir, barriers, '4000', '100']) + try { + await waitForFile(join(barriers, 'owner.acquired')) + const ownerToken = await readFile(join(barriers, 'owner.acquired'), 'utf-8') + + // The state a heartbeat leaves when writeFile() truncates and then fails + // (ENOSPC/EIO, swallowed by the heartbeat's own catch). No guard is held, + // and the body carries no token to compare against. + await writeFile(lockPath, '') + // This wait is shorter than one heartbeat period, so the body is still + // truncated throughout: the contender has nothing but the fresh mtime to + // go on, and that alone must keep it out. + const early = await acquireCacheRefreshLock({ cacheDir, waitMs: 80, pollMs: 5, staleMs: 90_000 }) + if (early.outcome === 'acquired') await early.handle.release() + expect(early.outcome).toBe('timed-out') + + // Past the age gate, the successor SHOULD win. Only the heartbeat + // advances mtime and it refuses to rewrite a body it cannot prove is + // ours, so a corrupt body freezes its own mtime and ages out. That is the + // intended end state, not a displacement to be prevented: an owner that + // cannot prove ownership must not publish, and the alternative — letting + // the heartbeat restamp its token over an unparseable body — resurrected + // legitimately-replaced writers and let release() delete a live + // successor's lock. + await writeFile(lockPath, '') + const late = await acquireCacheRefreshLock({ cacheDir, waitMs: 2_400, pollMs: 10, staleMs: 400 }) + expect(late.outcome).toBe('acquired') + if (late.outcome === 'acquired') { + // The successor owns it outright: the body carries its token, not the + // original owner's. + expect(JSON.parse(await readFile(lockPath, 'utf-8')).token).not.toBe(ownerToken) + await late.handle.release() + } + } finally { + await exited(owner) + } + // The displaced owner's fence must refuse. Discarding its parse is the + // fail-safe direction; two writers believing they own the lock is not. + expect(existsSync(join(barriers, 'owner.verify.false')), 'displaced owner still passed its fence').toBe(true) + expect(existsSync(join(barriers, 'owner.verify.true'))).toBe(false) + }, 30_000) +}) diff --git a/tests/cache-refresh-lock-process.test.ts b/tests/cache-refresh-lock-process.test.ts index cf907c31..d837f81e 100644 --- a/tests/cache-refresh-lock-process.test.ts +++ b/tests/cache-refresh-lock-process.test.ts @@ -85,6 +85,41 @@ describe('warm refresh child-process regression', () => { await expect(stat(join(cacheDir, 'session-refresh.lock.takeover'))).rejects.toMatchObject({ code: 'ENOENT' }) }) + it('gives exactly one contender ownership of a stale zero-byte lock', async () => { + const root = await mkdtemp(join(tmpdir(), 'cb-refresh-corrupt-')) + roots.push(root) + const cacheDir = join(root, 'cache') + const barriers = join(root, 'barriers') + await mkdir(cacheDir, { recursive: true }) + await mkdir(barriers, { recursive: true }) + process.env['CODEBURN_CACHE_DIR'] = cacheDir + const initial = emptyCache() + initial.complete = true + await saveCache(initial) + const corruptPath = join(cacheDir, 'session-refresh.lock') + await writeFile(corruptPath, '') + await utimes(corruptPath, new Date(1), new Date(1)) + const source = join(root, 'changed.json') + await writeFile(source, JSON.stringify({ output: 404 })) + + const a = worker(cacheDir, barriers, 'a', source) + const b = worker(cacheDir, barriers, 'b', source) + const winner = await Promise.race([ + waitFor(join(barriers, 'a.parsed')).then(() => 'a'), + waitFor(join(barriers, 'b.parsed')).then(() => 'b'), + ]) + const loser = winner === 'a' ? 'b' : 'a' + expect(Number(existsSync(join(barriers, 'a.parsed'))) + Number(existsSync(join(barriers, 'b.parsed')))).toBe(1) + const loserOutcome = await waitForAny(barriers, [ + `${loser}.timed-out`, `${loser}.parsed`, `${loser}.completed-by-other`, `${loser}.unavailable`, + ]) + expect(loserOutcome, (await readdir(barriers)).join(',')).toBe(`${loser}.timed-out`) + await writeFile(join(barriers, `${winner}.save`), '') + await Promise.all([waitForExit(a), waitForExit(b)]) + await expect(stat(join(cacheDir, 'session-refresh.lock.takeover'))).rejects.toMatchObject({ code: 'ENOENT' }) + expect(Object.keys((await loadCache()).providers['regression']?.files ?? {})).toEqual([source]) + }) + it('serializes disjoint parsed updates so the later publication cannot drop the first', async () => { const root = await mkdtemp(join(tmpdir(), 'cb-refresh-process-')) roots.push(root) diff --git a/tests/cache-refresh-lock.test.ts b/tests/cache-refresh-lock.test.ts index bb8e44b5..4ba633d3 100644 --- a/tests/cache-refresh-lock.test.ts +++ b/tests/cache-refresh-lock.test.ts @@ -1,5 +1,5 @@ import { afterEach, describe, expect, it } from 'vitest' -import { mkdir, mkdtemp, readFile, rm, stat, unlink, utimes, writeFile } from 'fs/promises' +import { chmod, mkdir, mkdtemp, readFile, rm, stat, unlink, utimes, writeFile } from 'fs/promises' import { tmpdir } from 'os' import { join } from 'path' @@ -7,6 +7,7 @@ import { acquireCacheRefreshLock, type RefreshLockClock, } from '../src/cache-refresh-lock.js' +import { clearSessionCache, parseAllSessions } from '../src/parser.js' import { emptyCache, loadCache, saveCache, sessionCachePath } from '../src/session-cache.js' const dirs: string[] = [] @@ -223,3 +224,225 @@ describe('warm session-cache refresh lock', () => { await rm(dir, { recursive: true, force: true }) }) }) + +// A lock body that never parses into a record is a corrupt leftover, not an +// unusable filesystem: classifying it as 'unavailable' routed every subsequent +// refresh to the read-only path and froze ingestion permanently. +describe('warm session-cache refresh lock: corrupt lock recovery', () => { + it('takes over a stale zero-byte lock', async () => { + const dir = await tempDir() + const clock = fakeClock(100_000) + const path = lockPath(dir) + await writeFile(path, '') + const old = new Date(1) + await utimes(path, old, old) + + const result = await acquireCacheRefreshLock({ cacheDir: dir, clock, staleMs: 90, waitMs: 100, pollMs: 1 }) + expect(result.outcome).toBe('acquired') + if (result.outcome !== 'acquired') return + expect(JSON.parse(await readFile(path, 'utf-8')).token).toBe(result.handle.token) + await result.handle.release() + await expect(stat(join(dir, 'session-refresh.lock.takeover'))).rejects.toMatchObject({ code: 'ENOENT' }) + }) + + it('takes over a stale malformed-JSON lock', async () => { + const dir = await tempDir() + const clock = fakeClock(100_000) + const path = lockPath(dir) + await writeFile(path, '{"pid":1,"token":') + const old = new Date(1) + await utimes(path, old, old) + + const result = await acquireCacheRefreshLock({ cacheDir: dir, clock, staleMs: 90, waitMs: 100, pollMs: 1 }) + expect(result.outcome).toBe('acquired') + if (result.outcome !== 'acquired') return + expect(JSON.parse(await readFile(path, 'utf-8')).token).toBe(result.handle.token) + await result.handle.release() + }) + + it('takes over a stale lock whose body parses but has the wrong shape', async () => { + const dir = await tempDir() + const clock = fakeClock(100_000) + const path = lockPath(dir) + await writeFile(path, JSON.stringify({ pid: 'one', token: 7 })) + const old = new Date(1) + await utimes(path, old, old) + + const result = await acquireCacheRefreshLock({ cacheDir: dir, clock, staleMs: 90, waitMs: 100, pollMs: 1 }) + expect(result.outcome).toBe('acquired') + if (result.outcome === 'acquired') await result.handle.release() + }) + + // Staleness is never waived for a corrupt body, so a FRESH one is waited out + // and left alone: it may belong to a live owner whose heartbeat is about to + // repair it. The resulting freeze is bounded by staleMs rather than + // permanent, which is the whole of the reported defect. + it('waits out a fresh malformed lock, then recovers it once it ages past staleMs', async () => { + const dir = await tempDir() + const clock = fakeClock(100_000) + const path = lockPath(dir) + await writeFile(path, '') + const now = new Date(clock.wallNow()) + await utimes(path, now, now) + + let polls = 0 + const fresh = await acquireCacheRefreshLock({ + cacheDir: dir, + clock, + staleMs: 90_000, + waitMs: 50, + pollMs: 10, + sleep: async ms => { polls++; clock.advance(ms) }, + }) + expect(polls).toBeGreaterThan(0) + expect(fresh).toEqual({ outcome: 'timed-out' }) + expect(await readFile(path, 'utf-8')).toBe('') + + // Nothing rewrote the body, so its mtime is still frozen. One stale window + // later the very next run recovers it through the unmodified age gate. + clock.advance(90_001) + const later = await acquireCacheRefreshLock({ cacheDir: dir, clock, staleMs: 90_000, waitMs: 50, pollMs: 10 }) + expect(later.outcome).toBe('acquired') + if (later.outcome !== 'acquired') return + expect(JSON.parse(await readFile(path, 'utf-8')).token).toBe(later.handle.token) + await later.handle.release() + }) + + it('never steals a fresh malformed lock from an owner that repairs it mid-wait', async () => { + const dir = await tempDir() + const clock = fakeClock(100_000) + const path = lockPath(dir) + await writeFile(path, '') + const now = new Date(clock.wallNow()) + await utimes(path, now, now) + + let polls = 0 + const result = await acquireCacheRefreshLock({ + cacheDir: dir, + clock, + staleMs: 90_000, + waitMs: 50, + pollMs: 10, + sleep: async ms => { + if (++polls === 1) { + await writeFile(path, JSON.stringify({ pid: 1, token: 'holder', at: clock.wallNow() })) + const t = new Date(clock.wallNow()) + await utimes(path, t, t) + } + clock.advance(ms) + }, + }) + expect(result).toEqual({ outcome: 'timed-out' }) + expect(JSON.parse(await readFile(path, 'utf-8')).token).toBe('holder') + }) + + it('treats a body truncated mid-heartbeat as contention, not corruption', async () => { + const dir = await tempDir() + const path = lockPath(dir) + const record = (): string => JSON.stringify({ pid: 1, token: 'holder', at: Date.now() }) + await writeFile(path, record()) + // A real heartbeat rewrite exposes a zero-length body for an instant. The + // waiter must keep polling and leave the live owner alone. + const heartbeat = setInterval(() => { + void (async () => { + try { + await writeFile(path, '') + await writeFile(path, record()) + } catch { /* the winner may have replaced the file */ } + })() + }, 1) + let result + try { + result = await acquireCacheRefreshLock({ cacheDir: dir, staleMs: 30, waitMs: 120, pollMs: 5 }) + } finally { + clearInterval(heartbeat) + } + await new Promise(resolve => { setTimeout(resolve, 20) }) + expect(result).toEqual({ outcome: 'timed-out' }) + expect(JSON.parse(await readFile(path, 'utf-8')).token).toBe('holder') + }) + + // The sidecar is created by the same createExclusive as the lock, so it can be + // left 0-byte by exactly the same crash. Before observe() separated corrupt + // from unreadable this returned 'unavailable' from acquireTakeoverGuard and + // froze recovery even when the primary lock was perfectly fine. + it('reclaims a stale zero-byte takeover sidecar', async () => { + const dir = await tempDir() + const path = lockPath(dir) + const sidecar = join(dir, 'session-refresh.lock.takeover') + await writeFile(path, '') + await writeFile(sidecar, '') + const old = new Date(1) + await utimes(path, old, old) + await utimes(sidecar, old, old) + + const result = await acquireCacheRefreshLock({ cacheDir: dir, staleMs: 90, waitMs: 200, pollMs: 5 }) + expect(result.outcome).toBe('acquired') + if (result.outcome !== 'acquired') return + expect(JSON.parse(await readFile(path, 'utf-8')).token).toBe(result.handle.token) + await result.handle.release() + await expect(stat(sidecar)).rejects.toMatchObject({ code: 'ENOENT' }) + }) + + it('still reports unavailable when the lock body cannot be read', async () => { + if (process.getuid?.() === 0) return + const dir = await tempDir() + const path = lockPath(dir) + await writeFile(path, '') + await chmod(path, 0o000) + const old = new Date(1) + await utimes(path, old, old) + try { + const result = await acquireCacheRefreshLock({ cacheDir: dir, staleMs: 1, waitMs: 50, pollMs: 5 }) + expect(result).toEqual({ outcome: 'unavailable' }) + } finally { + await chmod(path, 0o600) + } + }) + + it('resumes ingesting changed sources after a corrupt lock froze the refresh', async () => { + const root = await tempDir() + const cacheDir = join(root, 'cache') + const config = join(root, 'claude') + const projectDir = join(config, 'projects', 'frozen-proj') + await mkdir(cacheDir, { recursive: true }) + await mkdir(projectDir, { recursive: true }) + process.env['CODEBURN_CACHE_DIR'] = cacheDir + process.env['CLAUDE_CONFIG_DIR'] = config + process.env['CODEBURN_DESKTOP_SESSIONS_DIR'] = join(root, 'desktop-sessions') + + const session = (id: string, ts: string): string => [ + JSON.stringify({ type: 'user', sessionId: id, timestamp: ts, cwd: '/tmp/frozen-proj', message: { role: 'user', content: 'hi' } }), + JSON.stringify({ + type: 'assistant', sessionId: id, timestamp: ts, cwd: '/tmp/frozen-proj', + message: { id: `msg-${id}`, type: 'message', role: 'assistant', model: 'claude-sonnet-4-5', content: [], usage: { input_tokens: 100, output_tokens: 20 } }, + }), + ].join('\n') + '\n' + + await writeFile(join(projectDir, 'sess-1.jsonl'), session('sess-1', '2026-05-01T10:00:00.000Z')) + clearSessionCache() + const warm = await parseAllSessions() + expect(warm[0]?.sessions.map(s => s.sessionId)).toEqual(['sess-1']) + + // The field state: a 0-byte lock AND the takeover sidecar of the dead owner + // that was mid-recovery when it died, both older than the stale window. + // Nothing else on the machine ever repairs either file, so on every later + // run the lock read as 'unavailable', the parser fell back to a read-only + // re-parse, and sess-2 was never ingested. + const old = new Date(1) + for (const name of ['session-refresh.lock', 'session-refresh.lock.takeover']) { + await writeFile(join(cacheDir, name), name.endsWith('.takeover') ? JSON.stringify({ pid: 999_999, token: 'dead-owner', at: 1 }) : '') + await utimes(join(cacheDir, name), old, old) + } + + await writeFile(join(projectDir, 'sess-2.jsonl'), session('sess-2', '2026-05-01T11:00:00.000Z')) + clearSessionCache() + const after = await parseAllSessions() + expect(after[0]?.sessions.map(s => s.sessionId).sort()).toEqual(['sess-1', 'sess-2']) + expect(Object.keys((await loadCache()).providers['claude']?.files ?? {}).length).toBe(2) + + clearSessionCache() + delete process.env['CLAUDE_CONFIG_DIR'] + delete process.env['CODEBURN_DESKTOP_SESSIONS_DIR'] + }) +}) diff --git a/tests/fixtures/cache-refresh-corrupt-owner.ts b/tests/fixtures/cache-refresh-corrupt-owner.ts new file mode 100644 index 00000000..dfacd261 --- /dev/null +++ b/tests/fixtures/cache-refresh-corrupt-owner.ts @@ -0,0 +1,22 @@ +import { writeFile } from 'fs/promises' +import { join } from 'path' + +import { acquireCacheRefreshLock } from '../../src/cache-refresh-lock.js' + +// A plain owner in its own process. It records its outcome, its token, and the +// result of the publication fence into the barrier directory so the parent can +// assert what the owner itself believed while a contender was racing it. +const [cacheDir, barrierDir, holdMs, heartbeatMs = '200'] = process.argv.slice(2) +if (!cacheDir || !barrierDir || !holdMs) throw new Error('missing owner argument') + +const refresh = await acquireCacheRefreshLock({ cacheDir, heartbeatMs: Number(heartbeatMs) }) +if (refresh.outcome !== 'acquired') { + await writeFile(join(barrierDir, `owner.${refresh.outcome}`), '') + process.exit(0) +} +await writeFile(join(barrierDir, 'owner.acquired'), refresh.handle.token) +await new Promise(resolve => { setTimeout(resolve, Number(holdMs)) }) +// The publication fence, exactly as parser.ts uses it before saving. +await writeFile(join(barrierDir, `owner.verify.${await refresh.handle.verifyStillOwner()}`), '') +await refresh.handle.release() +await writeFile(join(barrierDir, 'owner.done'), '') diff --git a/tests/fixtures/cache-refresh-slow-owner.ts b/tests/fixtures/cache-refresh-slow-owner.ts new file mode 100644 index 00000000..70b57555 --- /dev/null +++ b/tests/fixtures/cache-refresh-slow-owner.ts @@ -0,0 +1,25 @@ +import { pbkdf2 } from 'crypto' +import { writeFile } from 'fs/promises' +import { join } from 'path' + +import { acquireCacheRefreshLock } from '../../src/cache-refresh-lock.js' + +// Saturate the (size-1) libuv threadpool so every fs operation inside +// createExclusive queues behind a pbkdf2 round. No test hook and no patched +// module: this is what an ordinary process looks like mid cold parse, and it +// widens the window in which the lock exists at zero bytes -- between +// open(path,'wx') and the awaited body write -- to something observable. +const [cacheDir, barrierDir] = process.argv.slice(2) +if (!cacheDir || !barrierDir) throw new Error('missing owner argument') + +let stop = false +const churn = (): void => { if (stop) return; pbkdf2('p', 's', 400_000, 32, 'sha512', () => churn()) } +churn() + +const refresh = await acquireCacheRefreshLock({ cacheDir, heartbeatMs: 10_000 }) +stop = true +await writeFile(join(barrierDir, `owner.${refresh.outcome}`), refresh.outcome === 'acquired' ? refresh.handle.token : '') +if (refresh.outcome === 'acquired') { + await writeFile(join(barrierDir, `owner.verify.${await refresh.handle.verifyStillOwner()}`), '') + await refresh.handle.release() +}