cache: strict cross-process gate for the warm session-cache refresh transaction (#743)

* cache: strict cross-process gate for the warm session-cache refresh transaction (#645)

* lock: serialize the fence against the owner's own heartbeat

verifyStillOwner and the heartbeat tick both take the takeover guard;
without in-process serialization the fence could observe its own
heartbeat's guard file, read it as displacement, and abort a legitimate
publication. Fail-safe, but it discarded the parse the lock exists to
protect (~6% of verifies at a 1ms heartbeat in the repro). Owner-side
guard operations now run through one serializer; cross-process guard
semantics are unchanged. Regression test at heartbeatMs 1, mutation-
verified against the unserialized code.

---------

Co-authored-by: Aditya Vikram Singh <247195684+avs-io@users.noreply.github.com>
Co-authored-by: reviewer <review@local>
This commit is contained in:
Aditya Vikram Singh 2026-07-20 21:36:35 +05:30 committed by GitHub
parent 17a23ead40
commit e27fe36bd5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 982 additions and 35 deletions

344
src/cache-refresh-lock.ts Normal file
View file

@ -0,0 +1,344 @@
import { randomBytes } from 'crypto'
import { existsSync } from 'fs'
import { mkdir, open, readFile, stat, unlink, utimes, writeFile } from 'fs/promises'
import { homedir } from 'os'
import { join } from 'path'
const LOCK_FILE = 'session-refresh.lock'
const TAKEOVER_FILE = `${LOCK_FILE}.takeover`
const DEFAULT_HEARTBEAT_MS = 10_000
const DEFAULT_STALE_MS = 90_000
const DEFAULT_WAIT_MS = 30_000
const DEFAULT_POLL_MS = 100
const WINDOWS_RETRIES = 3
type LockRecord = { pid: number; token: string; at: number }
export type RefreshLockClock = {
monotonicNow: () => number
wallNow: () => number
}
export type RefreshLockOptions = {
cacheDir?: string
clock?: RefreshLockClock
heartbeatMs?: number
staleMs?: number
waitMs?: number
pollMs?: number
sleep?: (ms: number) => Promise<void>
}
export type RefreshLockHandle = {
token: string
release: () => Promise<void>
verifyStillOwner: () => Promise<boolean>
}
export type RefreshLockOutcome =
| { outcome: 'acquired'; handle: RefreshLockHandle }
| { outcome: 'completed-by-other' }
| { outcome: 'timed-out' }
| { outcome: 'unavailable' }
const defaultClock: RefreshLockClock = {
monotonicNow: () => Number(process.hrtime.bigint()) / 1_000_000,
wallNow: () => Date.now(),
}
function defaultCacheDir(): string {
return process.env['CODEBURN_CACHE_DIR'] ?? join(homedir(), '.cache', 'codeburn')
}
function delay(ms: number): Promise<void> {
return new Promise(resolve => { setTimeout(resolve, ms) })
}
function isBusyError(err: unknown): boolean {
const code = (err as NodeJS.ErrnoException | undefined)?.code
return code === 'EPERM' || code === 'EBUSY'
}
function isExistsError(err: unknown): boolean {
return (err as NodeJS.ErrnoException | undefined)?.code === 'EEXIST'
}
function isMissingError(err: unknown): boolean {
return (err as NodeJS.ErrnoException | undefined)?.code === 'ENOENT'
}
async function retryWindowsMutation(operation: () => Promise<void>, sleep: (ms: number) => Promise<void>): Promise<boolean> {
for (let attempt = 0; attempt < WINDOWS_RETRIES; attempt++) {
try {
await operation()
return true
} catch (err) {
if (isMissingError(err)) return true
if (!isBusyError(err) || attempt === WINDOWS_RETRIES - 1) return false
await sleep(10 * (attempt + 1))
}
}
return false
}
async function createExclusive(path: string, body: string): Promise<'created' | 'exists' | 'unavailable'> {
try {
const handle = await open(path, 'wx', 0o600)
try { await handle.writeFile(body, { encoding: 'utf-8' }) }
finally { await handle.close() }
return 'created'
} catch (err) {
return isExistsError(err) ? 'exists' : 'unavailable'
}
}
type Observation = { record: LockRecord; mtimeMs: number }
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
for (let attempt = 0; attempt < 3; attempt++) {
try {
const before = await stat(path)
const raw = await readFile(path, 'utf-8')
const after = await stat(path)
if (before.mtimeMs !== after.mtimeMs || before.size !== after.size) {
sawChange = true
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 }
}
} 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'
}
function sameObservation(a: Observation, b: Observation): boolean {
return a.record.token === b.record.token && a.mtimeMs === b.mtimeMs
}
let singleFlightTail: Promise<void> = Promise.resolve()
async function enterSingleFlight(): Promise<() => void> {
const previous = singleFlightTail
let leave!: () => void
singleFlightTail = new Promise<void>(resolve => { leave = resolve })
await previous
return leave
}
/**
* Strict gate for the warm session-cache read/reconcile/parse/save transaction.
* Lock ordering, when the daily-cache follow-up lands, is daily session.
*/
export async function acquireCacheRefreshLock(options: RefreshLockOptions = {}): Promise<RefreshLockOutcome> {
const leaveSingleFlight = await enterSingleFlight()
let ownsSingleFlight = true
const leave = (): void => {
if (!ownsSingleFlight) return
ownsSingleFlight = false
leaveSingleFlight()
}
const cacheDir = options.cacheDir ?? defaultCacheDir()
const clock = options.clock ?? defaultClock
const heartbeatMs = options.heartbeatMs ?? DEFAULT_HEARTBEAT_MS
const staleMs = options.staleMs ?? DEFAULT_STALE_MS
const waitMs = options.waitMs ?? DEFAULT_WAIT_MS
const pollMs = options.pollMs ?? DEFAULT_POLL_MS
const sleep = options.sleep ?? delay
const lockPath = join(cacheDir, LOCK_FILE)
const takeoverPath = join(cacheDir, TAKEOVER_FILE)
const token = randomBytes(16).toString('hex')
const body = (): string => JSON.stringify({ pid: process.pid, token, at: clock.wallNow() })
// In-process serializer for every operation that takes the takeover guard on
// behalf of THIS owner (heartbeat tick, publication fence). Without it the
// fence can observe its own heartbeat's guard file and read "guard held" as
// "displaced", aborting a legitimate publication — fail-safe but it throws
// away the parse the lock exists to protect. Cross-process semantics are
// untouched: the guard file still arbitrates between processes.
let ownerOpTail: Promise<unknown> = Promise.resolve()
const serializeOwnerOp = <T>(fn: () => Promise<T>): Promise<T> => {
const next = ownerOpTail.then(fn)
ownerOpTail = next.catch(() => undefined)
return next
}
const acquireTakeoverGuard = async (): Promise<'created' | 'exists' | 'unavailable'> => {
const created = await createExclusive(takeoverPath, body())
if (created !== 'exists') return created
const staleGuard = await observe(takeoverPath)
if (staleGuard === 'missing') return createExclusive(takeoverPath, body())
if (staleGuard === 'changing') return 'exists'
if (staleGuard === 'unavailable') return 'unavailable'
if (Math.max(0, clock.wallNow() - staleGuard.mtimeMs) <= staleMs) return 'exists'
const reverified = await observe(takeoverPath)
if (reverified === 'missing') return createExclusive(takeoverPath, body())
if (reverified === 'changing') return 'exists'
if (reverified === 'unavailable') return 'unavailable'
if (!sameObservation(staleGuard, reverified)) return 'exists'
if (!await retryWindowsMutation(() => unlink(takeoverPath), sleep)) return 'unavailable'
return createExclusive(takeoverPath, body())
}
const removeIfOwned = async (): Promise<boolean> => {
// A contender holds the takeover guard only for milliseconds at a time;
// retry briefly rather than abandoning our lock to 90s stale-timeout,
// which would stall every waiting process for that long.
let guard: 'created' | 'exists' | 'unavailable' = 'exists'
for (let attempt = 0; attempt < 20 && guard !== 'created'; attempt++) {
guard = await acquireTakeoverGuard()
if (guard === 'unavailable') return false
if (guard !== 'created') await sleep(pollMs)
}
if (guard !== 'created') return false
try {
const current = await observe(lockPath)
if (current === 'missing') return true
if (current === 'changing') return false
if (current === 'unavailable') return false
if (current.record.token !== token) return true
return retryWindowsMutation(() => unlink(lockPath), sleep)
} finally {
await retryWindowsMutation(() => unlink(takeoverPath), sleep)
}
}
const verifyStillOwner = (): Promise<boolean> => serializeOwnerOp(async () => {
const guard = await acquireTakeoverGuard()
if (guard !== 'created') return false
try {
const current = await observe(lockPath)
return current !== 'missing' && current !== 'changing' && current !== 'unavailable' && current.record.token === token
} finally {
await retryWindowsMutation(() => unlink(takeoverPath), sleep)
}
})
const makeHandle = (): RefreshLockHandle => {
let released = false
let heartbeatRunning = false
const heartbeat = setInterval(() => {
void serializeOwnerOp(async () => {
if (released || heartbeatRunning) return
heartbeatRunning = true
const guard = await acquireTakeoverGuard()
if (guard !== 'created') { heartbeatRunning = false; return }
try {
const current = await observe(lockPath)
if (current === 'missing' || current === 'changing' || current === 'unavailable' || current.record.token !== token) return
await writeFile(lockPath, body(), { encoding: 'utf-8' })
const now = new Date(clock.wallNow())
await utimes(lockPath, now, now)
} catch { /* verify/release will turn displacement or I/O failure into a closed gate */ }
finally {
await retryWindowsMutation(() => unlink(takeoverPath), sleep)
heartbeatRunning = false
}
})
}, heartbeatMs)
heartbeat.unref()
return {
token,
verifyStillOwner,
release: async () => {
if (released) return
released = true
clearInterval(heartbeat)
while (heartbeatRunning) await sleep(1)
await removeIfOwned()
leave()
},
}
}
const tryCreateOwner = async (): Promise<RefreshLockOutcome | null> => {
const result = await createExclusive(lockPath, body())
if (result === 'created') return { outcome: 'acquired', handle: makeHandle() }
if (result === 'unavailable') return { outcome: 'unavailable' }
return null
}
const tryTakeover = async (stale: Observation): Promise<RefreshLockOutcome | null> => {
const guard = await acquireTakeoverGuard()
if (guard === 'unavailable') return { outcome: 'unavailable' }
if (guard === 'exists') return null
try {
const current = await observe(lockPath)
if (current === 'unavailable') return { outcome: 'unavailable' }
if (current === 'changing') return null
if (current === 'missing' || !sameObservation(stale, current)) return null
if (Math.max(0, clock.wallNow() - current.mtimeMs) <= staleMs) return null
if (!await retryWindowsMutation(() => unlink(lockPath), sleep)) return { outcome: 'unavailable' }
// Publish the successor while the takeover guard is still canonical.
// Otherwise a waiter can observe neither file and misclassify the narrow
// unlink/create gap as a clean completion by the stale owner.
const successor = await createExclusive(lockPath, body())
if (successor === 'created') return { outcome: 'acquired', handle: makeHandle() }
if (successor === 'unavailable') return { outcome: 'unavailable' }
return null
} finally {
// Never override the try-block's outcome from here: returning
// 'unavailable' after 'acquired' would abandon a live heartbeating
// handle that then blocks every other process until this one exits.
// A guard file we fail to remove reads as contention to others and is
// replaced once stale.
await retryWindowsMutation(() => unlink(takeoverPath), sleep)
}
}
try {
if (!existsSync(cacheDir)) await mkdir(cacheDir, { recursive: true })
const immediate = await tryCreateOwner()
if (immediate) {
if (immediate.outcome !== 'acquired') leave()
return immediate
}
const deadline = clock.monotonicNow() + waitMs
while (clock.monotonicNow() < deadline) {
const observation = await observe(lockPath)
if (observation === 'unavailable') { leave(); return { outcome: 'unavailable' } }
if (observation === 'changing') { await sleep(pollMs); continue }
if (observation === 'missing') {
// A stale taker removes the primary while holding the guard, then
// exclusively creates its successor. Do not misreport that narrow gap
// as a clean completion by the previous owner.
const guard = await observe(takeoverPath)
if (guard === 'unavailable') { leave(); return { outcome: 'unavailable' } }
if (guard === 'changing') { await sleep(pollMs); continue }
if (guard === 'missing') { leave(); return { outcome: 'completed-by-other' } }
await sleep(pollMs)
continue
}
const age = Math.max(0, clock.wallNow() - observation.mtimeMs)
if (age > staleMs) {
const takeover = await tryTakeover(observation)
if (takeover) {
if (takeover.outcome !== 'acquired') leave()
return takeover
}
}
await sleep(pollMs)
}
leave()
return { outcome: 'timed-out' }
} catch {
leave()
return { outcome: 'unavailable' }
}
}

View file

@ -25,6 +25,7 @@ import {
reconcileFile,
saveCache,
} from './session-cache.js'
import { acquireCacheRefreshLock, type RefreshLockHandle } from './cache-refresh-lock.js'
import type { ParsedProviderCall, SessionSource } from './providers/types.js'
import type {
ApiUsageIteration,
@ -1801,6 +1802,7 @@ async function scanProjectDirs(
// caller (parseAllSessions) can persist partial progress. A run killed
// mid-scan then resumes from a warm cache instead of re-parsing from zero.
onFileParsed?: () => Promise<void>,
readOnly = false,
): Promise<ProjectSummary[]> {
const section = getOrCreateProviderSection(diskCache, 'claude')
const allDiscoveredFiles = new Set<string>()
@ -1818,16 +1820,19 @@ async function scanProjectDirs(
const fp = await fingerprintFile(filePath)
if (!fp) continue
const action = reconcileFile(fp, section.files[filePath])
if (action.action === 'unchanged') {
const cached = section.files[filePath]
const action = reconcileFile(fp, cached)
if (cached && (readOnly || action.action === 'unchanged')) {
unchangedFiles.push({ filePath, dirName, source, cached: section.files[filePath]! })
} else if (action.action === 'appended') {
changedFiles.push({
filePath,
info: { dirName, fp, source },
append: { cached: section.files[filePath]!, readFromOffset: action.readFromOffset },
})
} else {
} else if (!readOnly) {
if (action.action === 'appended') {
changedFiles.push({
filePath,
info: { dirName, fp, source },
append: { cached: section.files[filePath]!, readFromOffset: action.readFromOffset },
})
continue
}
changedFiles.push({ filePath, info: { dirName, fp, source } })
}
}
@ -1836,6 +1841,16 @@ async function scanProjectDirs(
}
discoverProgress.finish()
if (readOnly) {
for (const [filePath, cached] of Object.entries(section.files)) {
if (allDiscoveredFiles.has(filePath)) continue
const dirName = cached.canonicalProjectName
?? cached.turns[0]?.calls[0]?.project
?? basename(dirname(filePath))
unchangedFiles.push({ filePath, dirName, cached })
}
}
// Pre-seed dedup set from cached (unchanged) files
for (const { cached } of unchangedFiles) {
for (const turn of cached.turns) {
@ -1975,7 +1990,7 @@ async function scanProjectDirs(
}
parseProgress.finish()
if (dirs.length > 0) {
if (!readOnly && dirs.length > 0) {
for (const cachedPath of Object.keys(section.files)) {
if (!allDiscoveredFiles.has(cachedPath)) {
delete section.files[cachedPath]
@ -2523,12 +2538,14 @@ async function parseProviderSources(
seenKeys: Set<string>,
diskCache: SessionCache,
dateRange?: DateRange,
readOnly = false,
): Promise<ProjectSummary[]> {
const provider = await getProvider(providerName)
if (!provider) return []
const section = getOrCreateProviderSection(diskCache, providerName)
const allDiscoveredFiles = new Set<string>()
const servedSources = [...sources]
type SourceInfo = { source: SessionSource; fp: NonNullable<Awaited<ReturnType<typeof fingerprintFile>>> }
const unchangedSources: Array<{ source: SessionSource; cached: CachedFile }> = []
@ -2541,7 +2558,7 @@ async function parseProviderSources(
// comes from a live API fetch in createSessionParser. There's nothing to
// fingerprint or incrementally cache, so re-fetch every run with a synthetic
// fingerprint (mtime=now so the date-range filter below never excludes it).
if (provider.network) {
if (provider.network && !readOnly) {
changedSources.push({ source, fp: { dev: 0, ino: 0, mtimeMs: Date.now(), sizeBytes: 0 } })
continue
}
@ -2554,13 +2571,26 @@ async function parseProviderSources(
// A cached parse failure at this same fingerprint stays skipped — don't
// re-read a file that already threw and hasn't changed. It re-parses only
// when the file changes (then `reconcileFile` reports non-'unchanged').
if (action.action === 'unchanged' && cached && (cached.failed || !cachedFileNeedsProviderReparse(providerName, source.path, cached))) {
if (cached && (readOnly || (action.action === 'unchanged' && (cached.failed || !cachedFileNeedsProviderReparse(providerName, source.path, cached))))) {
unchangedSources.push({ source, cached })
} else {
} else if (!readOnly) {
changedSources.push({ source, fp })
}
}
if (readOnly) {
for (const [path, cached] of Object.entries(section.files)) {
if (allDiscoveredFiles.has(path)) continue
servedSources.push({
provider: providerName,
path,
project: cached.turns[0]?.calls[0]?.project ?? providerName,
})
allDiscoveredFiles.add(path)
unchangedSources.push({ source: servedSources[servedSources.length - 1]!, cached })
}
}
// Parser dedup: cross-provider keys + cached file keys.
// Separate from seenKeys so parsing doesn't suppress query-time output.
const parserDedup = new Set(seenKeys)
@ -2664,12 +2694,12 @@ async function parseProviderSources(
// Stamp the durable flag into the cache section so the orphan-bootstrap in
// parseAllSessions can fast-check without a getProvider() round-trip.
if (provider.durableSources && !section.durable) {
if (!readOnly && provider.durableSources && !section.durable) {
section.durable = true
;(diskCache as { _dirty?: boolean })._dirty = true
}
if (sources.length > 0 && !provider.durableSources) {
if (!readOnly && sources.length > 0 && !provider.durableSources) {
for (const cachedPath of Object.keys(section.files)) {
if (!allDiscoveredFiles.has(cachedPath)) {
delete section.files[cachedPath]
@ -2680,7 +2710,7 @@ async function parseProviderSources(
// 90-day age-out for durable providers: remove entries whose newest call is
// older than 90 days so the cache doesn't grow unboundedly over time.
if (provider.durableSources) {
if (!readOnly && provider.durableSources) {
const cutoffMs = Date.now() - 90 * 24 * 60 * 60 * 1000
for (const [cachedPath, cachedFile] of Object.entries(section.files)) {
const newestTs = cachedFile.turns
@ -2699,7 +2729,7 @@ async function parseProviderSources(
// Uses seenKeys (shared across providers) for cross-provider dedup.
const sessionMap = new Map<string, { project: string; projectPath?: string; turns: ClassifiedTurn[] }>()
for (const source of sources) {
for (const source of servedSources) {
const cachedFile = section.files[source.path]
if (!cachedFile) continue
@ -2967,21 +2997,59 @@ export async function parseAllSessions(dateRange?: DateRange, providerFilter?: s
// If another live process is already hydrating, wait for it, then reload the
// now-warm cache instead of double-parsing. Never a correctness gate: on any
// doubt it proceeds unlocked.
const hydration = await beginColdHydration(!isCacheComplete(diskCache))
if (hydration.waited) diskCache = await loadCache()
if (!isCacheComplete(diskCache)) {
const hydration = await beginColdHydration(true)
if (hydration.waited) diskCache = await loadCache()
const isCold = !isCacheComplete(diskCache)
try {
return await runParse(key, diskCache, dateRange, providerFilter, { isCold })
} finally {
await hydration.release()
}
}
// A complete cache refresh is a strict read/reconcile/parse/save transaction.
// Keep the snapshot loaded before acquisition: timeout/unavailable paths serve
// exactly this complete snapshot and never mutate or invalidate the holder.
const priorSnapshot = diskCache
const refresh = await acquireCacheRefreshLock()
if (refresh.outcome === 'timed-out' || refresh.outcome === 'unavailable') {
return runParse(key, priorSnapshot, dateRange, providerFilter, { readOnly: true })
}
if (refresh.outcome === 'completed-by-other') {
return runParse(key, await loadCache(), dateRange, providerFilter, { readOnly: true })
}
// Cold = this run finishes a genuine (possibly resumed) full parse. A warm run
// reconciles an already-complete corpus. Drives the splash's cold-only reveal
// and the daily backfill's "don't finalize partial history" guard.
const isCold = !isCacheComplete(diskCache)
try {
return await runParse(key, diskCache, dateRange, providerFilter, isCold)
// Reload only after ownership is canonical; this closes the lost-update
// window between the pre-gate read and the holder's completed publication.
diskCache = await loadCache()
return await runParse(key, diskCache, dateRange, providerFilter, { refreshLock: refresh.handle })
} catch (err) {
if (!(err instanceof RefreshFenceLostError) && !(err instanceof RefreshPublicationUnavailableError)) throw err
return runParse(key, await loadCache(), dateRange, providerFilter, { readOnly: true })
} finally {
await hydration.release()
await refresh.handle.release()
}
}
async function runParse(key: string, diskCache: SessionCache, dateRange?: DateRange, providerFilter?: string, isCold = false): Promise<ProjectSummary[]> {
class RefreshFenceLostError extends Error {}
class RefreshPublicationUnavailableError extends Error {}
type RunParseOptions = {
isCold?: boolean
readOnly?: boolean
refreshLock?: RefreshLockHandle
}
async function runParse(
key: string,
diskCache: SessionCache,
dateRange?: DateRange,
providerFilter?: string,
options: RunParseOptions = {},
): Promise<ProjectSummary[]> {
const { isCold = false, readOnly = false, refreshLock } = options
const seenMsgIds = new Set<string>()
const seenKeys = new Set<string>()
const allSources = await discoverAllSessions(providerFilter)
@ -3002,6 +3070,7 @@ async function runParse(key: string, diskCache: SessionCache, dateRange?: DateRa
// never races the final save below.
let lastSaveAt = Date.now()
const saveProgress = async (): Promise<void> => {
if (!isCold || readOnly) return
if (!(diskCache as { _dirty?: boolean })._dirty) return
if (Date.now() - lastSaveAt < PROGRESS_SAVE_THROTTLE_MS) return
lastSaveAt = Date.now()
@ -3023,7 +3092,7 @@ async function runParse(key: string, diskCache: SessionCache, dateRange?: DateRa
if (claudeSources.length > 0) emitScanProgress({ kind: 'provider', provider: 'claude', state: 'start' })
let claudeProjects: ProjectSummary[] = []
try {
claudeProjects = await scanProjectDirs(claudeDirs, seenMsgIds, diskCache, dateRange, saveProgress)
claudeProjects = await scanProjectDirs(claudeDirs, seenMsgIds, diskCache, dateRange, saveProgress, readOnly)
if (claudeSources.length > 0) emitScanProgress({ kind: 'provider', provider: 'claude', state: 'done', files: claudeSources.length })
} catch (err) {
if (!isPermissionError(err)) throw err
@ -3035,7 +3104,7 @@ async function runParse(key: string, diskCache: SessionCache, dateRange?: DateRa
for (const [providerName, sources] of providerGroups) {
emitScanProgress({ kind: 'provider', provider: providerName, state: 'start' })
try {
const projects = await parseProviderSources(providerName, sources, seenKeys, diskCache, dateRange)
const projects = await parseProviderSources(providerName, sources, seenKeys, diskCache, dateRange, readOnly)
emitScanProgress({ kind: 'provider', provider: providerName, state: 'done', files: sources.length })
otherProjects.push(...projects)
} catch (err) {
@ -3064,7 +3133,7 @@ async function runParse(key: string, diskCache: SessionCache, dateRange?: DateRa
// constant — both checks are O(1) and avoid a getProvider() dynamic-import
// round-trip for every unprocessed provider in the disk cache.
if (!section.durable && !DURABLE_PROVIDER_NAMES.has(providerName)) continue
const projects = await parseProviderSources(providerName, [], seenKeys, diskCache, dateRange)
const projects = await parseProviderSources(providerName, [], seenKeys, diskCache, dateRange, readOnly)
otherProjects.push(...projects)
}
@ -3075,9 +3144,15 @@ async function runParse(key: string, diskCache: SessionCache, dateRange?: DateRa
// on is durable. A run killed before here never reaches this, so its throttled
// partial saves keep `complete: false` and the next launch resumes cold.
const wasComplete = isCacheComplete(diskCache)
if (!wasComplete) diskCache.complete = true
if ((diskCache as { _dirty?: boolean })._dirty || !wasComplete) {
try { await saveCache(diskCache) } catch {}
if (!readOnly && !wasComplete) diskCache.complete = true
if (!readOnly && ((diskCache as { _dirty?: boolean })._dirty || !wasComplete)) {
try {
const published = await saveCache(diskCache, refreshLock?.verifyStillOwner)
if (!published) throw new RefreshFenceLostError()
} catch (err) {
if (err instanceof RefreshFenceLostError) throw err
if (refreshLock) throw new RefreshPublicationUnavailableError()
}
}
sessionHydrationComplete = true

View file

@ -383,7 +383,7 @@ async function adoptLegacyCache(): Promise<SessionCache> {
}
}
export async function saveCache(cache: SessionCache): Promise<void> {
export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Promise<boolean>): Promise<boolean> {
const dir = getCacheDir()
if (!existsSync(dir)) await mkdir(dir, { recursive: true })
@ -401,13 +401,48 @@ export async function saveCache(cache: SessionCache): Promise<void> {
}
try {
await rename(tempPath, finalPath)
// The warm refresh transaction passes an ownership fence. It must be the
// final operation before publication so a displaced writer cannot replace
// the canonical cache with its stale snapshot.
if (verifyStillOwner && !await verifyStillOwner()) {
await retryCacheFileMutation(() => unlink(tempPath))
return false
}
let renamed = false
for (let attempt = 0; attempt < 3; attempt++) {
try {
await rename(tempPath, finalPath)
renamed = true
break
} catch (err) {
const code = (err as NodeJS.ErrnoException).code
if ((code !== 'EPERM' && code !== 'EBUSY') || attempt === 2) throw err
await new Promise(resolve => { setTimeout(resolve, 10 * (attempt + 1)) })
}
}
if (!renamed) throw new Error('session cache rename failed')
return true
} catch (err) {
try { await unlink(tempPath) } catch {}
await retryCacheFileMutation(() => unlink(tempPath))
throw err
}
}
async function retryCacheFileMutation(operation: () => Promise<void>): Promise<boolean> {
for (let attempt = 0; attempt < 3; attempt++) {
try {
await operation()
return true
} catch (err) {
const code = (err as NodeJS.ErrnoException).code
if (code === 'ENOENT') return true
if ((code !== 'EPERM' && code !== 'EBUSY') || attempt === 2) return false
await new Promise(resolve => { setTimeout(resolve, 10 * (attempt + 1)) })
}
}
return false
}
// ── File Fingerprinting ────────────────────────────────────────────────
//
// Fingerprints cover the source's transcript file only. Providers that keep

View file

@ -0,0 +1,167 @@
import { afterEach, describe, expect, it } from 'vitest'
import { spawn, type ChildProcess } from 'child_process'
import { existsSync } from 'fs'
import { mkdir, mkdtemp, readdir, rm, stat, utimes, writeFile } from 'fs/promises'
import { tmpdir } from 'os'
import { join } from 'path'
import { emptyCache, loadCache, saveCache } from '../src/session-cache.js'
const roots: string[] = []
async function waitFor(path: string, timeoutMs = 5_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) })
}
}
async function waitForAny(dir: string, names: string[], timeoutMs = 5_000): Promise<string> {
const deadline = Date.now() + timeoutMs
while (Date.now() < deadline) {
for (const name of names) if (existsSync(join(dir, name))) return name
await new Promise(resolve => { setTimeout(resolve, 5) })
}
throw new Error(`timed out waiting for ${names.join(', ')}; saw ${(await readdir(dir)).join(', ')}`)
}
function waitForExit(child: ChildProcess): Promise<void> {
if (child.exitCode !== null) return child.exitCode === 0 ? Promise.resolve() : Promise.reject(new Error(`worker exited ${child.exitCode}`))
return new Promise((resolve, reject) => {
let stderr = ''
child.stderr?.on('data', chunk => { stderr += String(chunk) })
child.once('error', reject)
child.once('exit', code => code === 0 ? resolve() : reject(new Error(`worker exited ${code}: ${stderr}`)))
})
}
function worker(cacheDir: string, barriers: string, id: string, source: string, bypass = false): ChildProcess {
return spawn(process.execPath, ['--import', 'tsx', join(process.cwd(), 'tests/fixtures/cache-refresh-worker.ts'), cacheDir, barriers, id, source, String(bypass)], {
cwd: process.cwd(),
stdio: ['ignore', 'ignore', 'pipe'],
})
}
afterEach(async () => {
delete process.env['CODEBURN_CACHE_DIR']
await Promise.all(roots.splice(0).map(root => rm(root, { recursive: true, force: true })))
})
describe('warm refresh child-process regression', () => {
it('gives exactly one contender ownership of a stale lock', async () => {
const root = await mkdtemp(join(tmpdir(), 'cb-refresh-stale-'))
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 stalePath = join(cacheDir, 'session-refresh.lock')
await writeFile(stalePath, JSON.stringify({ pid: 1, token: 'abandoned', at: 1 }))
await utimes(stalePath, new Date(1), new Date(1))
const source = join(root, 'changed.json')
await writeFile(source, JSON.stringify({ output: 303 }))
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)
// Keep the winner alive through the loser's full waiter budget: a second
// stale contender must never publish or steal a heartbeating successor.
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' })
})
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)
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 sourceA = join(root, 'changed-a.json')
const sourceB = join(root, 'changed-b.json')
await writeFile(sourceA, JSON.stringify({ output: 101 }))
await writeFile(sourceB, JSON.stringify({ output: 202 }))
const a = worker(cacheDir, barriers, 'a', sourceA)
const b = worker(cacheDir, barriers, 'b', sourceB)
// Exactly one child can cross the acquisition barrier. Let it publish and
// release; only then can the other reload the first child's update.
let retry: ChildProcess | undefined
await Promise.race([
waitFor(join(barriers, 'a.parsed')).then(() => 'a'),
waitFor(join(barriers, 'b.parsed')).then(() => 'b'),
]).then(async first => {
const second = first === 'a' ? 'b' : 'a'
const secondSource = second === 'a' ? sourceA : sourceB
await writeFile(join(barriers, `${first}.save`), '')
await waitFor(join(barriers, `${first}.published`))
// A waiter that observes the clean release correctly serves the holder's
// fresh snapshot instead of mutating. A subsequent refresh of that
// process then acquires normally and applies its independently visible
// source change on top of the holder's publication.
const outcome = await waitForAny(barriers, [`${second}.completed-by-other`, `${second}.parsed`])
if (outcome === `${second}.parsed`) {
await writeFile(join(barriers, `${second}.save`), '')
} else {
await waitForExit(second === 'a' ? a : b)
retry = worker(cacheDir, barriers, `${second}-retry`, secondSource)
await waitFor(join(barriers, `${second}-retry.parsed`))
await writeFile(join(barriers, `${second}-retry.save`), '')
}
})
await Promise.all([waitForExit(a), waitForExit(b), ...(retry ? [waitForExit(retry)] : [])])
const files = (await loadCache()).providers['regression']?.files ?? {}
expect(Object.keys(files).sort()).toEqual([sourceA, sourceB].sort())
})
it('proves the barrier reproducer loses one update when the transaction gate is bypassed', async () => {
const root = await mkdtemp(join(tmpdir(), 'cb-refresh-control-'))
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 sourceA = join(root, 'changed-a.json')
const sourceB = join(root, 'changed-b.json')
await writeFile(sourceA, JSON.stringify({ output: 101 }))
await writeFile(sourceB, JSON.stringify({ output: 202 }))
const a = worker(cacheDir, barriers, 'a', sourceA, true)
const b = worker(cacheDir, barriers, 'b', sourceB, true)
await Promise.all([waitFor(join(barriers, 'a.parsed')), waitFor(join(barriers, 'b.parsed'))])
await writeFile(join(barriers, 'a.save'), '')
await waitFor(join(barriers, 'a.published'))
await writeFile(join(barriers, 'b.save'), '')
await Promise.all([waitForExit(a), waitForExit(b)])
const files = (await loadCache()).providers['regression']?.files ?? {}
expect(Object.keys(files)).toEqual([sourceB])
})
})

View file

@ -0,0 +1,221 @@
import { afterEach, describe, expect, it } from 'vitest'
import { mkdir, mkdtemp, readFile, rm, stat, unlink, utimes, writeFile } from 'fs/promises'
import { tmpdir } from 'os'
import { join } from 'path'
import {
acquireCacheRefreshLock,
type RefreshLockClock,
} from '../src/cache-refresh-lock.js'
import { emptyCache, loadCache, saveCache, sessionCachePath } from '../src/session-cache.js'
const dirs: string[] = []
async function tempDir(): Promise<string> {
const dir = await mkdtemp(join(tmpdir(), 'cb-refresh-lock-'))
dirs.push(dir)
return dir
}
function lockPath(dir: string): string {
return join(dir, 'session-refresh.lock')
}
function fakeClock(start = 1_000): RefreshLockClock & { advance: (ms: number) => void } {
let wall = start
let monotonic = start
return {
wallNow: () => wall,
monotonicNow: () => monotonic,
advance: ms => { wall += ms; monotonic += ms },
}
}
afterEach(async () => {
delete process.env['CODEBURN_CACHE_DIR']
await Promise.all(dirs.splice(0).map(dir => rm(dir, { recursive: true, force: true })))
})
describe('warm session-cache refresh lock', () => {
it('returns acquired and releases its own token', async () => {
const dir = await tempDir()
const result = await acquireCacheRefreshLock({ cacheDir: dir })
expect(result.outcome).toBe('acquired')
if (result.outcome !== 'acquired') return
const record = JSON.parse(await readFile(lockPath(dir), 'utf-8'))
expect(record).toMatchObject({ pid: process.pid, token: result.handle.token })
expect(typeof record.at).toBe('number')
await result.handle.release()
await expect(stat(lockPath(dir))).rejects.toMatchObject({ code: 'ENOENT' })
})
it('reports completed-by-other after a clean release', async () => {
const dir = await tempDir()
const path = lockPath(dir)
await writeFile(path, JSON.stringify({ pid: 1, token: 'holder', at: Date.now() }))
let polls = 0
const result = await acquireCacheRefreshLock({
cacheDir: dir,
waitMs: 100,
pollMs: 1,
sleep: async () => {
if (++polls === 1) await unlink(path)
},
})
expect(result).toEqual({ outcome: 'completed-by-other' })
})
it('uses a monotonic deadline and times out without invalidating the holder', async () => {
const dir = await tempDir()
const clock = fakeClock()
const path = lockPath(dir)
await writeFile(path, JSON.stringify({ pid: 1, token: 'holder', at: clock.wallNow() }))
const now = new Date(clock.wallNow())
await utimes(path, now, now)
const result = await acquireCacheRefreshLock({
cacheDir: dir,
clock,
waitMs: 30,
staleMs: 90,
pollMs: 10,
sleep: async ms => { clock.advance(ms) },
})
expect(result).toEqual({ outcome: 'timed-out' })
expect(JSON.parse(await readFile(path, 'utf-8')).token).toBe('holder')
})
it('reports unavailable when lock infrastructure is unusable', async () => {
const dir = await tempDir()
const notDirectory = join(dir, 'file')
await writeFile(notDirectory, 'x')
expect(await acquireCacheRefreshLock({ cacheDir: notDirectory })).toEqual({ outcome: 'unavailable' })
})
it('serializes same-process acquisitions before touching the filesystem lock', async () => {
const dir = await tempDir()
const first = await acquireCacheRefreshLock({ cacheDir: dir })
expect(first.outcome).toBe('acquired')
if (first.outcome !== 'acquired') return
let settled = false
const secondPromise = acquireCacheRefreshLock({ cacheDir: dir }).then(result => {
settled = true
return result
})
await new Promise(resolve => { setTimeout(resolve, 20) })
expect(settled).toBe(false)
await first.handle.release()
const second = await secondPromise
expect(second.outcome).toBe('acquired')
if (second.outcome === 'acquired') {
expect(second.handle.token).not.toBe(first.handle.token)
await second.handle.release()
}
})
it('does not take over a heartbeating owner', async () => {
const dir = await tempDir()
const path = lockPath(dir)
await writeFile(path, JSON.stringify({ pid: 1, token: 'holder', at: Date.now() }))
const heartbeat = setInterval(() => {
const now = new Date()
void utimes(path, now, now)
}, 5)
try {
const result = await acquireCacheRefreshLock({ cacheDir: dir, staleMs: 20, waitMs: 60, pollMs: 5 })
expect(result).toEqual({ outcome: 'timed-out' })
expect(JSON.parse(await readFile(path, 'utf-8')).token).toBe('holder')
} finally {
clearInterval(heartbeat)
}
})
it('heartbeats its own lock body and mtime with the injected clock', async () => {
const dir = await tempDir()
const clock = fakeClock(10_000)
const result = await acquireCacheRefreshLock({ cacheDir: dir, clock, heartbeatMs: 5 })
expect(result.outcome).toBe('acquired')
if (result.outcome !== 'acquired') return
try {
const before = (await stat(lockPath(dir))).mtimeMs
clock.advance(1_000)
await new Promise(resolve => { setTimeout(resolve, 100) })
const record = JSON.parse(await readFile(lockPath(dir), 'utf-8'))
expect(record.at).toBe(clock.wallNow())
expect((await stat(lockPath(dir))).mtimeMs).not.toBe(before)
} finally {
await result.handle.release()
}
})
it('takes over only after re-verifying a stale token and mtime', async () => {
const dir = await tempDir()
const clock = fakeClock(100_000)
const path = lockPath(dir)
await writeFile(path, JSON.stringify({ pid: 1, token: 'stale', at: 1 }))
const old = new Date(1)
await utimes(path, old, old)
const result = await acquireCacheRefreshLock({ cacheDir: dir, clock, staleMs: 90, waitMs: 100 })
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('reclaims an abandoned stale takeover guard', async () => {
const dir = await tempDir()
const clock = fakeClock(100_000)
await writeFile(lockPath(dir), JSON.stringify({ pid: 1, token: 'stale', at: 1 }))
await writeFile(join(dir, 'session-refresh.lock.takeover'), JSON.stringify({ pid: 2, token: 'stale-guard', at: 1 }))
const old = new Date(1)
await utimes(lockPath(dir), old, old)
await utimes(join(dir, 'session-refresh.lock.takeover'), old, old)
const result = await acquireCacheRefreshLock({ cacheDir: dir, clock, staleMs: 90, waitMs: 100 })
expect(result.outcome).toBe('acquired')
if (result.outcome === 'acquired') await result.handle.release()
})
it('fences publication and release removes only its own token', async () => {
const dir = await tempDir()
process.env['CODEBURN_CACHE_DIR'] = dir
const original = emptyCache()
original.complete = true
await saveCache(original)
const result = await acquireCacheRefreshLock({ cacheDir: dir, heartbeatMs: 60_000 })
expect(result.outcome).toBe('acquired')
if (result.outcome !== 'acquired') return
await writeFile(lockPath(dir), JSON.stringify({ pid: 999, token: 'successor', at: Date.now() }))
const changed = emptyCache()
changed.complete = true
changed.providers['claude'] = { parseVersion: 'test', envFingerprint: 'test', files: {} }
expect(await saveCache(changed, result.handle.verifyStillOwner)).toBe(false)
expect((await loadCache()).providers['claude']).toBeUndefined()
await result.handle.release()
expect(JSON.parse(await readFile(lockPath(dir), 'utf-8')).token).toBe('successor')
expect(sessionCachePath()).toContain(dir)
})
it('the fence never loses to its own heartbeat (in-process serialization)', async () => {
// Regression: verifyStillOwner and the heartbeat tick both take the
// takeover guard; without in-process serialization the fence could observe
// its own heartbeat's guard file and abort a legitimate publication.
// At a 1ms heartbeat this raced ~6% of the time before the fix.
const dir = await mkdtemp(join(tmpdir(), 'refresh-lock-'))
const result = await acquireCacheRefreshLock({ cacheDir: dir, heartbeatMs: 1 })
if (result.outcome !== 'acquired') throw new Error(`expected acquired, got ${result.outcome}`)
for (let i = 0; i < 300; i++) {
expect(await result.handle.verifyStillOwner()).toBe(true)
}
await result.handle.release()
await rm(dir, { recursive: true, force: true })
})
})

43
tests/fixtures/cache-refresh-worker.ts vendored Normal file
View file

@ -0,0 +1,43 @@
import { existsSync } from 'fs'
import { mkdir, readFile, writeFile } from 'fs/promises'
import { join } from 'path'
import { acquireCacheRefreshLock } from '../../src/cache-refresh-lock.js'
import { loadCache, saveCache } from '../../src/session-cache.js'
const [cacheDir, barrierDir, id, sourcePath, bypass = 'false'] = process.argv.slice(2)
if (!cacheDir || !barrierDir || !id || !sourcePath) throw new Error('missing worker argument')
async function waitFor(name: string): Promise<void> {
const path = join(barrierDir!, name)
while (!existsSync(path)) await new Promise(resolve => { setTimeout(resolve, 5) })
}
process.env['CODEBURN_CACHE_DIR'] = cacheDir
await mkdir(barrierDir, { recursive: true })
const refresh = bypass === 'true' ? null : await acquireCacheRefreshLock({ cacheDir, waitMs: 2_000, pollMs: 5 })
if (refresh && refresh.outcome !== 'acquired') {
await writeFile(join(barrierDir, `${id}.${refresh.outcome}`), '')
process.exit(0)
}
try {
const cache = await loadCache()
// Parsing is deliberately tiny; the files and barrier make the transaction
// interleaving deterministic rather than relying on parser runtime variance.
const parsed = JSON.parse(await readFile(sourcePath, 'utf-8')) as { output: number }
cache.providers['regression'] ??= { parseVersion: 'test', envFingerprint: 'test', files: {} }
cache.providers['regression'].files[sourcePath] = {
fingerprint: { dev: 1, ino: parsed.output, mtimeMs: parsed.output, sizeBytes: parsed.output },
mcpInventory: [],
turns: [],
}
;(cache as { _dirty?: boolean })._dirty = true
await writeFile(join(barrierDir, `${id}.parsed`), '')
await waitFor(`${id}.save`)
const published = await saveCache(cache, refresh?.handle.verifyStillOwner)
await writeFile(join(barrierDir, `${id}.${published ? 'published' : 'fenced'}`), '')
} finally {
await refresh?.handle.release()
}

View file

@ -0,0 +1,62 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { mkdir, mkdtemp, readFile, rm, writeFile } from 'fs/promises'
import { tmpdir } from 'os'
import { join } from 'path'
vi.mock('../src/cache-refresh-lock.js', () => ({
acquireCacheRefreshLock: async () => ({ outcome: 'timed-out' as const }),
}))
import { clearSessionCache, parseAllSessions } from '../src/parser.js'
import { sessionCachePath } from '../src/session-cache.js'
let root: string
let sessionPath: string
function output(projects: Awaited<ReturnType<typeof parseAllSessions>>): number {
return projects.flatMap(p => p.sessions).flatMap(s => s.turns)
.flatMap(t => t.assistantCalls).reduce((sum, call) => sum + call.usage.outputTokens, 0)
}
async function writeSession(value: number): Promise<void> {
await writeFile(sessionPath, JSON.stringify({
type: 'assistant',
sessionId: 'sess',
timestamp: '2026-05-15T10:00:00Z',
cwd: '/tmp/proj',
message: {
id: `msg-${value}`, type: 'message', role: 'assistant', model: 'claude-sonnet-4-5',
content: [], usage: { input_tokens: 100, output_tokens: value },
},
}) + '\n')
}
beforeEach(async () => {
clearSessionCache()
root = await mkdtemp(join(tmpdir(), 'cb-refresh-timeout-'))
const home = join(root, 'home')
const project = join(home, 'projects', 'proj')
await mkdir(project, { recursive: true })
sessionPath = join(project, 'sess.jsonl')
process.env['CLAUDE_CONFIG_DIR'] = home
process.env['CODEBURN_CACHE_DIR'] = join(root, 'cache')
process.env['CODEBURN_DESKTOP_SESSIONS_DIR'] = join(home, 'desktop-sessions')
})
afterEach(async () => {
clearSessionCache()
await rm(root, { recursive: true, force: true })
})
describe('parseAllSessions warm refresh timeout', () => {
it('serves the prior complete snapshot and leaves the holder cache untouched', async () => {
await writeSession(50)
expect(output(await parseAllSessions(undefined, 'claude'))).toBe(50)
const before = await readFile(sessionCachePath(), 'utf-8')
await writeSession(5000)
clearSessionCache()
expect(output(await parseAllSessions(undefined, 'claude'))).toBe(50)
expect(await readFile(sessionCachePath(), 'utf-8')).toBe(before)
})
})