From e9dcb363af865ed39179aed015e957cdd2c802d8 Mon Sep 17 00:00:00 2001 From: iamtoruk Date: Mon, 17 Aug 2026 00:11:08 -0700 Subject: [PATCH 1/4] perf(cache): shard the session cache by provider and month A provider's shard held its whole history, so one appended session rewrote 95 MB. Each provider's files are now split by the UTC month of their first turn - a bucket that is stable across appends, so a growing session never migrates shards - and every shard records the newest month it holds so a ranged load can skip the ones that cannot contribute. Dirty tracking is per bucket: markCacheDirty takes an optional file path and marks both the bucket the entry was last saved in and the one it is in now. A save writes only dirty buckets, carries the refs of months it never loaded, and merges the on-disk shard back in when a bucket is dirty but was never loaded. v8 and v7 caches re-lay-out losslessly. --- src/session-cache.ts | 446 +++++++++++++++++++++++++++++++++++-------- 1 file changed, 365 insertions(+), 81 deletions(-) diff --git a/src/session-cache.ts b/src/session-cache.ts index 07d7ef52..90dbc40e 100644 --- a/src/session-cache.ts +++ b/src/session-cache.ts @@ -1,4 +1,4 @@ -import { readFile, stat, open, rename, unlink, readdir, mkdir } from 'fs/promises' +import { readFile, stat, open, rename, unlink, readdir, mkdir, rm } from 'fs/promises' import { existsSync, readFileSync, unlinkSync } from 'fs' import { createHash, randomBytes } from 'crypto' import { join } from 'path' @@ -161,14 +161,21 @@ export type SessionCache = { // shards plus a small envelope, so a launch that only touched one provider // rewrites just that provider's file. The turn shape is unchanged, so a v7 file // migrates losslessly (migrateSingleFileCache) rather than re-parsing. -export const CACHE_VERSION = 8 +// v9: on-disk layout only - a provider's shard split further by the UTC month of +// each cached file, so one appended session rewrites one month instead of the +// provider's whole (100MB-scale) history, and a ranged query loads only the +// months it can possibly report on. Turn shape unchanged, so v8 and v7 both +// migrate losslessly. +export const CACHE_VERSION = 9 // The cache directory is version-suffixed for the same reason the file used to // be: different binaries (an old launchd menubar, a newer desktop app) each own // a distinct layout and can never clobber each other's incompatible schema. const CACHE_DIR_NAME = `session-cache.v${CACHE_VERSION}` -// Written LAST on every save: it names the shard file of every provider, so the -// rename that publishes it is the single point at which a save becomes visible. +// The v8 shard directory, read once by the lossless v8 -> v9 re-layout. +const PRIOR_SHARD_DIR_NAME = 'session-cache.v8' +// Written LAST on every save: it names the shard file of every provider-month, so +// the rename that publishes it is the single point at which a save becomes visible. const ENVELOPE_FILE = 'envelope.json' // The pre-versioning filename. Never written or deleted anymore — old binaries // still own it. On first load we adopt-copy it once (see loadCache) when the @@ -304,34 +311,117 @@ export function sessionCacheDir(): string { return join(getCodeburnCacheDir(), CACHE_DIR_NAME) } +// `until` is the UTC month of the newest turn any file in the shard holds. The +// shard's own key is the month of the OLDEST (a file is bucketed by its first +// turn), so the pair bounds every turn the shard can contribute and a ranged +// load can skip the shard outright when the two do not overlap the query. +type ShardRef = { name: string; until: string } +type EnvelopeProvider = { + envFingerprint: string + durable?: boolean + /** month (`YYYY-MM`, or `0000-00` for turn-less files) -> shard */ + shards: Record +} type CacheEnvelope = { version: number complete?: boolean nonce: string - shards: Record + providers: Record +} + +// Files with no turns (failure markers, empty sessions) have no month to bucket +// by. They live in one always-loaded bucket, which is also what makes the only +// possible re-bucketing safe: a file leaves this bucket the first time it gains +// a turn, and the bucket it leaves is guaranteed to be in memory. +const UNDATED_BUCKET = '0000-00' +// Sentinel inside `dirtyBuckets`: every bucket of the provider is dirty. +const ALL_BUCKETS = '*' + +function monthKey(timestamp: string | undefined): string | null { + if (!timestamp) return null + const ms = Date.parse(timestamp) + if (Number.isNaN(ms)) return null + const d = new Date(ms) + return `${d.getUTCFullYear()}-${String(d.getUTCMonth() + 1).padStart(2, '0')}` +} + +/** The shard bucket a cached file belongs to: the UTC month of its FIRST turn. + * First-turn (not last, not mtime) is the only choice that is stable across + * appends — an appended session must never migrate shards, or the entry would + * be written to a new shard while the old one still holds a stale copy. */ +export function cacheBucketMonth(file: CachedFile): string { + return monthKey(file.turns[0]?.timestamp) ?? UNDATED_BUCKET +} + +// Turns are appended in order, so the last one carries the newest timestamp. +function newestMonth(file: CachedFile): string { + return monthKey(file.turns[file.turns.length - 1]?.timestamp) ?? cacheBucketMonth(file) } // Save bookkeeping, held beside the cache rather than on it so it never lands in -// a shard's JSON or in a caller's deep-equality. `shards` is the -// provider -> shard filename map the last load/save published; a provider that -// is neither dirty nor already sharded is written on the next save. -type CacheState = { dirty: boolean; dirtyProviders: Set; shards: Record } +// a shard's JSON or in a caller's deep-equality. +type CacheState = { + dirty: boolean + /** provider -> dirty months (or `ALL_BUCKETS`). */ + dirtyBuckets: Map> + /** provider -> the shard refs the last load/save published. */ + shards: Map> + /** provider -> months held in memory; `null` when the whole provider loaded. */ + loaded: Map | null> + /** provider -> the envFingerprint the published envelope recorded. */ + fingerprints: Map + /** `provider\0path` -> the bucket the entry was loaded/saved under, so a + * delete or a re-bucketing can dirty the bucket it is leaving. */ + bucketOf: Map + /** The load scope this cache was read under, for the cross-request memo. */ + scope: string +} const cacheStates = new WeakMap() function stateOf(cache: SessionCache): CacheState { let state = cacheStates.get(cache) if (!state) { - state = { dirty: false, dirtyProviders: new Set(), shards: {} } + state = { + dirty: false, + dirtyBuckets: new Map(), + shards: new Map(), + loaded: new Map(), + fingerprints: new Map(), + bucketOf: new Map(), + scope: 'all', + } cacheStates.set(cache, state) } return state } -/** Record that `provider`'s section changed, so the next save rewrites its shard. */ -export function markCacheDirty(cache: SessionCache, provider: string): void { +function markBucketDirty(state: CacheState, provider: string, bucket: string): void { + state.dirty = true + let buckets = state.dirtyBuckets.get(provider) + if (!buckets) { buckets = new Set(); state.dirtyBuckets.set(provider, buckets) } + buckets.add(bucket) +} + +function isBucketDirty(state: CacheState, provider: string, bucket: string): boolean { + const buckets = state.dirtyBuckets.get(provider) + return buckets !== undefined && (buckets.has(ALL_BUCKETS) || buckets.has(bucket)) +} + +/** Record that `provider`'s section changed, so the next save rewrites the + * affected shards. Pass `filePath` whenever the change is scoped to one cached + * file — both the bucket it was last saved in and the bucket it is in now are + * marked, so a delete, a rewrite and a re-bucketing are all covered whichever + * order the caller mutates and marks in. Omitting it dirties every bucket. */ +export function markCacheDirty(cache: SessionCache, provider: string, filePath?: string): void { const state = stateOf(cache) + if (filePath === undefined) { markBucketDirty(state, provider, ALL_BUCKETS); return } + const prior = state.bucketOf.get(`${provider}\0${filePath}`) + if (prior !== undefined) markBucketDirty(state, provider, prior) + const file = cache.providers[provider]?.files[filePath] + if (file) markBucketDirty(state, provider, cacheBucketMonth(file)) + // A path with neither a prior bucket nor a live entry (deleted before this + // process ever saw it) still has to move `dirty`, or the save is skipped. state.dirty = true - state.dirtyProviders.add(provider) } /** True when any provider section changed since the last save. */ @@ -481,6 +571,12 @@ function validateCachedFile(f: unknown): f is CachedFile { && (o['turns'] as unknown[]).every(validateTurn) } +// A shard's payload: the provider's `files` map, restricted to one month. +function validateFiles(v: unknown): v is Record { + if (!v || typeof v !== 'object' || Array.isArray(v)) return false + return Object.values(v as Record).every(validateCachedFile) +} + function validateProviderSection(s: unknown): s is ProviderSection { if (!s || typeof s !== 'object') return false const o = s as Record @@ -589,34 +685,79 @@ async function adoptNewestPriorCache(): Promise { // process mints a new nonce and forces a reload, so cross-process freshness is // preserved; saveCache updates the memo write-through so the object handed out // stays the canonical one after a refresh. -let cacheMemo: { dir: string; nonce: string; cache: SessionCache } | null = null +let cacheMemo: { dir: string; nonce: string; scope: string; cache: SessionCache } | null = null export function clearLoadCacheMemo(): void { cacheMemo = null } +/** Months (UTC `YYYY-MM`, inclusive) a query can possibly report on. */ +export type CacheLoadScope = { fromMonth: string; toMonth: string } + +export function monthScopeForRange(start: Date, end: Date): CacheLoadScope { + return { fromMonth: monthKey(start.toISOString())!, toMonth: monthKey(end.toISOString())! } +} + +function previousMonth(month: string): string { + const [y, m] = month.split('-').map(Number) as [number, number] + return m === 1 ? `${y - 1}-12` : `${y}-${String(m - 1).padStart(2, '0')}` +} + +// A shard is in scope when its [bucket .. until] span overlaps the query. One +// extra month of slack below the range covers the cross-range carries that read +// turns from BEFORE the window: the pre-range PR set / git branch a session +// carries into its first in-range turn (both resolved from the same file, so +// they only need the file loaded at all), and the out-of-range subagent-spawn +// ANCHOR whose in-range child folds into it. The undated bucket has no span and +// is always loaded. +function shardInScope(bucket: string, until: string, scope: CacheLoadScope): boolean { + if (bucket === UNDATED_BUCKET) return true + return bucket <= scope.toMonth && until >= previousMonth(scope.fromMonth) +} + +function isShardRef(v: unknown): v is ShardRef { + if (!v || typeof v !== 'object') return false + const o = v as Record + return typeof o['name'] === 'string' && typeof o['until'] === 'string' +} + function isEnvelope(raw: unknown): raw is CacheEnvelope { if (!raw || typeof raw !== 'object') return false const o = raw as Record - return o['version'] === CACHE_VERSION - && typeof o['nonce'] === 'string' - && !!o['shards'] && typeof o['shards'] === 'object' && !Array.isArray(o['shards']) - && Object.values(o['shards'] as Record).every(v => typeof v === 'string') + if (o['version'] !== CACHE_VERSION || typeof o['nonce'] !== 'string') return false + const providers = o['providers'] + if (!providers || typeof providers !== 'object' || Array.isArray(providers)) return false + return Object.values(providers as Record).every(p => { + if (!p || typeof p !== 'object') return false + const e = p as Record + if (typeof e['envFingerprint'] !== 'string') return false + if (!e['shards'] || typeof e['shards'] !== 'object' || Array.isArray(e['shards'])) return false + return Object.values(e['shards'] as Record).every(isShardRef) + }) } -// A shard that is missing or malformed is treated as an ABSENT provider, not as -// a corrupt cache: only that provider re-parses, instead of the old all-or- -// nothing where one bad turn discarded every provider's history. -async function loadShard(path: string): Promise { +// A shard that is missing or malformed costs exactly the provider-months it +// held, not the provider and never the whole cache: those files re-parse while +// every other month keeps serving. +async function loadShard(path: string): Promise | null> { try { const parsed = JSON.parse(await readFile(path, 'utf-8')) - return validateProviderSection(parsed) ? parsed : null + return validateFiles(parsed) ? parsed : null } catch { return null } } -export async function loadCache(): Promise { +/** + * Read the cache. With a `scope`, only the shards whose months can contribute a + * turn to that range are read — everything else stays on disk and is carried + * across the next save untouched (see saveCache). Durable providers and any + * provider whose recorded fingerprint no longer matches are always read in + * full: the first because its cache is the only surviving record of pruned + * usage, the second because a fingerprint change discards the whole section and + * must see every entry it is discarding. + */ +export async function loadCache(scope?: CacheLoadScope): Promise { const dir = sessionCacheDir() let envelope: CacheEnvelope try { @@ -626,28 +767,56 @@ export async function loadCache(): Promise { } catch { return afterMissingShardCache() } - if (cacheMemo && cacheMemo.dir === dir && cacheMemo.nonce === envelope.nonce) return cacheMemo.cache + const scopeKey = scope ? `${scope.fromMonth}..${scope.toMonth}` : 'all' + if (cacheMemo && cacheMemo.dir === dir && cacheMemo.nonce === envelope.nonce + && (cacheMemo.scope === 'all' || cacheMemo.scope === scopeKey)) return cacheMemo.cache const cache: SessionCache = { version: CACHE_VERSION, providers: {}, complete: envelope.complete === true } - const shards: Record = {} - for (const [provider, file] of Object.entries(envelope.shards)) { - const section = await loadShard(join(dir, file)) - if (!section) continue + const state = stateOf(cache) + const reads: Promise[] = [] + for (const [provider, meta] of Object.entries(envelope.providers)) { + const section: ProviderSection = { + envFingerprint: meta.envFingerprint, + files: {}, + ...(meta.durable ? { durable: true } : {}), + } + // Recorded even when every shard is skipped or unreadable: the section is + // what tells the next save which provider these carried-forward shard refs + // belong to, and what stops the reconcile from re-parsing under a + // fingerprint the envelope already agrees with. cache.providers[provider] = section - shards[provider] = file + const full = !scope || meta.durable === true || meta.envFingerprint !== computeEnvFingerprint(provider) + const loaded: Set | null = full ? null : new Set() + for (const [bucket, ref] of Object.entries(meta.shards)) { + if (loaded && !shardInScope(bucket, ref.until, scope!)) continue + loaded?.add(bucket) + reads.push(loadShard(join(dir, ref.name)).then(files => { + // Unreadable: the bucket counts as loaded-and-empty and is marked + // dirty, so the re-parsed files replace it instead of the stale shard + // being carried forward forever. + if (!files) { markBucketDirty(state, provider, bucket); return } + for (const path of Object.keys(files)) state.bucketOf.set(`${provider}\0${path}`, bucket) + Object.assign(section.files, files) + })) + } + state.loaded.set(provider, loaded) + state.shards.set(provider, meta.shards) + state.fingerprints.set(provider, meta.envFingerprint) } - stateOf(cache).shards = shards - cacheMemo = { dir, nonce: envelope.nonce, cache } + await Promise.all(reads) + state.scope = scopeKey + cacheMemo = { dir, nonce: envelope.nonce, scope: scopeKey, cache } return cache } -// The shard directory is absent/unreadable. Prefer the LOSSLESS re-layout of the -// v7 single-file cache (same turn shape, so nothing re-parses); failing that, +// The shard directory is absent/unreadable. Prefer a LOSSLESS re-layout of the +// newest prior layout that is present (v8 provider shards, then the v7 single +// file — both hold the current turn shape, so nothing re-parses); failing that, // adopt the prior versions' expired-source PR orphans, then the legacy // unversioned file. Either way the shard directory is minted on the next save. async function afterMissingShardCache(): Promise { - const migrated = await migrateSingleFileCache() - if (migrated) return migrated + const relaid = await migrateProviderShardCache() ?? await migrateSingleFileCache() + if (relaid) return relaid const prior = await adoptNewestPriorCache() if (prior) return prior // validateCache requires the version to match, so a different-version legacy @@ -656,10 +825,33 @@ async function afterMissingShardCache(): Promise { return adoptLegacyCache() } -// One-time, lossless migration of the v7 single-file cache: v8 changed the -// on-disk LAYOUT only, so every section moves across verbatim and nothing -// re-parses. Every section is marked dirty so the save below writes each shard; -// the v7 file is removed only once that save has published. +// One-time, lossless re-layout of the v8 per-provider shard directory: v9 +// changed the on-disk LAYOUT only, so every entry moves across verbatim (just +// re-bucketed by month in memory) and nothing re-parses. The v8 directory is +// removed only once the v9 save has published. +async function migrateProviderShardCache(): Promise { + const dir = join(getCodeburnCacheDir(), PRIOR_SHARD_DIR_NAME) + let envelope: { complete?: boolean; shards: Record } + try { + const parsed = JSON.parse(await readFile(join(dir, ENVELOPE_FILE), 'utf-8')) as Record + if (parsed['version'] !== 8 || !parsed['shards'] || typeof parsed['shards'] !== 'object') return null + envelope = parsed as { complete?: boolean; shards: Record } + } catch { + return null + } + const cache: SessionCache = { version: CACHE_VERSION, providers: {}, complete: envelope.complete === true } + await Promise.all(Object.entries(envelope.shards).map(async ([provider, name]) => { + try { + const parsed = JSON.parse(await readFile(join(dir, name), 'utf-8')) + if (validateProviderSection(parsed)) cache.providers[provider] = parsed + } catch { /* one unreadable v8 shard costs that provider, as it already did */ } + })) + return publishRelaidCache(cache, () => rm(dir, { recursive: true, force: true })) +} + +// One-time, lossless re-layout of the v7 single-file cache. v7 never wrote a +// shard directory, so it is migrated straight to v9 without minting a v8 in +// between. async function migrateSingleFileCache(): Promise { const v7Path = join(getCodeburnCacheDir(), priorCacheFile(7)) let parsed: unknown @@ -669,14 +861,18 @@ async function migrateSingleFileCache(): Promise { return null } if (!validateCache(parsed, 7)) return null - const cache: SessionCache = { - version: CACHE_VERSION, - providers: parsed.providers, - complete: parsed.complete === true, - } + return publishRelaidCache( + { version: CACHE_VERSION, providers: parsed.providers, complete: parsed.complete === true }, + () => unlink(v7Path), + ) +} + +// Every section is marked dirty so the save writes each month's shard; the old +// layout is retired only once that save has published. +async function publishRelaidCache(cache: SessionCache, retire: () => Promise): Promise { for (const provider of Object.keys(cache.providers)) markCacheDirty(cache, provider) const published = await saveCache(cache).catch(() => false) - if (published) await retryCacheFileMutation(() => unlink(v7Path)) + if (published) await retryCacheFileMutation(async () => { await retire() }) return cache } @@ -697,8 +893,8 @@ async function adoptLegacyCache(): Promise { // the file the currently-published envelope points at: readers keep seeing a // consistent set until the envelope rename publishes the new one, and a writer // that loses the ownership fence leaves the canonical shards untouched. -function shardFileName(provider: string): string { - return `${provider.replace(/[^A-Za-z0-9_-]/g, '_')}.${randomBytes(8).toString('hex')}.json` +function shardFileName(provider: string, bucket: string): string { + return `${provider.replace(/[^A-Za-z0-9_-]/g, '_')}.${bucket}.${randomBytes(8).toString('hex')}.json` } // The temp name carries a nonce: two processes writing the SAME final path @@ -730,34 +926,94 @@ async function writeFileAtomic(finalPath: string, payload: string): Promise> { + const buckets = new Map>() + for (const [path, file] of Object.entries(section.files)) { + const bucket = cacheBucketMonth(file) + let group = buckets.get(bucket) + if (!group) { group = {}; buckets.set(bucket, group) } + group[path] = file + } + return buckets +} + +function untilMonth(files: Record): string { + let until = UNDATED_BUCKET + for (const file of Object.values(files)) { + const month = newestMonth(file) + if (month > until) until = month + } + return until +} + export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Promise): Promise { const dir = sessionCacheDir() if (!existsSync(dir)) await mkdir(dir, { recursive: true, mode: 0o700 }) const state = stateOf(cache) - const priorShards = state.shards - const shards: Record = {} + const providers: Record = {} const written: string[] = [] + // Deferred so the fence check below stays the last thing before publication. + const payloads = new Map>() - const writeShard = async (provider: string): Promise => { - const name = shardFileName(provider) - await writeFileAtomic(join(dir, name), JSON.stringify(cache.providers[provider])) + const writeShard = async (provider: string, bucket: string): Promise => { + const files = payloads.get(`${provider}\0${bucket}`)! + const name = shardFileName(provider, bucket) + await writeFileAtomic(join(dir, name), JSON.stringify(files)) written.push(name) - shards[provider] = name + return { name, until: untilMonth(files) } } try { - for (const provider of Object.keys(cache.providers)) { - const prior = priorShards[provider] - // `priorShards` is this process's snapshot from its last load or save. - // ANOTHER process may have republished that provider since, unlinking the - // file we are about to name — so reuse is conditional on the file still - // being there, and a vanished one is rewritten from memory. - if (prior && !state.dirtyProviders.has(provider) && existsSync(join(dir, prior))) { - shards[provider] = prior - continue + for (const [provider, section] of Object.entries(cache.providers)) { + const priorRefs = state.shards.get(provider) ?? {} + const loaded = state.loaded.get(provider) ?? null + // A fingerprint change discards the section outright (see + // getOrCreateProviderSection), so the months it did not load must be + // dropped rather than carried — they hold entries under the old + // fingerprint. loadCache never scopes such a provider, so `loaded` is + // null here in practice; the guard is what makes that safe to rely on. + const priorFingerprint = state.fingerprints.get(provider) + const reset = priorFingerprint !== undefined && priorFingerprint !== section.envFingerprint + const refs: Record = {} + + for (const [bucket, files] of bucketFiles(section)) { + const prior = priorRefs[bucket] + // `priorRefs` is this process's snapshot from its last load or save. + // ANOTHER process may have republished that shard since, unlinking the + // file we are about to name — so reuse is conditional on the file still + // being there, and a vanished one is rewritten from memory. + if (prior && !isBucketDirty(state, provider, bucket) && existsSync(join(dir, prior.name))) { + refs[bucket] = prior + continue + } + let payload = files + // Dirty but never loaded: memory holds only the entries this run wrote + // into the bucket, so the shard's other entries have to be merged back + // in or the save would silently drop them. Nothing in an unloaded + // bucket can have been deleted — every delete goes through + // `section.files`, which only holds what was loaded. + if (loaded && !loaded.has(bucket) && prior) { + const onDisk = await loadShard(join(dir, prior.name)) + if (onDisk) payload = { ...onDisk, ...files } + } + payloads.set(`${provider}\0${bucket}`, payload) + refs[bucket] = await writeShard(provider, bucket) + } + + // Months this run never loaded keep their published shard verbatim. This + // is the invariant that makes a scoped load safe to save from. + if (loaded && !reset) { + for (const [bucket, ref] of Object.entries(priorRefs)) { + if (refs[bucket] || loaded.has(bucket)) continue + if (existsSync(join(dir, ref.name))) refs[bucket] = ref + } + } + providers[provider] = { + envFingerprint: section.envFingerprint, + ...(section.durable ? { durable: true } : {}), + shards: refs, } - await writeShard(provider) } // The warm refresh transaction passes an ownership fence. It must be the @@ -771,35 +1027,61 @@ export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Pr } // Last look before publishing: a concurrent save may have unlinked a reused - // shard while this one was writing its own. An envelope must never name a - // file that is already gone — that reads back as a corrupt provider and - // drops its history, including orphans no re-parse can recover. - for (const [provider, name] of Object.entries(shards)) { - if (written.includes(name) || existsSync(join(dir, name))) continue - await writeShard(provider) + // or carried shard while this one was writing its own. An envelope must + // never name a file that is already gone — that reads back as a corrupt + // month and drops its history, including orphans no re-parse can recover. + for (const [provider, meta] of Object.entries(providers)) { + for (const [bucket, ref] of Object.entries(meta.shards)) { + if (written.includes(ref.name) || existsSync(join(dir, ref.name))) continue + if (!payloads.has(`${provider}\0${bucket}`)) { + // A carried month whose file vanished: its content was never in + // memory, so there is nothing to rewrite. Dropping the reference is + // the only honest option, and the sweep retires the name. + delete meta.shards[bucket] + continue + } + meta.shards[bucket] = await writeShard(provider, bucket) + } } const envelope: CacheEnvelope = { version: CACHE_VERSION, complete: cache.complete === true, nonce: randomBytes(8).toString('hex'), - shards, + providers, } await writeFileAtomic(join(dir, ENVELOPE_FILE), JSON.stringify(envelope)) - state.dirty = false - state.dirtyProviders.clear() - state.shards = shards - // Write-through: the object just published IS the freshest state, so the - // next loadCache in this process reuses it instead of re-parsing. - cacheMemo = { dir, nonce: envelope.nonce, cache } // Shards the new envelope no longer references are garbage; a reader that // already opened one keeps reading it, and any failure here is swept later // by cleanupOrphanedTempFiles. - for (const [provider, name] of Object.entries(priorShards)) { - if (shards[provider] === name) continue - await retryCacheFileMutation(() => unlink(join(dir, name))) + const retired: string[] = [] + for (const [provider, priorRefs] of state.shards) { + const kept = providers[provider]?.shards ?? {} + for (const [bucket, ref] of Object.entries(priorRefs)) { + if (kept[bucket]?.name !== ref.name) retired.push(ref.name) + } + } + + state.dirty = false + state.dirtyBuckets.clear() + state.shards.clear() + state.fingerprints.clear() + state.bucketOf.clear() + // `loaded` deliberately survives: a merged-and-rewritten shard is complete + // on disk but still partial in memory, so the next save has to merge again. + for (const [provider, meta] of Object.entries(providers)) { + state.shards.set(provider, meta.shards) + state.fingerprints.set(provider, meta.envFingerprint) + for (const [path, file] of Object.entries(cache.providers[provider]!.files)) { + state.bucketOf.set(`${provider}\0${path}`, cacheBucketMonth(file)) + } } + // Write-through: the object just published IS the freshest state, so the + // next loadCache in this process reuses it instead of re-parsing. Its scope + // is whatever was loaded, not `all` — a save never widens what is in memory. + cacheMemo = { dir, nonce: envelope.nonce, scope: state.scope, cache } + for (const name of retired) await retryCacheFileMutation(() => unlink(join(dir, name))) return true } catch (err) { for (const name of written) await retryCacheFileMutation(() => unlink(join(dir, name))) @@ -988,7 +1270,9 @@ export async function cleanupOrphanedTempFiles(): Promise { try { const parsed = JSON.parse(await readFile(join(dir, ENVELOPE_FILE), 'utf-8')) if (isEnvelope(parsed)) { - for (const name of Object.values(parsed.shards)) referenced.add(name) + for (const meta of Object.values(parsed.providers)) { + for (const ref of Object.values(meta.shards)) referenced.add(ref.name) + } envelopeRead = true } } catch {} From 317fbf05df18933ba4b7abfea221962b5a996242 Mon Sep 17 00:00:00 2001 From: iamtoruk Date: Mon, 17 Aug 2026 00:11:13 -0700 Subject: [PATCH 2/4] perf(parser): mark cache buckets per file and scope the load to the query Every per-file markCacheDirty call site now names the file, so a parse, a re-parse, a failure marker, an orphan eviction and the durable age-out each dirty exactly the month they touched. The two section-level marks (a fingerprint reset, the durable stamp) stay provider-wide. parseAllSessions derives a month scope from its dateRange and threads it through every loadCache call, so a today/week query stops reading the months it cannot report on. --- src/parser.ts | 39 ++++++++++++++++++++++++--------------- 1 file changed, 24 insertions(+), 15 deletions(-) diff --git a/src/parser.ts b/src/parser.ts index 89f6e57d..17ca3f20 100644 --- a/src/parser.ts +++ b/src/parser.ts @@ -26,6 +26,7 @@ import { isCacheDirty, loadCache, markCacheDirty, + monthScopeForRange, reconcileFile, saveCache, } from './session-cache.js' @@ -2015,7 +2016,7 @@ async function scanProjectDirs( // Marked here, not after the re-parse: an unreadable file `continue`s out // below, and the deletion would otherwise live only in memory. delete section.files[filePath] - markCacheDirty(diskCache, 'claude') + markCacheDirty(diskCache, 'claude', filePath) try { if (append) { @@ -2120,7 +2121,7 @@ async function scanProjectDirs( ...(Object.keys(mergedSpawnLinks).length > 0 ? { agentSpawnLinks: mergedSpawnLinks } : {}), ...(mergedAmbiguousIds.length > 0 ? { ambiguousSpawnAgentIds: mergedAmbiguousIds } : {}), } - markCacheDirty(diskCache, 'claude') + markCacheDirty(diskCache, 'claude', filePath) filesDone++ await parseProgress.tick(filesDone) if (filesDone % 50 === 0 || filesDone === progressTotal) { @@ -2157,7 +2158,7 @@ async function scanProjectDirs( ...(Object.keys(sessionMeta.agentSpawnLinks).length > 0 ? { agentSpawnLinks: sessionMeta.agentSpawnLinks } : {}), ...(sessionMeta.ambiguousSpawnAgentIds.length > 0 ? { ambiguousSpawnAgentIds: sessionMeta.ambiguousSpawnAgentIds } : {}), } - markCacheDirty(diskCache, 'claude') + markCacheDirty(diskCache, 'claude', filePath) } catch (err) { // A single malformed Claude session file must not abort the whole run — that // would empty the daily-cache backfill and wipe the trend/history (issue #441, @@ -2165,7 +2166,7 @@ async function scanProjectDirs( // by the current fingerprint so it isn't re-read and re-thrown every run; it // re-parses only if the file changes. section.files[filePath] = { fingerprint: info.fp, mcpInventory: [], turns: [], failed: true } - markCacheDirty(diskCache, 'claude') + markCacheDirty(diskCache, 'claude', filePath) warnProviderParseFailure('claude', filePath, err) } filesDone++ @@ -2186,7 +2187,7 @@ async function scanProjectDirs( // but they carry attributable PR spend (surfaced above as a legacy split). if (section.files[cachedPath]?.prLinks?.length) continue delete section.files[cachedPath] - markCacheDirty(diskCache, 'claude') + markCacheDirty(diskCache, 'claude', cachedPath) } } @@ -2990,7 +2991,7 @@ async function parseProviderSources( // that pruned-away data is preserved for monotonic monthly totals. if (!provider.durableSources && !clearedPaths.has(source.path)) { delete section.files[source.path] - markCacheDirty(diskCache, providerName) + markCacheDirty(diskCache, providerName, source.path) clearedPaths.add(source.path) } @@ -3036,7 +3037,7 @@ async function parseProviderSources( } } didParse = true - markCacheDirty(diskCache, providerName) + markCacheDirty(diskCache, providerName, source.path) } catch (err) { if (isSqliteBusyError(err)) { warnProviderReadFailureOnce(providerName, err) @@ -3049,7 +3050,7 @@ async function parseProviderSources( // on every refresh; it re-parses only if it changes. Empty turns => no // usage contributed. section.files[source.path] = { fingerprint: fp, mcpInventory: [], turns: [], failed: true } - markCacheDirty(diskCache, providerName) + markCacheDirty(diskCache, providerName, source.path) warnProviderParseFailure(providerName, source.path, err) continue } @@ -3073,7 +3074,7 @@ async function parseProviderSources( for (const cachedPath of Object.keys(section.files)) { if (!allDiscoveredFiles.has(cachedPath)) { delete section.files[cachedPath] - markCacheDirty(diskCache, providerName) + markCacheDirty(diskCache, providerName, cachedPath) } } } @@ -3090,7 +3091,7 @@ async function parseProviderSources( .reduce((max, ts) => Math.max(max, ts), 0) if (newestTs > 0 && newestTs < cutoffMs) { delete section.files[cachedPath] - markCacheDirty(diskCache, providerName) + markCacheDirty(diskCache, providerName, cachedPath) } } } @@ -3793,7 +3794,15 @@ async function parseAllSessionsInCacheScope(dateRange?: DateRange, providerFilte if (reused) return reused } - let diskCache = await loadCache() + // Load only the month shards a query over `dateRange` can possibly report + // on. Sessions whose every turn falls outside the range are dropped from the + // report anyway, so skipping their shards changes nothing except the bytes + // read — and a save writes only dirty months, leaving the skipped ones on + // disk untouched (see saveCache). Dedup is unaffected: a duplicate + // deduplicationKey is the same message at the same timestamp, so it can never + // straddle the loaded/skipped boundary. + const loadScope = dateRange ? monthScopeForRange(dateRange.start, dateRange.end) : undefined + let diskCache = await loadCache(loadScope) await cleanupOrphanedTempFiles() // Cold-hydration coordination (advisory, cross-process). Engages whenever the @@ -3806,7 +3815,7 @@ async function parseAllSessionsInCacheScope(dateRange?: DateRange, providerFilte // doubt it proceeds unlocked. if (!isCacheComplete(diskCache)) { const hydration = await beginColdHydration(true) - if (hydration.waited) diskCache = await loadCache() + if (hydration.waited) diskCache = await loadCache(loadScope) const isCold = !isCacheComplete(diskCache) try { return await runParse(key, diskCache, dateRange, providerFilter, { isCold, burstSig, parseStartedAt }) @@ -3824,17 +3833,17 @@ async function parseAllSessionsInCacheScope(dateRange?: DateRange, providerFilte return runParse(key, priorSnapshot, dateRange, providerFilter, { readOnly: true, burstSig, parseStartedAt }) } if (refresh.outcome === 'completed-by-other') { - return runParse(key, await loadCache(), dateRange, providerFilter, { readOnly: true, burstSig, parseStartedAt }) + return runParse(key, await loadCache(loadScope), dateRange, providerFilter, { readOnly: true, burstSig, parseStartedAt }) } try { // 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() + diskCache = await loadCache(loadScope) return await runParse(key, diskCache, dateRange, providerFilter, { refreshLock: refresh.handle, burstSig, parseStartedAt }) } catch (err) { if (!(err instanceof RefreshFenceLostError) && !(err instanceof RefreshPublicationUnavailableError)) throw err - return runParse(key, await loadCache(), dateRange, providerFilter, { readOnly: true, burstSig, parseStartedAt }) + return runParse(key, await loadCache(loadScope), dateRange, providerFilter, { readOnly: true, burstSig, parseStartedAt }) } finally { await refresh.handle.release() } From a2cc887a2d9fa1495e17f8d352fc5ec482c75c43 Mon Sep 17 00:00:00 2001 From: iamtoruk Date: Mon, 17 Aug 2026 00:11:15 -0700 Subject: [PATCH 3/4] test(cache): cover month buckets, scoped loads and the v8 re-layout --- CHANGELOG.md | 1 + tests/session-cache-shards.test.ts | 282 +++++++++++++++++++++++++++-- 2 files changed, 267 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e18aabf9..a3f08f81 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ## Unreleased ### Changed +- **A warm launch rewrites only the month that changed, and a ranged query reads only the months it can report on.** Per-provider shards still meant one appended session republished that provider's entire history — 95 MB for Claude on a 6 GB corpus. Each provider's shard is now split again by the UTC month of the cached session's FIRST turn, a bucket that never moves as a session grows, so an append rewrites one month. Every shard records the newest month it holds, which lets `--period today/week` skip the shards that cannot contribute a turn to the range; the skipped months stay on disk untouched across the save, and providers whose cache is the only surviving record (durable) or whose parse fingerprint moved are always read in full. Remaining shards are read concurrently. Existing v8 and v7 caches are re-laid-out losslessly on first load and the old layout removed once the new one is published: nothing re-parses. - **A warm launch rewrites only the provider that changed.** The session cache was a single blob, so any provider appending a few KB republished the whole thing — 147 MB of stringify + fsync on a 6 GB corpus, ~18% of a warm run. It is now a version-suffixed directory holding one shard per provider plus a small envelope, written per provider and published by a single envelope rename. An existing v7 cache is re-laid-out losslessly on first load and the old file removed once the new layout is on disk: nothing re-parses. One unreadable shard now costs that provider a re-parse instead of discarding every provider's history, and partial saves during a cold parse are triggered every 2000 files rather than every 5 seconds, so a slow cold parse no longer rewrites the growing cache on a wall clock. - **An appended Codex rollout parses only its tail.** Rollout files are append-only and the active ones run to hundreds of MB, but the Codex result cache keyed on mtime + size alone, so any growth re-read the file from byte 0. Each entry now records a restart point at the last task boundary — byte offset plus the state the single-pass decode carries across it — and a grown file with the same inode resumes there, producing output identical to a full re-parse. An entry without a usable restart point simply re-parses in full once and gains one. - **A date-ranged report classifies only the turns it keeps.** Every cached turn went through the turn classifier — category, retries, edit detection, and a full reconstruction of its API calls — before the date slice discarded most of them, so a week view paid to classify all of history to keep a few percent of it. The keep/drop decision is now taken on the raw cached turn and only the survivors are classified, still from their complete call list, with the branch and pull-request carries still walking the full ordered turn list. Output is byte-identical. diff --git a/tests/session-cache-shards.test.ts b/tests/session-cache-shards.test.ts index a448a597..465cc6a5 100644 --- a/tests/session-cache-shards.test.ts +++ b/tests/session-cache-shards.test.ts @@ -1,8 +1,9 @@ -// Per-provider shard layout (CACHE_VERSION 8): the on-disk cache is a directory -// holding one envelope plus one shard per provider. What matters here is that -// the move off the single v7 blob loses nothing, that a save rewrites only the -// providers that changed, and that one unreadable shard costs exactly one -// provider instead of the whole cache. +// Provider x month shard layout (CACHE_VERSION 9): the on-disk cache is a +// directory holding one envelope plus one shard per provider-month. What matters +// here is that the move off the older layouts loses nothing, that a file's +// bucket never moves when the session is appended to, that a save rewrites only +// the months that changed (including when the load was scoped to a subset of +// them), and that one unreadable shard costs exactly one month. import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { mkdir, readFile, readdir, rm, stat, utimes, writeFile } from 'fs/promises' import { existsSync } from 'fs' @@ -11,10 +12,13 @@ import { join } from 'path' import { CACHE_VERSION, + cacheBucketMonth, + computeEnvFingerprint, cleanupOrphanedTempFiles, clearLoadCacheMemo, loadCache, markCacheDirty, + monthScopeForRange, saveCache, sessionCacheDir, type CachedFile, @@ -34,6 +38,15 @@ afterEach(async () => { if (existsSync(TMP_DIR)) await rm(TMP_DIR, { recursive: true }) }) +function turnAt(timestamp: string, key = 'msg-1'): CachedFile['turns'][number] { + const base = cachedFile().turns[0]! + return { ...base, timestamp, calls: [{ ...base.calls[0]!, timestamp, deduplicationKey: key }] } +} + +function fileSpanning(first: string, last?: string): CachedFile { + return cachedFile({ turns: last ? [turnAt(first, 'a'), turnAt(last, 'b')] : [turnAt(first, 'a')] }) +} + function cachedFile(overrides: Partial = {}): CachedFile { return { fingerprint: { dev: 1, ino: 2, mtimeMs: 3, sizeBytes: 4 }, @@ -98,6 +111,18 @@ async function shardNames(): Promise { return (await readdir(sessionCacheDir())).sort() } +async function envelope(): Promise<{ providers: Record }> }> { + return JSON.parse(await readFile(join(sessionCacheDir(), 'envelope.json'), 'utf-8')) +} + +/** name -> bytes, for every shard on disk. */ +async function shardBytes(): Promise> { + const dir = sessionCacheDir() + const out = new Map() + for (const name of await shardNames()) out.set(name, await readFile(join(dir, name), 'utf-8')) + return out +} + describe('v7 -> shard migration', () => { it('is lossless: every entry survives, shards replace the v7 file, reload matches', async () => { const v7 = v7Cache() @@ -112,8 +137,9 @@ describe('v7 -> shard migration', () => { expect(existsSync(v7Path)).toBe(false) const names = await shardNames() expect(names).toContain('envelope.json') - expect(names.filter(n => n.startsWith('claude.'))).toHaveLength(1) - expect(names.filter(n => n.startsWith('codex.'))).toHaveLength(1) + // claude's three entries split by month: two dated 2026-05, one turn-less. + expect(Object.keys((await envelope()).providers['claude']!.shards).sort()).toEqual(['0000-00', '2026-05']) + expect(Object.keys((await envelope()).providers['codex']!.shards)).toEqual(['2026-05']) // A second load reads only the shards and produces the same cache. clearLoadCacheMemo() @@ -160,18 +186,37 @@ describe('per-provider dirty tracking', () => { }) describe('corrupt shard isolation', () => { - it('drops only the unreadable provider, keeping the rest intact', async () => { - await writeFile(join(TMP_DIR, 'session-cache.v7.json'), JSON.stringify(v7Cache())) - const cache = await loadCache() + it('drops only the unreadable month, keeping every other month and provider', async () => { + const cache: SessionCache = { + version: CACHE_VERSION, + complete: true, + providers: { + claude: { + envFingerprint: 'claude-fp', + files: { '/live/may.jsonl': fileSpanning('2026-05-15T10:00:00Z'), '/live/jun.jsonl': fileSpanning('2026-06-15T10:00:00Z') }, + }, + codex: { envFingerprint: 'codex-fp', files: { '/live/r.jsonl': fileSpanning('2026-05-15T10:00:00Z') } }, + }, + } + markCacheDirty(cache, 'claude') + markCacheDirty(cache, 'codex') + await saveCache(cache) const dir = sessionCacheDir() - const claudeShard = (await shardNames()).find(n => n.startsWith('claude.'))! - await writeFile(join(dir, claudeShard), '{"envFingerprint":"claude-fp","files":{"/x":{"turns":') + await writeFile(join(dir, (await envelope()).providers['claude']!.shards['2026-05']!.name), '{"/x":{"turns":') clearLoadCacheMemo() const reloaded = await loadCache() - expect(reloaded.providers['claude']).toBeUndefined() + expect(Object.keys(reloaded.providers['claude']!.files)).toEqual(['/live/jun.jsonl']) expect(reloaded.providers['codex']).toEqual(cache.providers['codex']) + + // Self-heals: the unreadable month is republished from whatever re-parses + // into it rather than being carried forward corrupt forever. + reloaded.providers['claude']!.files['/live/may.jsonl'] = fileSpanning('2026-05-15T10:00:00Z') + markCacheDirty(reloaded, 'claude', '/live/may.jsonl') + await saveCache(reloaded) + clearLoadCacheMemo() + expect(Object.keys((await loadCache()).providers['claude']!.files).sort()).toEqual(['/live/jun.jsonl', '/live/may.jsonl']) }) }) @@ -254,9 +299,10 @@ describe('concurrent writers', () => { async function assertReferentialIntegrity(expected: string[]): Promise { const dir = sessionCacheDir() - const envelope = JSON.parse(await readFile(join(dir, 'envelope.json'), 'utf-8')) - for (const name of Object.values(envelope.shards) as string[]) { - expect(existsSync(join(dir, name)), `envelope names a missing shard: ${name}`).toBe(true) + for (const meta of Object.values((await envelope()).providers)) { + for (const ref of Object.values(meta.shards)) { + expect(existsSync(join(dir, ref.name)), `envelope names a missing shard: ${ref.name}`).toBe(true) + } } clearLoadCacheMemo() const loaded = await loadCache() @@ -314,3 +360,207 @@ describe('concurrent writers', () => { expect(final.providers['codex']!.files['/live/r2.jsonl']).toBeDefined() }) }) + +describe('month buckets', () => { + it('keeps a file in its first-turn month when the session is appended to', async () => { + const cache: SessionCache = { + version: CACHE_VERSION, + complete: true, + providers: { claude: { envFingerprint: 'fp', files: { '/live/long.jsonl': fileSpanning('2026-05-15T10:00:00Z') } } }, + } + markCacheDirty(cache, 'claude') + await saveCache(cache) + expect(Object.keys((await envelope()).providers['claude']!.shards)).toEqual(['2026-05']) + + // Two months of appends later the bucket is unchanged; only `until` moves, + // which is what lets a ranged load still find this session. + const appended = fileSpanning('2026-05-15T10:00:00Z', '2026-07-02T10:00:00Z') + expect(cacheBucketMonth(appended)).toBe('2026-05') + cache.providers['claude']!.files['/live/long.jsonl'] = appended + markCacheDirty(cache, 'claude', '/live/long.jsonl') + await saveCache(cache) + const shards = (await envelope()).providers['claude']!.shards + expect(Object.keys(shards)).toEqual(['2026-05']) + expect(shards['2026-05']!.until).toBe('2026-07') + + // ...and a July query still loads it, despite the May bucket key. + clearLoadCacheMemo() + const scoped = await loadCache(monthScopeForRange(new Date('2026-07-01T00:00:00Z'), new Date('2026-07-31T23:59:59Z'))) + expect(scoped.providers['claude']!.files['/live/long.jsonl']).toBeDefined() + }) + + it('rewrites only the month that changed', async () => { + const cache: SessionCache = { + version: CACHE_VERSION, + complete: true, + providers: { + claude: { + envFingerprint: 'fp', + files: { + '/live/mar.jsonl': fileSpanning('2026-03-10T10:00:00Z'), + '/live/apr.jsonl': fileSpanning('2026-04-10T10:00:00Z'), + '/live/may.jsonl': fileSpanning('2026-05-10T10:00:00Z'), + }, + }, + }, + } + markCacheDirty(cache, 'claude') + await saveCache(cache) + const before = await shardBytes() + const untouched = [ + (await envelope()).providers['claude']!.shards['2026-03']!.name, + (await envelope()).providers['claude']!.shards['2026-04']!.name, + ] + + cache.providers['claude']!.files['/live/may.jsonl'] = cachedFile({ turns: [turnAt('2026-05-10T10:00:00Z', 'a')], mcpInventory: ['changed'] }) + markCacheDirty(cache, 'claude', '/live/may.jsonl') + await saveCache(cache) + + const after = await shardBytes() + for (const name of untouched) expect(after.get(name)).toBe(before.get(name)) + expect(after.has((await envelope()).providers['claude']!.shards['2026-05']!.name)).toBe(true) + }) + + it('dirties the month a deleted file was in', async () => { + const cache: SessionCache = { + version: CACHE_VERSION, + complete: true, + providers: { + claude: { + envFingerprint: 'fp', + files: { '/live/mar.jsonl': fileSpanning('2026-03-10T10:00:00Z'), '/live/mar2.jsonl': fileSpanning('2026-03-11T10:00:00Z') }, + }, + }, + } + markCacheDirty(cache, 'claude') + await saveCache(cache) + + delete cache.providers['claude']!.files['/live/mar2.jsonl'] + markCacheDirty(cache, 'claude', '/live/mar2.jsonl') + await saveCache(cache) + + clearLoadCacheMemo() + expect(Object.keys((await loadCache()).providers['claude']!.files)).toEqual(['/live/mar.jsonl']) + }) +}) + +describe('scoped load', () => { + async function seedThreeMonths(): Promise { + const cache: SessionCache = { + version: CACHE_VERSION, + complete: true, + providers: { + claude: { + envFingerprint: computeEnvFingerprint('claude'), + files: { + '/live/mar.jsonl': fileSpanning('2026-03-10T10:00:00Z'), + '/live/apr.jsonl': fileSpanning('2026-04-10T10:00:00Z'), + '/live/jun.jsonl': fileSpanning('2026-06-10T10:00:00Z'), + }, + }, + }, + } + markCacheDirty(cache, 'claude') + await saveCache(cache) + } + + const juneScope = monthScopeForRange(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-30T23:59:59Z')) + + it('reads only the months the range can report on, plus one of slack', async () => { + await seedThreeMonths() + clearLoadCacheMemo() + const scoped = await loadCache(juneScope) + // June is in range; May would be the slack month (absent here); March and + // April cannot contribute a June turn and stay on disk. + expect(Object.keys(scoped.providers['claude']!.files)).toEqual(['/live/jun.jsonl']) + }) + + it('save from a scoped load leaves the unloaded months byte-identical', async () => { + await seedThreeMonths() + const before = await shardBytes() + const kept = [ + (await envelope()).providers['claude']!.shards['2026-03']!.name, + (await envelope()).providers['claude']!.shards['2026-04']!.name, + ] + + clearLoadCacheMemo() + const scoped = await loadCache(juneScope) + scoped.providers['claude']!.files['/live/jun2.jsonl'] = fileSpanning('2026-06-20T10:00:00Z') + markCacheDirty(scoped, 'claude', '/live/jun2.jsonl') + await saveCache(scoped) + + const after = await shardBytes() + for (const name of kept) expect(after.get(name), `unloaded month rewritten: ${name}`).toBe(before.get(name)) + + clearLoadCacheMemo() + const full = await loadCache() + expect(Object.keys(full.providers['claude']!.files).sort()) + .toEqual(['/live/apr.jsonl', '/live/jun.jsonl', '/live/jun2.jsonl', '/live/mar.jsonl']) + }) + + it('merges rather than replaces when a re-parse lands in an unloaded month', async () => { + await seedThreeMonths() + clearLoadCacheMemo() + const scoped = await loadCache(juneScope) + // A March session that was never loaded is re-parsed (its mtime moved) and + // written straight back into the March bucket. + scoped.providers['claude']!.files['/live/mar.jsonl'] = cachedFile({ turns: [turnAt('2026-03-10T10:00:00Z', 'z')], mcpInventory: ['reparsed'] }) + markCacheDirty(scoped, 'claude', '/live/mar.jsonl') + await saveCache(scoped) + + clearLoadCacheMemo() + const full = await loadCache() + expect(full.providers['claude']!.files['/live/mar.jsonl']!.mcpInventory).toEqual(['reparsed']) + expect(Object.keys(full.providers['claude']!.files).sort()) + .toEqual(['/live/apr.jsonl', '/live/jun.jsonl', '/live/mar.jsonl']) + }) + + it('never scopes a provider whose fingerprint moved, or a durable one', async () => { + const cache: SessionCache = { + version: CACHE_VERSION, + complete: true, + providers: { + claude: { envFingerprint: 'stale-fp', files: { '/live/mar.jsonl': fileSpanning('2026-03-10T10:00:00Z') } }, + copilot: { envFingerprint: computeEnvFingerprint('copilot'), durable: true, files: { '/live/otel.db': fileSpanning('2026-03-10T10:00:00Z') } }, + }, + } + markCacheDirty(cache, 'claude') + markCacheDirty(cache, 'copilot') + await saveCache(cache) + + clearLoadCacheMemo() + const scoped = await loadCache(juneScope) + // Both would be skipped on month alone; both are read in full anyway, so the + // fingerprint reset and the durable orphan carry-forward see every entry. + expect(scoped.providers['claude']!.files['/live/mar.jsonl']).toBeDefined() + expect(scoped.providers['copilot']!.files['/live/otel.db']).toBeDefined() + }) +}) + +describe('v8 -> v9 migration', () => { + it('re-buckets the v8 provider shards losslessly and retires the v8 directory', async () => { + const v8Dir = join(TMP_DIR, 'session-cache.v8') + await mkdir(v8Dir, { recursive: true }) + const section = { + envFingerprint: 'claude-fp', + files: { + '/live/mar.jsonl': fileSpanning('2026-03-10T10:00:00Z'), + '/live/jun.jsonl': fileSpanning('2026-06-10T10:00:00Z'), + '/gone/pruned.jsonl': cachedFile({ prLinks: ['https://github.com/o/r/pull/1'] }), + }, + } + await writeFile(join(v8Dir, 'claude.abc.json'), JSON.stringify(section)) + await writeFile(join(v8Dir, 'envelope.json'), JSON.stringify({ + version: 8, complete: true, nonce: 'n', shards: { claude: 'claude.abc.json' }, + })) + + const loaded = await loadCache() + expect(loaded.providers['claude']).toEqual(section) + expect(loaded.complete).toBe(true) + expect(existsSync(v8Dir)).toBe(false) + expect(Object.keys((await envelope()).providers['claude']!.shards).sort()).toEqual(['2026-03', '2026-05', '2026-06']) + + clearLoadCacheMemo() + expect(await loadCache()).toEqual(loaded) + }) +}) From 148e7b2151262574f57046e3e3017279e38f1a75 Mon Sep 17 00:00:00 2001 From: iamtoruk Date: Mon, 17 Aug 2026 00:42:35 -0700 Subject: [PATCH 4/4] fix(cache): close three data-integrity holes in the month-shard layout A file's shard span was read off turns[0]/turns[-1], but several providers emit turns non-chronologically (cursor by ROWID, goose/crush/copilot by a DESC ordering). That produced until < bucket - an empty span, so the shard was unreachable at every scope and its sessions re-parsed every run. cacheFileSpan now takes the min and max month over all turns. An entry re-bucketing out of a month the run never loaded (a re-parse that moved its oldest turn, or the #441 failure marker that has no turns at all) left the old copy in the carried shard, so one path lived in two shards and a later load could resolve to the stale one. A save now prunes those paths from the shards it carries, and a load merges shards in envelope order, resolving any duplicate to the freshest fingerprint and dirtying both buckets so the next save retires the loser. A carried month whose shard another writer had republished was dropped from the envelope outright, losing expired-transcript PR orphans no re-parse can recover. The envelope is now re-read just before publishing and the current shard name adopted; a ref is dropped only when that envelope lacks it too. The same re-read moves every merge read after the ownership fence and gives the merge one optimistic retry, so the read-modify-write window shrinks to the publish itself. Also: retire an orphaned v8 directory / v7 file left by an interrupted re-layout, age-guarded, once a v9 envelope is published. --- src/parser.ts | 9 +- src/session-cache.ts | 331 ++++++++++++++++++++--------- tests/session-cache-shards.test.ts | 236 ++++++++++++++++++++ 3 files changed, 475 insertions(+), 101 deletions(-) diff --git a/src/parser.ts b/src/parser.ts index 17ca3f20..91bbffed 100644 --- a/src/parser.ts +++ b/src/parser.ts @@ -3798,9 +3798,12 @@ async function parseAllSessionsInCacheScope(dateRange?: DateRange, providerFilte // on. Sessions whose every turn falls outside the range are dropped from the // report anyway, so skipping their shards changes nothing except the bytes // read — and a save writes only dirty months, leaving the skipped ones on - // disk untouched (see saveCache). Dedup is unaffected: a duplicate - // deduplicationKey is the same message at the same timestamp, so it can never - // straddle the loaded/skipped boundary. + // disk untouched (see saveCache). Cross-file dedup is weakened, not broken: + // the pre-seed of `seenMsgIds` / `seenKeys` only covers loaded files, so a key + // that a skipped file also holds is no longer suppressed. Totals are + // unaffected (a suppressed duplicate contributes nothing either way), but for + // a proxied key emitted under two providers the attribution can land on a + // different provider than a full load would pick. const loadScope = dateRange ? monthScopeForRange(dateRange.start, dateRange.end) : undefined let diskCache = await loadCache(loadScope) await cleanupOrphanedTempFiles() diff --git a/src/session-cache.ts b/src/session-cache.ts index 90dbc40e..7fcb23a0 100644 --- a/src/session-cache.ts +++ b/src/session-cache.ts @@ -345,17 +345,28 @@ function monthKey(timestamp: string | undefined): string | null { return `${d.getUTCFullYear()}-${String(d.getUTCMonth() + 1).padStart(2, '0')}` } -/** The shard bucket a cached file belongs to: the UTC month of its FIRST turn. - * First-turn (not last, not mtime) is the only choice that is stable across - * appends — an appended session must never migrate shards, or the entry would - * be written to a new shard while the old one still holds a stale copy. */ -export function cacheBucketMonth(file: CachedFile): string { - return monthKey(file.turns[0]?.timestamp) ?? UNDATED_BUCKET +/** The UTC month span a cached file covers: `bucket` is its OLDEST turn's month + * (the shard it lives in), `until` its NEWEST (how far forward the shard can + * contribute). Both scan every turn rather than reading turns[0]/turns[-1]: + * several providers emit turns out of chronological order (cursor composers by + * ROWID, goose/crush/copilot by a DESC ordering), and a `until < bucket` span + * is empty, which makes the shard unreachable at EVERY scope. */ +export function cacheFileSpan(file: CachedFile): { bucket: string; until: string } { + let bucket: string | null = null + let until: string | null = null + for (const turn of file.turns) { + const month = monthKey(turn.timestamp) + if (month === null) continue + if (bucket === null || month < bucket) bucket = month + if (until === null || month > until) until = month + } + return bucket === null ? { bucket: UNDATED_BUCKET, until: UNDATED_BUCKET } : { bucket, until: until! } } -// Turns are appended in order, so the last one carries the newest timestamp. -function newestMonth(file: CachedFile): string { - return monthKey(file.turns[file.turns.length - 1]?.timestamp) ?? cacheBucketMonth(file) +/** The shard bucket a cached file belongs to. Derived from the file's own turns, + * so an APPEND never moves it: appending can only extend `until`. */ +export function cacheBucketMonth(file: CachedFile): string { + return cacheFileSpan(file).bucket } // Save bookkeeping, held beside the cache rather than on it so it never lands in @@ -691,7 +702,11 @@ export function clearLoadCacheMemo(): void { cacheMemo = null } -/** Months (UTC `YYYY-MM`, inclusive) a query can possibly report on. */ +/** Months (UTC `YYYY-MM`, inclusive) a query can possibly report on. The load + * widens this by one month BELOW `fromMonth` and none above (see + * shardInScope): every cross-range carry in the report reads BACKWARDS from the + * first in-range turn, never forwards, so there is nothing above the range to + * reach for. */ export type CacheLoadScope = { fromMonth: string; toMonth: string } export function monthScopeForRange(start: Date, end: Date): CacheLoadScope { @@ -704,12 +719,16 @@ function previousMonth(month: string): string { } // A shard is in scope when its [bucket .. until] span overlaps the query. One -// extra month of slack below the range covers the cross-range carries that read -// turns from BEFORE the window: the pre-range PR set / git branch a session -// carries into its first in-range turn (both resolved from the same file, so -// they only need the file loaded at all), and the out-of-range subagent-spawn -// ANCHOR whose in-range child folds into it. The undated bucket has no span and -// is always loaded. +// extra month of slack BELOW the range (and none above — every carry reads +// backwards) covers the cross-range carries that read turns from before the +// window: the pre-range PR set / git branch a session carries into its first +// in-range turn (both resolved from the same file, so they only need the file +// loaded at all), and the out-of-range subagent-spawn ANCHOR whose in-range +// child folds into it. LIMITATION: an anchor whose last turn is two or more +// months before its child's is not loaded, so that child attributes without the +// parent's PR set. One month of slack is the deliberate ceiling; widening it +// gives back the read savings the scope exists for. +// The undated bucket has no span and is always loaded. function shardInScope(bucket: string, until: string, scope: CacheLoadScope): boolean { if (bucket === UNDATED_BUCKET) return true return bucket <= scope.toMonth && until >= previousMonth(scope.fromMonth) @@ -736,6 +755,15 @@ function isEnvelope(raw: unknown): raw is CacheEnvelope { }) } +async function readEnvelope(dir: string): Promise { + try { + const parsed = JSON.parse(await readFile(join(dir, ENVELOPE_FILE), 'utf-8')) + return isEnvelope(parsed) ? parsed : null + } catch { + return null + } +} + // A shard that is missing or malformed costs exactly the provider-months it // held, not the provider and never the whole cache: those files re-parse while // every other month keeps serving. @@ -759,14 +787,8 @@ async function loadShard(path: string): Promise | nul */ export async function loadCache(scope?: CacheLoadScope): Promise { const dir = sessionCacheDir() - let envelope: CacheEnvelope - try { - const parsed = JSON.parse(await readFile(join(dir, ENVELOPE_FILE), 'utf-8')) - if (!isEnvelope(parsed)) return afterMissingShardCache() - envelope = parsed - } catch { - return afterMissingShardCache() - } + const envelope = await readEnvelope(dir) + if (!envelope) return afterMissingShardCache() const scopeKey = scope ? `${scope.fromMonth}..${scope.toMonth}` : 'all' if (cacheMemo && cacheMemo.dir === dir && cacheMemo.nonce === envelope.nonce && (cacheMemo.scope === 'all' || cacheMemo.scope === scopeKey)) return cacheMemo.cache @@ -787,18 +809,36 @@ export async function loadCache(scope?: CacheLoadScope): Promise { cache.providers[provider] = section const full = !scope || meta.durable === true || meta.envFingerprint !== computeEnvFingerprint(provider) const loaded: Set | null = full ? null : new Set() + // Shards are read concurrently but merged in envelope order, so the result + // never depends on which read finished first. A path that somehow ended up + // in two shards resolves to the FRESHEST fingerprint and dirties both + // buckets, so the next save prunes the loser instead of letting it linger. + const pending: { bucket: string; files: Promise | null> }[] = [] for (const [bucket, ref] of Object.entries(meta.shards)) { if (loaded && !shardInScope(bucket, ref.until, scope!)) continue loaded?.add(bucket) - reads.push(loadShard(join(dir, ref.name)).then(files => { + pending.push({ bucket, files: loadShard(join(dir, ref.name)) }) + } + reads.push((async () => { + for (const { bucket, files: read } of pending) { + const files = await read // Unreadable: the bucket counts as loaded-and-empty and is marked // dirty, so the re-parsed files replace it instead of the stale shard // being carried forward forever. - if (!files) { markBucketDirty(state, provider, bucket); return } - for (const path of Object.keys(files)) state.bucketOf.set(`${provider}\0${path}`, bucket) - Object.assign(section.files, files) - })) - } + if (!files) { markBucketDirty(state, provider, bucket); continue } + for (const [path, file] of Object.entries(files)) { + const key = `${provider}\0${path}` + const seenIn = state.bucketOf.get(key) + if (seenIn !== undefined) { + markBucketDirty(state, provider, seenIn) + markBucketDirty(state, provider, bucket) + if (section.files[path]!.fingerprint.mtimeMs >= file.fingerprint.mtimeMs) continue + } + state.bucketOf.set(key, bucket) + section.files[path] = file + } + } + })()) state.loaded.set(provider, loaded) state.shards.set(provider, meta.shards) state.fingerprints.set(provider, meta.envFingerprint) @@ -926,45 +966,74 @@ async function writeFileAtomic(finalPath: string, payload: string): Promise> { - const buckets = new Map>() +function bucketFiles(section: ProviderSection): { groups: Map>; until: Map } { + const groups = new Map>() + const until = new Map() for (const [path, file] of Object.entries(section.files)) { - const bucket = cacheBucketMonth(file) - let group = buckets.get(bucket) - if (!group) { group = {}; buckets.set(bucket, group) } + const span = cacheFileSpan(file) + let group = groups.get(span.bucket) + if (!group) { group = {}; groups.set(span.bucket, group) } group[path] = file + const seen = until.get(span.bucket) + if (seen === undefined || span.until > seen) until.set(span.bucket, span.until) } - return buckets + return { groups, until } } function untilMonth(files: Record): string { let until = UNDATED_BUCKET for (const file of Object.values(files)) { - const month = newestMonth(file) + const month = cacheFileSpan(file).until if (month > until) until = month } return until } +// What a save has decided about one provider, carried across the ownership +// fence so every shard READ that a save needs happens as late as possible (see +// the phase-two comment in saveCache). +type ProviderPlan = { + section: ProviderSection + groups: Map> + loaded: Set | null + priorRefs: Record + reset: boolean + /** Paths that may ALSO still sit in a shard this run never loaded. */ + moved: Set + /** Buckets whose payload has to be merged with the published shard first. */ + deferred: string[] + /** bucket -> the shard name the merge was built from, for the retry below. */ + mergedFrom: Map + refs: Record +} + export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Promise): Promise { const dir = sessionCacheDir() if (!existsSync(dir)) await mkdir(dir, { recursive: true, mode: 0o700 }) const state = stateOf(cache) - const providers: Record = {} - const written: string[] = [] - // Deferred so the fence check below stays the last thing before publication. - const payloads = new Map>() + const written = new Set() + const plans = new Map() - const writeShard = async (provider: string, bucket: string): Promise => { - const files = payloads.get(`${provider}\0${bucket}`)! + const writeShard = async (provider: string, bucket: string, files: Record): Promise => { const name = shardFileName(provider, bucket) await writeFileAtomic(join(dir, name), JSON.stringify(files)) - written.push(name) + written.add(name) return { name, until: untilMonth(files) } } + // Overlay this run's entries for `bucket` onto the published shard `from`, + // minus any path that has since moved to another month. + const mergeShard = async (provider: string, plan: ProviderPlan, bucket: string, from: string | undefined): Promise => { + const files = plan.groups.get(bucket)! + const onDisk = from ? await loadShard(join(dir, from)) : null + if (!onDisk) return writeShard(provider, bucket, files) + for (const path of plan.moved) delete onDisk[path] + return writeShard(provider, bucket, { ...onDisk, ...files }) + } + try { + // ── Phase one: everything that can be written from memory alone ────── for (const [provider, section] of Object.entries(cache.providers)) { const priorRefs = state.shards.get(provider) ?? {} const loaded = state.loaded.get(provider) ?? null @@ -975,44 +1044,41 @@ export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Pr // null here in practice; the guard is what makes that safe to rely on. const priorFingerprint = state.fingerprints.get(provider) const reset = priorFingerprint !== undefined && priorFingerprint !== section.envFingerprint - const refs: Record = {} + const { groups } = bucketFiles(section) + const plan: ProviderPlan = { section, groups, loaded, priorRefs, reset, moved: new Set(), deferred: [], mergedFrom: new Map(), refs: {} } + plans.set(provider, plan) + + // An entry whose bucket this run never loaded may ALSO still exist, under + // an older month, in a shard we are about to carry across verbatim — a + // re-parse that shifted the file's oldest turn, or (the common #441 path) + // a parse failure that left a turn-less marker with no month at all. Left + // alone, the path would live in two shards at once and a later load could + // resolve to the stale copy. Both cases are rare, so the prune they + // trigger below reads shards it otherwise would not. + if (loaded) { + for (const [path, file] of Object.entries(section.files)) { + if (state.bucketOf.has(`${provider}\0${path}`)) continue + const bucket = cacheFileSpan(file).bucket + if (!loaded.has(bucket) || bucket === UNDATED_BUCKET) plan.moved.add(path) + } + } - for (const [bucket, files] of bucketFiles(section)) { + for (const [bucket, files] of groups) { const prior = priorRefs[bucket] // `priorRefs` is this process's snapshot from its last load or save. // ANOTHER process may have republished that shard since, unlinking the // file we are about to name — so reuse is conditional on the file still // being there, and a vanished one is rewritten from memory. if (prior && !isBucketDirty(state, provider, bucket) && existsSync(join(dir, prior.name))) { - refs[bucket] = prior + plan.refs[bucket] = prior continue } - let payload = files // Dirty but never loaded: memory holds only the entries this run wrote - // into the bucket, so the shard's other entries have to be merged back - // in or the save would silently drop them. Nothing in an unloaded - // bucket can have been deleted — every delete goes through - // `section.files`, which only holds what was loaded. - if (loaded && !loaded.has(bucket) && prior) { - const onDisk = await loadShard(join(dir, prior.name)) - if (onDisk) payload = { ...onDisk, ...files } - } - payloads.set(`${provider}\0${bucket}`, payload) - refs[bucket] = await writeShard(provider, bucket) - } - - // Months this run never loaded keep their published shard verbatim. This - // is the invariant that makes a scoped load safe to save from. - if (loaded && !reset) { - for (const [bucket, ref] of Object.entries(priorRefs)) { - if (refs[bucket] || loaded.has(bucket)) continue - if (existsSync(join(dir, ref.name))) refs[bucket] = ref - } - } - providers[provider] = { - envFingerprint: section.envFingerprint, - ...(section.durable ? { durable: true } : {}), - shards: refs, + // into the bucket, so the published shard's other entries have to be + // merged back in or the save would drop them. Deferred to phase two so + // the read happens against the CURRENT shard, not a stale name. + if (loaded && !loaded.has(bucket) && prior) { plan.deferred.push(bucket); continue } + plan.refs[bucket] = await writeShard(provider, bucket, files) } } @@ -1026,21 +1092,84 @@ export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Pr return false } - // Last look before publishing: a concurrent save may have unlinked a reused - // or carried shard while this one was writing its own. An envelope must - // never name a file that is already gone — that reads back as a corrupt - // month and drops its history, including orphans no re-parse can recover. - for (const [provider, meta] of Object.entries(providers)) { - for (const [bucket, ref] of Object.entries(meta.shards)) { - if (written.includes(ref.name) || existsSync(join(dir, ref.name))) continue - if (!payloads.has(`${provider}\0${bucket}`)) { - // A carried month whose file vanished: its content was never in - // memory, so there is nothing to rewrite. Dropping the reference is - // the only honest option, and the sweep retires the name. - delete meta.shards[bucket] - continue - } - meta.shards[bucket] = await writeShard(provider, bucket) + // ── Phase two: everything that has to read the published shards ────── + // Re-read the envelope first. Between our load and now, another process may + // have republished any month we are carrying or merging into; adopting its + // CURRENT name is what keeps a carried orphan (an expired transcript's PR + // spend, unrecoverable by any re-parse) from being dropped just because the + // name we remembered was retired. It also shrinks the read-modify-write + // window for a merge down to the publish itself. That window is not zero: + // two processes merging into the same unloaded month can still interleave, + // and the loser's entries are re-derived on the next parse rather than lost + // for good — a full lock here would cost every save the contention. + const live = await readEnvelope(dir) + for (const [provider, plan] of plans) { + const liveShards = live?.providers[provider]?.shards ?? {} + const currentName = (bucket: string): string | undefined => { + const name = liveShards[bucket]?.name ?? plan.priorRefs[bucket]?.name + return name && existsSync(join(dir, name)) ? name : undefined + } + + for (const bucket of plan.deferred) { + plan.refs[bucket] = await mergeShard(provider, plan, bucket, currentName(bucket)) + plan.mergedFrom.set(bucket, currentName(bucket)) + } + + // Months this run never loaded keep their published shard. This is the + // invariant that makes a scoped load safe to save from. A month another + // process published while we held a partial view is adopted for the same + // reason: dropping it would delete history we never even saw. + if (!plan.loaded || plan.reset) continue + const carried = new Set([...Object.keys(plan.priorRefs), ...Object.keys(liveShards)]) + for (const bucket of carried) { + if (plan.refs[bucket] || plan.groups.has(bucket) || plan.loaded.has(bucket)) continue + const name = currentName(bucket) + if (!name) continue + const ref = { name, until: (liveShards[bucket] ?? plan.priorRefs[bucket])!.until } + if (plan.moved.size === 0) { plan.refs[bucket] = ref; continue } + // A path that moved into another month must not survive here too. + const onDisk = await loadShard(join(dir, name)) + if (!onDisk || !Object.keys(onDisk).some(p => plan.moved.has(p))) { plan.refs[bucket] = ref; continue } + for (const path of plan.moved) delete onDisk[path] + if (Object.keys(onDisk).length > 0) plan.refs[bucket] = await writeShard(provider, bucket, onDisk) + } + } + + // One optimistic retry: if another process republished a month we merged + // into while we were reading it, our shard was built on a superseded + // pre-image and would drop that process's entries. Redoing the merge from + // the current shard narrows the read-modify-write window from a shard read + // down to the envelope publish below. It does not close it — a save that + // loses the remaining race has its entries re-derived by the next parse + // (the reconcile sees no cache entry and re-reads the file), never silently + // dropped for good. A lock here would tax every save for a rare interleave. + const settled = await readEnvelope(dir) + for (const [provider, plan] of plans) { + for (const [bucket, mergedFrom] of plan.mergedFrom) { + const now = settled?.providers[provider]?.shards[bucket]?.name + if (!now || now === mergedFrom || !existsSync(join(dir, now))) continue + plan.refs[bucket] = await mergeShard(provider, plan, bucket, now) + } + } + + // Last look before publishing: a concurrent save may have unlinked a shard + // in the moment since. An envelope must never name a file that is already + // gone — that reads back as a corrupt month and drops its history. + const providers: Record = {} + for (const [provider, plan] of plans) { + const shards: Record = {} + for (const [bucket, ref] of Object.entries(plan.refs)) { + if (written.has(ref.name) || existsSync(join(dir, ref.name))) { shards[bucket] = ref; continue } + const files = plan.groups.get(bucket) + // A carried month whose file vanished and whose content was never in + // memory cannot be rewritten; dropping the reference is the only honest + // option, and the sweep retires the name. + if (files) shards[bucket] = await writeShard(provider, bucket, files) + } + providers[provider] = { + envFingerprint: plan.section.envFingerprint, + ...(plan.section.durable ? { durable: true } : {}), + shards, } } @@ -1074,7 +1203,7 @@ export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Pr state.shards.set(provider, meta.shards) state.fingerprints.set(provider, meta.envFingerprint) for (const [path, file] of Object.entries(cache.providers[provider]!.files)) { - state.bucketOf.set(`${provider}\0${path}`, cacheBucketMonth(file)) + state.bucketOf.set(`${provider}\0${path}`, cacheFileSpan(file).bucket) } } // Write-through: the object just published IS the freshest state, so the @@ -1266,16 +1395,22 @@ export async function cleanupOrphanedTempFiles(): Promise { if (!existsSync(dir)) return const referenced = new Set([ENVELOPE_FILE]) - let envelopeRead = false - try { - const parsed = JSON.parse(await readFile(join(dir, ENVELOPE_FILE), 'utf-8')) - if (isEnvelope(parsed)) { - for (const meta of Object.values(parsed.providers)) { - for (const ref of Object.values(meta.shards)) referenced.add(ref.name) - } - envelopeRead = true + const envelope = await readEnvelope(dir) + if (envelope) { + for (const meta of Object.values(envelope.providers)) { + for (const ref of Object.values(meta.shards)) referenced.add(ref.name) } - } catch {} + // A published v9 envelope means the re-layout completed. Its retirement of + // the old layout is a separate, unsynchronised step, so a crash in between + // leaves 100MB+ of superseded cache behind forever. Age-guarded for the + // same reason the shard sweep is: an OLD binary may still be writing there. + await unlinkIfOlderThan(join(getCodeburnCacheDir(), priorCacheFile(7)), UNREFERENCED_SHARD_MAX_AGE_MS, now) + const v8Dir = join(getCodeburnCacheDir(), PRIOR_SHARD_DIR_NAME) + try { + const s = await stat(join(v8Dir, ENVELOPE_FILE)) + if (now - s.mtimeMs > UNREFERENCED_SHARD_MAX_AGE_MS) await rm(v8Dir, { recursive: true, force: true }) + } catch {} + } try { for (const entry of await readdir(dir)) { @@ -1283,7 +1418,7 @@ export async function cleanupOrphanedTempFiles(): Promise { await unlinkIfOlderThan(join(dir, entry), TEMP_FILE_MAX_AGE_MS, now) continue } - if (!envelopeRead || referenced.has(entry)) continue + if (!envelope || referenced.has(entry)) continue await unlinkIfOlderThan(join(dir, entry), UNREFERENCED_SHARD_MAX_AGE_MS, now) } } catch {} diff --git a/tests/session-cache-shards.test.ts b/tests/session-cache-shards.test.ts index 465cc6a5..ff53c9b6 100644 --- a/tests/session-cache-shards.test.ts +++ b/tests/session-cache-shards.test.ts @@ -13,6 +13,7 @@ import { join } from 'path' import { CACHE_VERSION, cacheBucketMonth, + cacheFileSpan, computeEnvFingerprint, cleanupOrphanedTempFiles, clearLoadCacheMemo, @@ -564,3 +565,238 @@ describe('v8 -> v9 migration', () => { expect(await loadCache()).toEqual(loaded) }) }) + +// Several providers emit turns in a non-chronological order (cursor composers by +// ROWID, goose / crush / copilot by a DESC ordering). Reading the span off +// turns[0]/turns[-1] then gives `until < bucket` — an empty span, unreachable at +// every scope. +describe('out-of-order turns', () => { + const outOfOrder = () => cachedFile({ turns: [turnAt('2026-08-10T10:00:00Z', 'a'), turnAt('2026-03-04T10:00:00Z', 'b')] }) + + it('spans oldest to newest whatever order the turns arrive in', () => { + const span = cacheFileSpan(outOfOrder()) + expect(span).toEqual({ bucket: '2026-03', until: '2026-08' }) + expect(cacheBucketMonth(outOfOrder())).toBe('2026-03') + }) + + it('stays reachable at the scope of either end', async () => { + const cache: SessionCache = { + version: CACHE_VERSION, + complete: true, + providers: { claude: { envFingerprint: computeEnvFingerprint('claude'), files: { '/live/desc.jsonl': outOfOrder() } } }, + } + markCacheDirty(cache, 'claude') + await saveCache(cache) + expect((await envelope()).providers['claude']!.shards['2026-03']!.until).toBe('2026-08') + + for (const [from, to] of [['2026-03-01', '2026-03-31'], ['2026-08-01', '2026-08-31']] as const) { + clearLoadCacheMemo() + const scoped = await loadCache(monthScopeForRange(new Date(`${from}T00:00:00Z`), new Date(`${to}T23:59:59Z`))) + expect(scoped.providers['claude']!.files['/live/desc.jsonl'], `unreachable at ${from}`).toBeDefined() + } + }) +}) + +// A path must never end up in two shards at once: on a later load the two copies +// race and the stale one can win, and nothing sweeps it because the envelope +// names both. +describe('re-bucketing out of an unloaded month', () => { + const juneScope = monthScopeForRange(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-30T23:59:59Z')) + + async function seed(): Promise { + const cache: SessionCache = { + version: CACHE_VERSION, + complete: true, + providers: { + claude: { + envFingerprint: computeEnvFingerprint('claude'), + files: { + '/live/moving.jsonl': fileSpanning('2026-01-10T10:00:00Z'), + '/live/stay.jsonl': fileSpanning('2026-01-11T10:00:00Z'), + '/live/jun.jsonl': fileSpanning('2026-06-10T10:00:00Z'), + }, + }, + }, + } + markCacheDirty(cache, 'claude') + await saveCache(cache) + } + + /** Every shard's view of `path`, so a duplicate is visible directly. */ + async function copiesOf(path: string): Promise { + const dir = sessionCacheDir() + const found: string[] = [] + for (const [bucket, ref] of Object.entries((await envelope()).providers['claude']!.shards)) { + const files = JSON.parse(await readFile(join(dir, ref.name), 'utf-8')) + if (files[path]) found.push(bucket) + } + return found.sort() + } + + it('(a) drops the old copy when a re-parse moves the file to another month', async () => { + await seed() + clearLoadCacheMemo() + const scoped = await loadCache(juneScope) + expect(scoped.providers['claude']!.files['/live/moving.jsonl']).toBeUndefined() + // Re-parsed from byte 0 after a rewrite: its oldest turn is now in May. + scoped.providers['claude']!.files['/live/moving.jsonl'] = cachedFile({ turns: [turnAt('2026-05-02T10:00:00Z', 'new')], mcpInventory: ['reparsed'] }) + markCacheDirty(scoped, 'claude', '/live/moving.jsonl') + await saveCache(scoped) + + expect(await copiesOf('/live/moving.jsonl')).toEqual(['2026-05']) + clearLoadCacheMemo() + const full = await loadCache() + expect(full.providers['claude']!.files['/live/moving.jsonl']!.mcpInventory).toEqual(['reparsed']) + expect(full.providers['claude']!.files['/live/stay.jsonl']).toBeDefined() + }) + + it('(b) drops the old copy when a parse failure leaves a turn-less marker', async () => { + await seed() + clearLoadCacheMemo() + const scoped = await loadCache(juneScope) + // The #441 path: the file threw, so only a failure marker is cached. + scoped.providers['claude']!.files['/live/moving.jsonl'] = { fingerprint: { dev: 1, ino: 2, mtimeMs: 9, sizeBytes: 4 }, mcpInventory: [], turns: [], failed: true } + markCacheDirty(scoped, 'claude', '/live/moving.jsonl') + await saveCache(scoped) + + expect(await copiesOf('/live/moving.jsonl')).toEqual(['0000-00']) + clearLoadCacheMemo() + const full = await loadCache() + expect(full.providers['claude']!.files['/live/moving.jsonl']!.failed).toBe(true) + expect(full.providers['claude']!.files['/live/stay.jsonl']).toBeDefined() + }) + + it('resolves a duplicate to the freshest copy and prunes it on the next save', async () => { + await seed() + // Forge the split state directly: the same path in two shards. + const dir = sessionCacheDir() + const env = await envelope() + const janName = env.providers['claude']!.shards['2026-01']!.name + const junName = env.providers['claude']!.shards['2026-06']!.name + const jun = JSON.parse(await readFile(join(dir, junName), 'utf-8')) + jun['/live/moving.jsonl'] = cachedFile({ turns: [turnAt('2026-06-02T10:00:00Z', 'fresh')], fingerprint: { dev: 1, ino: 2, mtimeMs: 999, sizeBytes: 4 }, mcpInventory: ['fresh'] }) + await writeFile(join(dir, junName), JSON.stringify(jun)) + + clearLoadCacheMemo() + const full = await loadCache() + // Newest fingerprint wins, whichever shard finished reading first. + expect(full.providers['claude']!.files['/live/moving.jsonl']!.mcpInventory).toEqual(['fresh']) + await saveCache(full) + expect(await copiesOf('/live/moving.jsonl')).toEqual(['2026-06']) + expect(existsSync(join(dir, janName))).toBe(false) + }) +}) + +describe('carried months under a concurrent writer', () => { + it('adopts the current shard rather than dropping an orphan-bearing month', async () => { + const orphan = cachedFile({ turns: [turnAt('2026-03-10T10:00:00Z', 'm')], prLinks: ['https://github.com/o/r/pull/1'] }) + const initial: SessionCache = { + version: CACHE_VERSION, + complete: true, + providers: { + claude: { + envFingerprint: computeEnvFingerprint('claude'), + files: { '/gone/mar.jsonl': orphan, '/live/jun.jsonl': fileSpanning('2026-06-10T10:00:00Z') }, + }, + }, + } + markCacheDirty(initial, 'claude') + await saveCache(initial) + + // Process B loads June-scoped: March is carried, by the name it saw. + clearLoadCacheMemo() + const b = await loadCache(monthScopeForRange(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-30T23:59:59Z'))) + + // Process A independently republishes March, retiring the file B remembers. + clearLoadCacheMemo() + const a = await loadCache(monthScopeForRange(new Date('2026-03-01T00:00:00Z'), new Date('2026-03-31T23:59:59Z'))) + a.providers['claude']!.files['/gone/mar2.jsonl'] = cachedFile({ turns: [turnAt('2026-03-12T10:00:00Z', 'n')], prLinks: ['https://github.com/o/r/pull/2'] }) + markCacheDirty(a, 'claude', '/gone/mar2.jsonl') + await saveCache(a) + + // B saves its own unrelated June change. + b.providers['claude']!.files['/live/jun2.jsonl'] = fileSpanning('2026-06-20T10:00:00Z') + markCacheDirty(b, 'claude', '/live/jun2.jsonl') + await saveCache(b) + + clearLoadCacheMemo() + const final = await loadCache() + // March survived under A's name, with both orphans; June has both files. + expect(Object.keys(final.providers['claude']!.files).sort()) + .toEqual(['/gone/mar.jsonl', '/gone/mar2.jsonl', '/live/jun.jsonl', '/live/jun2.jsonl']) + for (const ref of Object.values((await envelope()).providers['claude']!.shards)) { + expect(existsSync(join(sessionCacheDir(), ref.name))).toBe(true) + } + }) + + // Two saves merging into the SAME unloaded month are a read-modify-write with + // no lock between them. In the product they are serialised by the warm refresh + // lock; this covers what survives when they are not. The optimistic retry in + // saveCache narrows the window to the envelope publish, and a loser's entries + // are re-derived by the next parse (the reconcile finds no cache entry and + // re-reads the file) rather than being lost for good. + it('two scoped saves merging into the same unloaded month keep the envelope sound', async () => { + const base: SessionCache = { + version: CACHE_VERSION, + complete: true, + providers: { + claude: { + envFingerprint: computeEnvFingerprint('claude'), + files: { '/live/mar.jsonl': fileSpanning('2026-03-10T10:00:00Z'), '/live/jun.jsonl': fileSpanning('2026-06-10T10:00:00Z') }, + }, + }, + } + markCacheDirty(base, 'claude') + await saveCache(base) + const juneScope = monthScopeForRange(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-30T23:59:59Z')) + + clearLoadCacheMemo() + const p1 = await loadCache(juneScope) + clearLoadCacheMemo() + const p2 = await loadCache(juneScope) + // Both re-parse a different March session neither of them loaded. + p1.providers['claude']!.files['/live/mar-a.jsonl'] = fileSpanning('2026-03-20T10:00:00Z') + markCacheDirty(p1, 'claude', '/live/mar-a.jsonl') + p2.providers['claude']!.files['/live/mar-b.jsonl'] = fileSpanning('2026-03-21T10:00:00Z') + markCacheDirty(p2, 'claude', '/live/mar-b.jsonl') + await Promise.allSettled([saveCache(p1), saveCache(p2)]) + + clearLoadCacheMemo() + const final = await loadCache() + // The envelope is internally consistent and the pre-existing March session + // survived; a read-modify-write loser is re-derived by the next parse. + for (const ref of Object.values((await envelope()).providers['claude']!.shards)) { + expect(existsSync(join(sessionCacheDir(), ref.name))).toBe(true) + } + expect(final.providers['claude']!.files['/live/mar.jsonl']).toBeDefined() + expect(final.providers['claude']!.files['/live/jun.jsonl']).toBeDefined() + const landed = ['/live/mar-a.jsonl', '/live/mar-b.jsonl'].filter(p => final.providers['claude']!.files[p]) + expect(landed.length).toBeGreaterThanOrEqual(1) + }) +}) + +describe('retiring an orphaned prior layout', () => { + it('sweeps a v8 directory and a v7 file left behind by an interrupted re-layout', async () => { + await saveCache({ version: CACHE_VERSION, complete: true, providers: { + claude: { envFingerprint: 'fp', files: { '/a.jsonl': fileSpanning('2026-05-10T10:00:00Z') } }, + } }) + const v8Dir = join(TMP_DIR, 'session-cache.v8') + await mkdir(v8Dir, { recursive: true }) + await writeFile(join(v8Dir, 'envelope.json'), JSON.stringify({ version: 8, nonce: 'n', shards: {} })) + await writeFile(join(v8Dir, 'claude.abc.json'), '{}') + const v7 = join(TMP_DIR, 'session-cache.v7.json') + await writeFile(v7, '{}') + + // Fresh: an in-flight write by an older binary must be left alone. + await cleanupOrphanedTempFiles() + expect(existsSync(v8Dir)).toBe(true) + expect(existsSync(v7)).toBe(true) + + const old = new Date(Date.now() - 90 * 60 * 1000) + await utimes(join(v8Dir, 'envelope.json'), old, old) + await utimes(v7, old, old) + await cleanupOrphanedTempFiles() + expect(existsSync(v8Dir)).toBe(false) + expect(existsSync(v7)).toBe(false) + }) +})