diff --git a/src/cache-refresh-lock.ts b/src/cache-refresh-lock.ts new file mode 100644 index 00000000..a0120098 --- /dev/null +++ b/src/cache-refresh-lock.ts @@ -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 +} + +export type RefreshLockHandle = { + token: string + release: () => Promise + verifyStillOwner: () => Promise +} + +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 { + 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, sleep: (ms: number) => Promise): Promise { + 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 { + // 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 + 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 = Promise.resolve() + +async function enterSingleFlight(): Promise<() => void> { + const previous = singleFlightTail + let leave!: () => void + singleFlightTail = new Promise(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 { + 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 = Promise.resolve() + const serializeOwnerOp = (fn: () => Promise): Promise => { + 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 => { + // 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 => 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 => { + 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 => { + 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' } + } +} diff --git a/src/parser.ts b/src/parser.ts index bd6aaf68..5abd0c51 100644 --- a/src/parser.ts +++ b/src/parser.ts @@ -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, + readOnly = false, ): Promise { const section = getOrCreateProviderSection(diskCache, 'claude') const allDiscoveredFiles = new Set() @@ -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, diskCache: SessionCache, dateRange?: DateRange, + readOnly = false, ): Promise { const provider = await getProvider(providerName) if (!provider) return [] const section = getOrCreateProviderSection(diskCache, providerName) const allDiscoveredFiles = new Set() + const servedSources = [...sources] type SourceInfo = { source: SessionSource; fp: NonNullable>> } 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() - 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 { +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 { + const { isCold = false, readOnly = false, refreshLock } = options const seenMsgIds = new Set() const seenKeys = new Set() 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 => { + 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 diff --git a/src/session-cache.ts b/src/session-cache.ts index c876b31c..99bb31df 100644 --- a/src/session-cache.ts +++ b/src/session-cache.ts @@ -383,7 +383,7 @@ async function adoptLegacyCache(): Promise { } } -export async function saveCache(cache: SessionCache): Promise { +export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Promise): Promise { const dir = getCacheDir() if (!existsSync(dir)) await mkdir(dir, { recursive: true }) @@ -401,13 +401,48 @@ export async function saveCache(cache: SessionCache): Promise { } 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): Promise { + 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 diff --git a/tests/cache-refresh-lock-process.test.ts b/tests/cache-refresh-lock-process.test.ts new file mode 100644 index 00000000..cf907c31 --- /dev/null +++ b/tests/cache-refresh-lock-process.test.ts @@ -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 { + 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 { + 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 { + 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]) + }) +}) diff --git a/tests/cache-refresh-lock.test.ts b/tests/cache-refresh-lock.test.ts new file mode 100644 index 00000000..aac8b50a --- /dev/null +++ b/tests/cache-refresh-lock.test.ts @@ -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 { + 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 }) + }) +}) diff --git a/tests/fixtures/cache-refresh-worker.ts b/tests/fixtures/cache-refresh-worker.ts new file mode 100644 index 00000000..a2807df1 --- /dev/null +++ b/tests/fixtures/cache-refresh-worker.ts @@ -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 { + 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() +} diff --git a/tests/parser-cache-refresh-timeout.test.ts b/tests/parser-cache-refresh-timeout.test.ts new file mode 100644 index 00000000..92415775 --- /dev/null +++ b/tests/parser-cache-refresh-timeout.test.ts @@ -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>): 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 { + 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) + }) +})