cache: recover a corrupt session-refresh lock instead of freezing ingestion

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<void>, 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,14 +99,32 @@ 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<ObservationResult> {
   // Exclusive create exposes the directory entry just before its small body is
   // 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,22 +135,43 @@ async function observe(path: string): Promise<ObservationResult> {
         await delay(1)
         continue
       }
-      const parsed = JSON.parse(raw) as Partial<LockRecord>
-      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<LockRecord> | undefined
+      try { parsed = JSON.parse(raw) as Partial<LockRecord> } 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
       if (code === 'EACCES' || code === 'EPERM') return 'unavailable'
     }
     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<void> = 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)

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<void> {
+  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<unknown> => 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)
+})

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)