diff --git a/CHANGELOG.md b/CHANGELOG.md index c667da83..53330881 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ - **Menu bar apps 0.9.14 to 0.9.24 load Today again with a 0.9.25 CLI.** Homebrew and npm update the CLI on their own, and 0.9.25 left `avgCostPerSession` out of a project row whose session count is a lower bound, a key those apps require, so the popover showed "Couldn't load Today". The payload always carries it again, with `sessionCountBasis` still telling 0.9.25 and later not to show a partial average. A new test checks the payload against the keys every released menu bar app requires. Reported in #1541, thanks @AleksandrLiadov. - **Update in the menu bar app restarts it into the new version.** The app runs `codeburn menubar --force` as its own child, and `pkill`/`pgrep` skip their own ancestors unless told otherwise, so the running app was never stopped: the bundle was replaced and `open` re-activated the old copy. The CLI now finds the app among its ancestors, swaps the bundle, exits, and a detached helper stops the old copy and opens the new one. Run from a terminal, it stops the app first as before. Because the fix is in the CLI, older menu bar apps update correctly once the CLI is current. Reported in #1541, thanks @AleksandrLiadov. - **OpenCode 2.x: upgraded history is no longer lost and v2 turns price like 1.x ones.** OpenCode 2 copies 1.x sessions into new tables on first start, but the copy drops compaction summary and task-tool turns, skips messages it cannot decode, and can stop partway. CodeBurn read only the copy for any migrated session, so on one real upgraded database more than half the sessions and three quarters of the all-time OpenCode cost disappeared. A session present in both generations now reads its legacy turns first and adds only the turns written after the upgrade, so pre-upgrade figures match 1.x exactly. v2 turns also carried a `provider/model` id that failed pricing for ids like `opencode/claude-haiku-4-5` or `github-copilot/gpt-5` so they showed as $0 in reports; they now use the bare id 1.x used. OpenCode sessions re-parse once. Users on CodeBurn 0.9.24 or older see no OpenCode 2 usage at all and need 0.9.25 or later. Reported by a user. +- **The session cache is stored by day, so a today view reads only today.** Each provider keeps an index of every cached session and one file per day of activity. A narrow range reads the index, the duplicate-check keys and only the sessions it reports on, and saving an appended session rewrites one day instead of a month. Older Claude transcripts are no longer re-parsed on every narrow query. The old cache converts once on first run, a session at a time. On a heavy corpus `overview -p today` drops from about 3.1 s and 690 MB to 0.8 s and 560 MB; on a normal one from 0.54 s to 0.41 s. Figures are unchanged. - **`overview`, `status` and the menu bar load only the part of the session cache a date range can use.** Cached sessions outside the range stay on disk and are held in memory as small stubs that still take part in duplicate checks and clean-up, so figures are unchanged. Shards are written one session per line, so a month past 512 MB is saved instead of silently re-parsed on every run, and Codex project names come from a small index instead of the whole results cache. On a heavy corpus `overview -p today` drops from about 3.4 GB to 1.2 GB and runs under a 1 GB heap where it used to crash; on a normal one it drops from about 970 MB to 280 MB and gets faster. Closes #1504. - **An expired Claude Code login is recovered from the Keychain, and `codeburn quota` and the desktop app say so when it cannot be.** Both providers read the OAuth credential from `~/.claude/.credentials.json` and fall back to the macOS Keychain, which is where Claude Code keeps it on a normal Mac. After a 401 the retry re-read the file whatever the credential had come from, so a Keychain user got `Temporarily unavailable.` until Claude Code was run again, and the desktop Plans card kept showing the last percentage it had fetched. A 401 now re-reads the store the credential came from, which picks up a login Claude Code has since renewed. When nothing newer is there and the credential's `expiresAt` has passed, the row reports a terminal failure with a line telling you to run Claude Code once. The desktop app still reads the Keychain only on Check now or a forced refresh, and a background poll keeps that reconnect prompt instead of falling back to Check now. A 401 on a credential still within its life stays transient with its backoff. (#1516, thanks @ozymandiashh) - Export daily rows now bucket each call on its own day, like the history and the summary, and no longer drop a turn that has no timestamp. diff --git a/scripts/upgrade-path/run.mjs b/scripts/upgrade-path/run.mjs index bf0a7774..aef7a7ea 100644 --- a/scripts/upgrade-path/run.mjs +++ b/scripts/upgrade-path/run.mjs @@ -32,7 +32,7 @@ const WORK = process.env['UPGRADE_PATH_WORK'] || join(tmpdir(), 'codeburn upgrad // else these two are the knobs to move, and the assertions below will say so. const OLD_SESSION_CACHE = 'session-cache.v7.json' const OLD_DAILY_CACHE = 'daily-cache.v17.json' -const NEW_SESSION_CACHE_DIR = 'session-cache.v9' +const NEW_SESSION_CACHE_DIR = 'session-cache.v10' const HOME = join(WORK, 'user home') const PAYLOADS = join(WORK, 'payloads') @@ -136,24 +136,34 @@ function capture(bin, cacheDir, outDir, extra = {}) { // product, so two invocations legitimately differ there. const stripGenerated = obj => JSON.parse(JSON.stringify(obj, (k, v) => (k.startsWith('generated') || k === 'liveSessions' ? undefined : v))) -// Shards carry real stat data and are published under a random filename, so -// "identical" means identical after normalizing both away. +// Pieces, their key files and the indexes carry real stat data and are +// published under a random filename, so "identical" means identical after +// normalizing both away. An index row is +// [path, day, month, until, dev, ino, mtimeMs, size, first, last, flags, offset, length]. +// Rows are compared by path, not position: a date-ranged run reorders them +// exactly as v9 reordered its month shards, and v9's check ignored order too. function shardSnapshot(cacheDir) { const dir = join(cacheDir, NEW_SESSION_CACHE_DIR) if (!existsSync(dir)) return null const out = {} for (const name of readdirSync(dir).sort()) { - if (name === 'envelope.json') continue + if (name === 'envelope.json' || name.endsWith('.tmp')) continue const body = JSON.parse(readFileSync(join(dir, name), 'utf8')) - for (const entry of Object.values(body)) { - if (entry && typeof entry === 'object' && entry.fingerprint) { - delete entry.fingerprint.dev - delete entry.fingerprint.ino - delete entry.fingerprint.mtimeMs + if (name.startsWith('index.')) { + body.pieces = Object.keys(body.pieces).sort() + for (const row of body.rows) row.splice(4, 3) + body.rows.sort((a, b) => (a[0] < b[0] ? -1 : a[0] > b[0] ? 1 : 0)) + } else { + for (const entry of Object.values(body)) { + if (entry && typeof entry === 'object' && entry.fingerprint) { + delete entry.fingerprint.dev + delete entry.fingerprint.ino + delete entry.fingerprint.mtimeMs + } } } - // Key the bucket off the shard's provider.month prefix, dropping the nonce. - out[name.replace(/\.[0-9a-f]{16}\.json$/, '')] = sortDeep(body) + // Key by provider.day (or index.provider), dropping the nonce. + out[name.replace(/\.[0-9a-f]{16}(\.keys)?\.json$/, '$1')] = sortDeep(body) } return out } @@ -211,9 +221,26 @@ check(!existsSync(join(upgradeCache, OLD_SESSION_CACHE)), `${OLD_SESSION_CACHE} check(existsSync(join(upgradeCache, NEW_SESSION_CACHE_DIR)), `${NEW_SESSION_CACHE_DIR}/ present`) check(existsSync(join(upgradeCache, NEW_SESSION_CACHE_DIR, 'envelope.json')), `${NEW_SESSION_CACHE_DIR}/envelope.json present`) const envelope = JSON.parse(readFileSync(join(upgradeCache, NEW_SESSION_CACHE_DIR, 'envelope.json'), 'utf8')) -check(envelope.version === 9 && Object.keys(envelope.providers ?? {}).length > 0, +check(envelope.version === 10 && Object.keys(envelope.providers ?? {}).length > 0, `envelope at version ${envelope.version} with ${Object.keys(envelope.providers ?? {}).length} providers`) -check(readdirSync(join(upgradeCache, NEW_SESSION_CACHE_DIR)).some(n => n !== 'envelope.json'), 'shards published alongside the envelope') +// Every provider names an index, every index names its pieces, and every piece +// has its key file; together they hold one member per index row. +{ + const dir = join(upgradeCache, NEW_SESSION_CACHE_DIR) + let rows = 0, members = 0, missing = [] + for (const [provider, meta] of Object.entries(envelope.providers ?? {})) { + if (!meta.index || !existsSync(join(dir, meta.index))) { missing.push(`${provider} index`); continue } + const index = JSON.parse(readFileSync(join(dir, meta.index), 'utf8')) + rows += index.rows.length + for (const piece of Object.values(index.pieces)) { + const keys = piece.replace(/\.json$/, '.keys.json') + if (!existsSync(join(dir, piece)) || !existsSync(join(dir, keys))) { missing.push(piece); continue } + members += Object.keys(JSON.parse(readFileSync(join(dir, piece), 'utf8'))).length + } + } + check(missing.length === 0 && rows > 0, `indexes, pieces and key files published alongside the envelope${missing.length ? ` (missing: ${missing.join(', ')})` : ''}`) + check(rows === members, `every index row has its member: ${rows} rows, ${members} members`) +} const newDailyCache = newestDailyCacheAfter(upgradeCache, OLD_DAILY_CACHE) check(newDailyCache !== null, `${newDailyCache ?? 'current daily cache'} re-derived`) check(existsSync(join(upgradeCache, OLD_DAILY_CACHE)), `${OLD_DAILY_CACHE} kept as the carry-forward baseline`) @@ -301,7 +328,7 @@ const readExport = dir => stripGenerated(JSON.parse(readFileSync(join(PAYLOADS, check(JSON.stringify(readExport('workers-0')) === JSON.stringify(readExport('workers-3')), 'export payload identical with and without workers') check(JSON.stringify(shardSnapshot(serialCache)) === JSON.stringify(shardSnapshot(parallelCache)), - 'shard bodies identical with and without workers (fingerprint stat data and shard nonces normalized)') + 'piece, key file and index bodies identical with and without workers (fingerprint stat data and nonces normalized)') // A forced pool that never actually spawned would make the check above vacuous. const verbose = run(newBin.cmd, [...newBin.args, 'status', '--format', 'json', '--period', 'all'], { @@ -329,7 +356,7 @@ if (pending.length === 0) skip('no "parse workers=" line on a warm run; cannot c else check(pending.every(l => /0 pending files|no full parses pending/.test(l)), `nothing re-parsed on the warm run: ${pending.map(l => l.replace(/^codeburn: /, '').trim()).join(' | ')}`) -check(JSON.stringify(shardSnapshot(upgradeCache)) === JSON.stringify(beforeBodies), 'warm run left every shard body unchanged') +check(JSON.stringify(shardSnapshot(upgradeCache)) === JSON.stringify(beforeBodies), 'warm run left every piece, key file and index unchanged') check(JSON.stringify(stripGenerated(warm.menubar)) === JSON.stringify(stripGenerated(upgraded.menubar)), 'warm run reports the same payload as the run that migrated the cache') diff --git a/src/parser.ts b/src/parser.ts index 795c8bb5..a9c6e662 100644 --- a/src/parser.ts +++ b/src/parser.ts @@ -29,6 +29,7 @@ import { type SessionCache, beginColdHydration, cacheEntriesInLoadOrder, + cacheHiddenFingerprint, cacheStubs, cleanupOrphanedTempFiles, computeEnvFingerprint, @@ -42,6 +43,7 @@ import { isCacheDirty, isCacheStub, loadCache, + loadCacheHiddenKeys, loadCacheStubs, markCacheDirty, markProviderComplete, @@ -2062,6 +2064,7 @@ async function scanProjectDirs( })) } const servedStubs: string[][][] = [] + const hiddenUnchanged: string[] = [] for (const [i, { filePath, dirName, source }] of discovered.entries()) { allDiscoveredFiles.add(filePath) @@ -2071,7 +2074,13 @@ async function scanProjectDirs( const cached = section.files[filePath] const stub = cached ? undefined : stubs?.get(filePath) const action = reconcileFile(fp, cached ?? stub) - if (!readOnly && deferToBackgroundFill(filePath, fp, cached ?? stub)) { + // Cached, but in no month this range reports on, and unchanged: nothing to + // parse or serve. A changed one re-parses as if uncached, as it always did. + const hidden = cached || stub || readOnly ? undefined : cacheHiddenFingerprint(diskCache, 'claude', filePath) + if (hidden && reconcileFile(fp, { fingerprint: hidden }).action === 'unchanged') { + hiddenUnchanged.push(filePath) + continue + } else if (!readOnly && deferToBackgroundFill(filePath, fp, cached ?? stub)) { continue } else if (stub && (readOnly || action.action === 'unchanged')) { if (readOnly && action.action !== 'unchanged') readOnlyServedStale = true @@ -2134,6 +2143,11 @@ async function scanProjectDirs( } } for (const keys of servedStubs) for (const turn of keys) for (const key of turn) seenMsgIds.add(key) + // Only a whole-file parse can replay an older transcript's message ids (a + // fork or resume); an append's new region never does, so it reads no keys. + if (hiddenUnchanged.length > 0 && changedFiles.some(f => !f.append)) { + for (const keys of await loadCacheHiddenKeys(diskCache, 'claude', hiddenUnchanged)) for (const turn of keys) for (const key of turn) seenMsgIds.add(key) + } const parseProgress = createScanProgress('parsing changed claude sessions', changedFiles.length) const progressTotal = changedFiles.length diff --git a/src/session-cache.ts b/src/session-cache.ts index 8680008e..e3aaf80e 100644 --- a/src/session-cache.ts +++ b/src/session-cache.ts @@ -1,4 +1,4 @@ -import { readFile, stat, open, rename, unlink, readdir, mkdir, rm } from 'fs/promises' +import { readFile, stat, open, rename, unlink, readdir, mkdir, rm, type FileHandle } from 'fs/promises' import { existsSync, readFileSync, unlinkSync } from 'fs' import { createHash, randomBytes } from 'crypto' import { join } from 'path' @@ -226,16 +226,24 @@ export type SessionCache = { // 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 +// v10: on-disk layout only - a provider's files are stored in pieces by the UTC +// day of each file's newest call, with a per-provider index of every cached +// file (piece, fingerprint, call span) and a per-piece file of dedup keys. A +// narrow query reads the index, the keys, and only the members it reports on; +// an append rewrites one day. Turn shape unchanged, so v9 migrates losslessly. +export const CACHE_VERSION = 10 // 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}` -// The v8 shard directory, read once by the lossless v8 -> v9 re-layout. +// The v9 month shards and the v8 provider shards, each read once by a lossless +// re-layout. +const MONTH_SHARD_DIR_NAME = 'session-cache.v9' 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. +// Written LAST on every save: it names every provider's index, which names its +// pieces, 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 @@ -528,18 +536,15 @@ 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 } +// A provider's envelope entry names its index; the index names the provider's +// pieces (UTC day -> file) and holds one row per cached file, in the order a +// full load holds them. type EnvelopeProvider = { envFingerprint: string durable?: boolean complete?: boolean completeFrom?: number - /** month (`YYYY-MM`, or `0000-00` for turn-less files) -> shard */ - shards: Record + index: string } type CacheEnvelope = { version: number @@ -548,13 +553,49 @@ type CacheEnvelope = { providers: Record } +/** Where a cached file lives and what a load decides from without reading it. + * `day` names its piece: the UTC day of its newest call. `month`/`until` are + * its v9 span (cacheFileSpan): the month scope a ranged load reports on, and + * the order rows are kept in, are both still the v9 ones, so everything that + * depends on load order (orphans, the first-paint snapshot, dedup first-wins) + * sees what it saw on v9. */ +type IndexRow = { + day: string + month: string + until: string + fingerprint: FileFingerprint + /** Oldest and newest call (ms): Infinity / -Infinity when the file has none. */ + lo: number + hi: number + flags: number + /** Byte span of the member line in the piece. */ + offset: number + length: number +} + +// Needed by every range: a failure marker, a turn-less file, a spawn anchor. +const ROW_ALWAYS = 1 +const ROW_PR = 2 + +type ProviderBase = { + index: string | undefined + pieces: Record + rows: Map + /** Hash of the published index text, so an unchanged save keeps its name. */ + hash: string +} + +function emptyBase(): ProviderBase { + return { index: undefined, pieces: {}, rows: new Map(), hash: '' } +} + // 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. +// by; they form one month group that every load reads. const UNDATED_BUCKET = '0000-00' -// Sentinel inside `dirtyBuckets`: every bucket of the provider is dirty. -const ALL_BUCKETS = '*' +// The piece of a file with no call. +const UNDATED_DAY = '0000-00-00' +// Sentinel inside a provider's dirty paths: every file of the provider is dirty. +const ALL_PATHS = '*' function monthKey(timestamp: string | undefined): string | null { if (!timestamp) return null @@ -564,6 +605,10 @@ function monthKey(timestamp: string | undefined): string | null { return `${d.getUTCFullYear()}-${String(d.getUTCMonth() + 1).padStart(2, '0')}` } +function dayKey(ms: number): string { + return new Date(ms).toISOString().slice(0, 10) +} + /** 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]: @@ -588,14 +633,50 @@ export function cacheBucketMonth(file: CachedFile): string { return cacheFileSpan(file).bucket } +function keysOf(file: CachedFile): string[][] { + return file.turns.map(turn => turn.calls.map(call => call.deduplicationKey)) +} + +function rowFor(path: string, file: CachedFile): IndexRow { + let lo = Infinity + let hi = -Infinity + for (const turn of file.turns) { + for (const call of turn.calls) { + const ms = Date.parse(call.timestamp) + if (ms < lo) lo = ms + if (ms > hi) hi = ms + } + } + const span = cacheFileSpan(file) + const anchor = !!file.prLinks?.length && file.isSidechain !== true && file.turns.some(t => t.spawnToolUseIds?.length) + return { + day: hi === -Infinity ? UNDATED_DAY : dayKey(hi), + month: span.bucket, + until: span.until, + fingerprint: file.fingerprint, + lo, + hi, + flags: (file.failed || file.turns.length === 0 || anchor ? ROW_ALWAYS : 0) | (file.prLinks?.length ? ROW_PR : 0), + offset: 0, + length: 0, + } +} + +// Needed by a query over [startMs, endMs]: a call in the range, or something a +// report reads whatever the range: a spawn anchor (an in-range child folds into +// it), a failure marker or turn-less file, and a WSL path (its orphan handling +// depends on the root being online). +function fileNeeded(path: string, row: IndexRow, startMs: number, endMs: number): boolean { + if ((row.flags & ROW_ALWAYS) !== 0 || isWslUncPath(path)) return true + return row.hi === -Infinity || (row.hi >= startMs && row.lo <= endMs) +} + /** A cached file a ranged load held back because the range cannot report on * it. It keeps what reconcile, the dedup pre-seeds and replays, orphan - * handling and the save need, plus where its member line lives so it can be - * loaded on demand. Never persisted. */ + * handling and the save need; its member stays in its piece and is loaded on + * demand. Never persisted. */ export type CacheStub = { fingerprint: FileFingerprint - bucket: string - until: string hasPr: boolean /** Dedup keys per turn, in turn order. */ keys: string[][] @@ -604,33 +685,31 @@ export type CacheStub = { hi: number /** Load position, shared with full entries (see `CacheState.seq`). */ seq: number - name: string - /** Byte span of the member line in `name`; offset -1 in a single-line shard. */ - offset: number - length: number } // 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. +// a piece'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 -> paths changed since the last save (or `ALL_PATHS`). */ + dirtyPaths: Map> + /** provider -> the published index this cache was read from or last saved. */ + base: Map + /** provider -> the v9 months the load reported on; null when it read them all. */ + visible: 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 nonce of the envelope `base` came from. */ + nonce: string | undefined /** The load scope this cache was read under, for the cross-request memo. */ scope: string - /** The shard directory this cache was read from. */ + /** The piece directory this cache was read from. */ dir: string /** provider -> stubbed paths, valid only while `section` is still the provider's section. */ stubs: Map }> + /** provider -> rows outside the months the load reported on. As on v9 they + * are not part of the section; only a reconcile consults them. */ + hidden: Map }> /** Load position of every entry read from disk. An entry replaced since has * none, and sorts after all loaded ones, which is where a full load's * `files` object would hold it. */ @@ -638,8 +717,6 @@ type CacheState = { /** Every cached file with a call in this range is held in full; null when * nothing was stubbed. */ covered: { startMs: number; endMs: number } | null - /** `provider\0bucket` of loaded single-line shards, rewritten by the next save. */ - legacy: Set } const cacheStates = new WeakMap() @@ -648,50 +725,35 @@ function stateOf(cache: SessionCache): CacheState { if (!state) { state = { dirty: false, - dirtyBuckets: new Map(), - shards: new Map(), - loaded: new Map(), + dirtyPaths: new Map(), + base: new Map(), + visible: new Map(), fingerprints: new Map(), - bucketOf: new Map(), + nonce: undefined, scope: 'all', dir: sessionCacheDir(), stubs: new Map(), + hidden: new Map(), seq: new WeakMap(), covered: null, - legacy: new Set(), } cacheStates.set(cache, state) } return state } -function markBucketDirty(state: CacheState, provider: string, bucket: string): void { +function markPathDirty(state: CacheState, provider: string, path: 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)) + let paths = state.dirtyPaths.get(provider) + if (!paths) { paths = new Set(); state.dirtyPaths.set(provider, paths) } + paths.add(path) } /** 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. */ + * pieces involved. Pass `filePath` whenever the change is scoped to one cached + * file (a write, a delete or a re-parse); omitting it dirties every file. */ 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 + markPathDirty(stateOf(cache), provider, filePath ?? ALL_PATHS) } /** True when any provider section changed since the last save. */ @@ -704,11 +766,59 @@ function stubsOf(state: CacheState, cache: SessionCache, provider: string): Map< return held && held.section === cache.providers[provider] ? held.byPath : undefined } +function hiddenOf(state: CacheState, cache: SessionCache, provider: string): Map | undefined { + const held = state.hidden.get(provider) + return held && held.section === cache.providers[provider] ? held.byPath : undefined +} + /** The paths of `provider` a ranged load holds as stubs (see {@link CacheStub}). */ export function cacheStubs(cache: SessionCache, provider: string): ReadonlyMap | undefined { return stubsOf(stateOf(cache), cache, provider) } +/** The fingerprint a cached file outside the load's months was saved at. v9 + * could not see such a file at all, so a narrow query re-parsed every older + * transcript it discovered; an unchanged one needs nothing from this range. */ +export function cacheHiddenFingerprint(cache: SessionCache, provider: string, path: string): FileFingerprint | undefined { + return hiddenOf(stateOf(cache), cache, provider)?.get(path)?.fingerprint +} + +/** The dedup keys of hidden files, per turn. A transcript parsed now can replay + * their message ids (a fork or resume of an old session), and the cached copy + * must win, as it does for every loaded month. */ +export async function loadCacheHiddenKeys(cache: SessionCache, provider: string, paths: Iterable): Promise { + const state = stateOf(cache) + const hidden = hiddenOf(state, cache, provider) + const base = state.base.get(provider) + const out: string[][][] = [] + if (!hidden || !base) return out + const byPiece = new Map>() + for (const path of paths) { + const row = hidden.get(path) + if (row) pushTo(byPiece, row.day, [path, row]) + } + for (const [day, rows] of byPiece) { + const name = base.pieces[day] + if (name) for (const list of (await readKeyLists(state.dir, name, rows)).values()) out.push(list) + } + return out +} + +// A key file is written without a sync, so one lost to a crash is rebuilt +// from its piece's members. +async function readKeyLists(dir: string, name: string, rows: Array<[string, IndexRow]>): Promise> { + const keys = await readKeys(dir, name) + const missing = rows.filter(([path]) => !isKeyList(keys?.[path])) + const files = missing.length > 0 ? await readMembersAt(join(dir, name), missing) : null + const out = new Map() + for (const [path] of rows) { + const file = files?.get(path) + const list = file ? keysOf(file) : keys?.[path] + if (isKeyList(list)) out.set(path, list) + } + return out +} + /** True when `provider` has any cached entry, full or stubbed. */ export function hasCachedEntries(cache: SessionCache, provider: string): boolean { const section = cache.providers[provider] @@ -738,7 +848,7 @@ function entriesInLoadOrder(state: CacheState, section: ProviderSection, stubs: } /** Every cached entry of `provider`, full or stubbed, in the order a full load - * would hold them in `files`: loaded entries in shard order, then entries + * would hold them in `files`: loaded entries in index order, then entries * written since in the order they were written. */ export function cacheEntriesInLoadOrder(cache: SessionCache, provider: string): Array<[string, CachedFile | CacheStub]> { const section = cache.providers[provider] @@ -758,25 +868,25 @@ export async function loadCacheStubs(cache: SessionCache, provider: string, path const state = stateOf(cache) const stubs = stubsOf(state, cache, provider) const section = cache.providers[provider] - if (!stubs || !section) return - const byName = new Map>() + const base = state.base.get(provider) + if (!stubs || !section || !base) return + const byPiece = new Map>() for (const path of paths) { - const stub = stubs.get(path) - if (!stub) continue - const list = byName.get(stub.name) - if (list) list.push([path, stub]) - else byName.set(stub.name, [[path, stub]]) + const row = stubs.has(path) ? base.rows.get(path) : undefined + if (row) pushTo(byPiece, row.day, [path, row]) } - for (const [name, list] of byName) { - const members = await readMembersAt(join(state.dir, name), list.map(([path, stub]) => [path, stub.offset, stub.length])) - for (const [path, stub] of list) { + for (const [day, rows] of byPiece) { + const name = base.pieces[day] + const members = name ? await readMembersAt(join(state.dir, name), rows) : new Map() + for (const [path] of rows) { + const stub = stubs.get(path)! stubs.delete(path) const file = members.get(path) if (file) { section.files[path] = file state.seq.set(file, stub.seq) } else { - markBucketDirty(state, provider, stub.bucket) + markPathDirty(state, provider, path) } } } @@ -805,12 +915,18 @@ async function coverRange(cache: SessionCache, startMs: number, endMs: number): * source can be pruned before the next publish, so such a window must never be * held back by the resident process's coalescing. */ export function hasDirtyDurableProvider(cache: SessionCache): boolean { - for (const provider of stateOf(cache).dirtyBuckets.keys()) { + for (const provider of stateOf(cache).dirtyPaths.keys()) { if (DURABLE_PROVIDER_NAMES.has(provider) || cache.providers[provider]?.durable) return true } return false } +function pushTo(map: Map, key: K, value: V): void { + const list = map.get(key) + if (list) list.push(value) + else map.set(key, [value]) +} + // ── Env Fingerprint ──────────────────────────────────────────────────── export function computeEnvFingerprint(provider: string): string { @@ -826,7 +942,6 @@ export function computeEnvFingerprint(provider: string): string { export function emptyCache(): SessionCache { return { version: CACHE_VERSION, providers: {}, complete: false } } - /** A cache is warm only when a full scan finished against it. Empty-but-marked * (a machine with no sessions) is complete; present-but-unmarked (an interrupted * cold start, or a pre-marker cache) is NOT — it is still cold. @@ -1168,7 +1283,6 @@ export async function isCacheCurrent(cache: SessionCache): Promise { const live = await readEnvelope(cacheMemo.dir) return live?.nonce === cacheMemo.nonce } - /** 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 @@ -1202,11 +1316,6 @@ function shardInScope(bucket: string, until: string, scope: CacheLoadScope): boo 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 @@ -1217,9 +1326,7 @@ function isEnvelope(raw: unknown): raw is CacheEnvelope { 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) + return typeof e['envFingerprint'] === 'string' && typeof e['index'] === 'string' }) } @@ -1232,11 +1339,79 @@ async function readEnvelope(dir: string): Promise { } } -// A shard is written one member per line: `{`, then `"path":{...}` lines -// comma-terminated except the last, then `}`. It is still one JSON object, so -// an older build's JSON.parse reads it unchanged, and it never has to exist as -// one string, which a shard past V8's ~512MB string limit cannot. -type ShardLine = { key: string; value?: string; file?: unknown; offset: number; length: number } +type RowTuple = [string, string, string, string, number, number, number, number, number | null, number | null, number, number, number] + +function rowTuple(path: string, row: IndexRow): RowTuple { + const fp = row.fingerprint + return [path, row.day, row.month, row.until, fp.dev, fp.ino, fp.mtimeMs, fp.sizeBytes, + row.lo === Infinity ? null : row.lo, row.hi === -Infinity ? null : row.hi, row.flags, row.offset, row.length] +} + +function parseRow(t: unknown): [string, IndexRow] | null { + if (!Array.isArray(t) || t.length !== 13) return null + const [path, day, month, until, dev, ino, mtimeMs, sizeBytes, lo, hi, flags, offset, length] = t as unknown[] + if (typeof path !== 'string' || typeof day !== 'string' || typeof month !== 'string' || typeof until !== 'string') return null + if (![dev, ino, mtimeMs, sizeBytes, flags, offset, length].every(isNum)) return null + if ((lo !== null && !isNum(lo)) || (hi !== null && !isNum(hi))) return null + return [path, { + day, month, until, + fingerprint: { dev: dev as number, ino: ino as number, mtimeMs: mtimeMs as number, sizeBytes: sizeBytes as number }, + lo: lo === null ? Infinity : lo as number, + hi: hi === null ? -Infinity : hi as number, + flags: flags as number, offset: offset as number, length: length as number, + }] +} + +function indexText(pieces: Record, rows: Iterable<[string, IndexRow]>): string { + const tuples: RowTuple[] = [] + for (const [path, row] of rows) tuples.push(rowTuple(path, row)) + return JSON.stringify({ version: CACHE_VERSION, pieces, rows: tuples }) +} + +function textHash(text: string): string { + return createHash('sha256').update(text).digest('hex') +} + +// Null when missing or malformed: the provider's files then re-parse, as a lost +// v9 shard's did. +async function readIndex(dir: string, name: string | undefined): Promise { + if (!name) return null + try { + const text = await readFile(join(dir, name), 'utf-8') + const parsed = JSON.parse(text) as { version?: unknown; pieces?: unknown; rows?: unknown } + if (parsed.version !== CACHE_VERSION || !parsed.pieces || typeof parsed.pieces !== 'object' || !Array.isArray(parsed.rows)) return null + const pieces = parsed.pieces as Record + if (!Object.values(pieces).every(v => typeof v === 'string')) return null + const rows = new Map() + for (const t of parsed.rows) { + const row = parseRow(t) + if (!row) return null + rows.set(row[0], row[1]) + } + return { index: name, pieces: pieces as Record, rows, hash: textHash(text) } + } catch { + return null + } +} + +function isKeyList(v: unknown): v is string[][] { + return Array.isArray(v) && v.every(isStringArray) +} + +async function readKeys(dir: string, piece: string): Promise | null> { + try { + const parsed = JSON.parse(await readFile(join(dir, keysFileName(piece)), 'utf-8')) as unknown + return parsed && typeof parsed === 'object' && !Array.isArray(parsed) ? parsed as Record : null + } catch { + return null + } +} + +// A piece (and its keys file) is written one member per line: `{`, then +// `"path":{...}` lines comma-terminated except the last, then `}`. It is still +// one JSON object, it never has to exist as one string, and a member can be read +// back by its byte span alone. +type ShardLine = { key: string; value: string; offset: number; length: number } const SHARD_READ_CHUNK = 1 << 20 @@ -1259,8 +1434,7 @@ function shardLine(text: string, offset: number, length: number): ShardLine { return { key: split[0], value: split[1], offset, length } } -// A shard's members one at a time, without holding the file. A single-line -// shard an older build wrote is parsed whole (offset -1). Throws on anything +// A piece's members one at a time, without holding the file. Throws on anything // malformed, as JSON.parse would. async function* shardLines(path: string): AsyncGenerator { const handle = await open(path, 'r') @@ -1268,10 +1442,8 @@ async function* shardLines(path: string): AsyncGenerator { const head = Buffer.alloc(2) await handle.read(head, 0, 2, 0) if (head[0] !== 0x7b || head[1] !== 0x0a) { - const files = JSON.parse(await readFile(path, 'utf-8')) as unknown - if (!files || typeof files !== 'object' || Array.isArray(files)) throw new Error('shard is not an object') - for (const [key, file] of Object.entries(files)) yield { key, file, offset: -1, length: 0 } - return + if (JSON.stringify(JSON.parse(await readFile(path, 'utf-8'))) === '{}') return + throw new Error('not a line-format shard') } // One reused buffer, and a line that spans reads is carried as decoded text: // per-line Buffers are off-heap memory only a GC returns, which shows up as @@ -1304,16 +1476,71 @@ async function* shardLines(path: string): AsyncGenerator { } } -function shardLineFile(line: ShardLine): unknown { - return line.file ?? JSON.parse(line.value!) +// The members of any JSON object file as [key, value text], a chunk at a time: +// the one-member-per-line form and a single-line object main wrote alike, so a +// v9 shard is never one string (or one parse) during the re-layout. +async function* objectMembers(path: string): AsyncGenerator<[string, string]> { + const handle = await open(path, 'r') + try { + const chunk = Buffer.allocUnsafe(SHARD_READ_CHUNK) + const decoder = new StringDecoder('utf-8') + let depth = 0 + let inString = false + let escaped = false + let member = '' + let pos = 0 + const split = (text: string): [string, string] => { + const pair = splitMemberLine(text.trim()) + if (!pair) throw new Error('malformed shard member') + return pair + } + for (;;) { + const { bytesRead } = await handle.read(chunk, 0, SHARD_READ_CHUNK, pos) + if (bytesRead === 0) break + pos += bytesRead + const text = decoder.write(chunk.subarray(0, bytesRead)) + let start = 0 + for (let i = 0; i < text.length; i++) { + const c = text.charCodeAt(i) + if (inString) { + if (escaped) escaped = false + else if (c === 92) escaped = true + else if (c === 34) inString = false + } else if (c === 34) { + inString = true + } else if (c === 123 || c === 91) { + if (depth++ === 0) start = i + 1 + } else if (c === 125 || c === 93) { + if (--depth === 0) { + const last = member + text.slice(start, i) + if (last.trim()) yield split(last) + return + } + } else if (c === 44 && depth === 1) { + const done = member + text.slice(start, i) + member = '' + start = i + 1 + yield split(done) + } + } + if (depth > 0) member += text.slice(start) + } + throw new Error('truncated shard') + } finally { + await handle.close() + } } function shardLineText(line: ShardLine): string { - return `${JSON.stringify(line.key)}:${line.value ?? JSON.stringify(line.file)}` + return `${JSON.stringify(line.key)}:${line.value}` +} + +function memberText(path: string, value: unknown): string { + return `${JSON.stringify(path)}:${JSON.stringify(value)}` } // The writer's side of the format above. `positions`, when given, receives each -// member's byte span, which is what lets a stub be loaded without the shard. +// member's byte span. async function* shardPayload(members: AsyncIterable | Iterable, positions?: Array<[number, number]>): AsyncGenerator { let pos = 0 let first = true @@ -1330,40 +1557,46 @@ async function* shardPayload(members: AsyncIterable | Iterable, yield first ? '{}' : '\n}' } -function* memberTexts(files: Record): Generator { - for (const [path, file] of Object.entries(files)) yield `${JSON.stringify(path)}:${JSON.stringify(file)}` -} - -// 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. +// A piece that is missing or malformed costs exactly the files it held: those +// re-parse while every other piece keeps serving. async function loadShard(path: string): Promise | null> { try { const files: Record = {} - for await (const line of shardLines(path)) files[line.key] = shardLineFile(line) + for await (const line of shardLines(path)) files[line.key] = JSON.parse(line.value) return validateFiles(files) ? files : null } catch { return null } } -async function readMembersAt(path: string, wanted: Array<[string, number, number]>): Promise> { +// The members of `rows` from one piece. Neighbouring members are read together, +// up to SHARD_READ_CHUNK a read, into one reused buffer: Buffers are off-heap +// memory only a GC returns, which shows up as RSS when members run to megabytes. +async function readMembersAt(path: string, rows: Array<[string, IndexRow]>, scratch: { buf: Buffer } = { buf: Buffer.allocUnsafe(0) }): Promise> { const out = new Map() try { - if (wanted.some(([, offset]) => offset < 0)) { - const files = await loadShard(path) - for (const [key] of wanted) if (files?.[key]) out.set(key, files[key]!) - return out - } const handle = await open(path, 'r') try { - for (const [key, offset, length] of wanted) { - const buf = Buffer.allocUnsafe(length) - const { bytesRead } = await handle.read(buf, 0, length, offset) - const split = splitMemberLine(buf.toString('utf-8', 0, bytesRead)) - if (!split || split[0] !== key) continue - const file: unknown = JSON.parse(split[1]) - if (validateCachedFile(file)) out.set(key, file) + const sorted = [...rows].sort((a, b) => a[1].offset - b[1].offset) + for (let i = 0; i < sorted.length;) { + const from = sorted[i]![1].offset + let j = i + 1 + while (j < sorted.length && sorted[j]![1].offset + sorted[j]![1].length - from <= SHARD_READ_CHUNK) j++ + const to = sorted[j - 1]![1].offset + sorted[j - 1]![1].length + if (scratch.buf.length < to - from) scratch.buf = Buffer.allocUnsafe(to - from) + const buf = scratch.buf + const { bytesRead } = await handle.read(buf, 0, to - from, from) + for (const [key, row] of sorted.slice(i, j)) { + const start = row.offset - from + if (start + row.length > bytesRead) continue + const split = splitMemberLine(buf.toString('utf-8', start, start + row.length)) + if (!split || split[0] !== key) continue + try { + const file: unknown = JSON.parse(split[1]) + if (validateCachedFile(file)) out.set(key, file) + } catch { /* a malformed member re-parses */ } + } + i = j } } finally { await handle.close() @@ -1372,60 +1605,6 @@ async function readMembersAt(path: string, wanted: Array<[string, number, number return out } -// Needed by a query over [startMs, endMs]: a call in the range, or something a -// report reads whatever the range: a spawn anchor (an in-range child folds into -// it), a failure marker or turn-less file, and a WSL path (its orphan handling -// depends on the root being online). `lo`/`hi` span the file's call timestamps. -function fileNeeded(path: string, file: CachedFile, lo: number, hi: number, startMs: number, endMs: number): boolean { - if (file.failed || file.turns.length === 0 || isWslUncPath(path)) return true - if (file.prLinks?.length && file.isSidechain !== true && file.turns.some(t => t.spawnToolUseIds?.length)) return true - return hi === -Infinity || (hi >= startMs && lo <= endMs) -} - -// One in-scope shard of a ranged load: members the range needs in full, the -// rest as stubs. Null when any member is malformed, as loadShard would be. -async function readScopedShard(dir: string, name: string, bucket: string, startMs: number, endMs: number): Promise<{ members: Array<[string, CachedFile | CacheStub]>; legacy: boolean } | null> { - const members: Array<[string, CachedFile | CacheStub]> = [] - let legacy = false - try { - for await (const line of shardLines(join(dir, name))) { - if (line.offset < 0) legacy = true - const file = shardLineFile(line) - if (!validateCachedFile(file)) return null - members.push([line.key, scopedMember(line.key, file, bucket, name, line.offset, line.length, startMs, endMs)]) - } - return { members, legacy } - } catch { - return null - } -} - -function scopedMember(path: string, file: CachedFile, bucket: string, name: string, offset: number, length: number, startMs: number, endMs: number): CachedFile | CacheStub { - let lo = Infinity - let hi = -Infinity - for (const turn of file.turns) { - for (const call of turn.calls) { - const ms = Date.parse(call.timestamp) - if (ms < lo) lo = ms - if (ms > hi) hi = ms - } - } - if (fileNeeded(path, file, lo, hi, startMs, endMs)) return file - return { - fingerprint: file.fingerprint, - bucket, - until: cacheFileSpan(file).until, - hasPr: (file.prLinks?.length ?? 0) > 0, - keys: file.turns.map(turn => turn.calls.map(call => call.deduplicationKey)), - lo, - hi, - seq: 0, - name, - offset, - length, - } -} - // Shards a resident process (codeburn serve) keeps parsed between requests, // keyed by shard FILE NAME. A name carries a fresh nonce on every write // (shardFileName), so a name that is still published names the same bytes and @@ -1500,14 +1679,18 @@ export async function loadShardMemoized(dir: string, name: string): Promise { if (process.env['CODEBURN_CACHE_SCOPE'] === 'all') scope = undefined const dir = sessionCacheDir() - const envelope = await readEnvelope(dir) + let envelope = await readEnvelope(dir) + if (!envelope) { + // A process that loses the re-layout race reads the winner's publish. + await migrateMonthShardCache(dir) + 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 @@ -1532,6 +1720,7 @@ export async function loadCache(scope?: CacheLoadScope): Promise { const cache: SessionCache = { version: CACHE_VERSION, providers: {}, complete: envelope.complete === true } const state = stateOf(cache) state.dir = dir + state.nonce = envelope.nonce const ranged = scope?.startMs !== undefined && scope.endMs !== undefined ? { startMs: scope.startMs, endMs: scope.endMs } : null @@ -1544,80 +1733,21 @@ export async function loadCache(scope?: CacheLoadScope): Promise { ...(meta.complete === true ? { complete: true } : {}), ...(typeof meta.completeFrom === 'number' ? { completeFrom: meta.completeFrom } : {}), } - // 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. + // Recorded even when every file is skipped or unreadable: the section is + // what tells the next save which provider the carried rows belong to, and + // what stops the reconcile from re-parsing under a fingerprint the envelope + // already agrees with. cache.providers[provider] = section + state.fingerprints.set(provider, meta.envFingerprint) // Durable providers are ALWAYS loaded in full, by name as well as by the // envelope flag: copilot's serve-time reconciliation pairs store rows and // retires residuals over the complete cached serve set, so a scoped load // of a copilot section persisted before the durable stamp landed would // make pairing range-dependent. The name check closes that window. const full = !scope || meta.durable === true || DURABLE_PROVIDER_NAMES.has(provider) || meta.envFingerprint !== computeEnvFingerprint(provider) - const loaded: Set | null = full ? null : new Set() - const stubs = loaded && ranged && !FULL_LOAD_PROVIDER_NAMES.has(provider) ? new Map() : undefined - if (stubs) { - state.stubs.set(provider, { section, byPath: stubs }) - state.covered = ranged - } - // 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. - // A ranged load streams its shards one after another instead, so it never - // holds more than one shard's needed members beyond what it keeps. - const pending: { bucket: string; name: string; files: Promise | null> | null }[] = [] - for (const [bucket, ref] of Object.entries(meta.shards)) { - if (loaded && !shardInScope(bucket, ref.until, scope!)) continue - loaded?.add(bucket) - pending.push({ bucket, name: ref.name, files: stubs ? null : loadShardMemoized(dir, ref.name) }) - } - reads.push((async () => { - let seq = 0 - for (const { bucket, name, files: read } of pending) { - let members: Array<[string, CachedFile | CacheStub]> | null = null - if (read) { - const files = await read - if (files) members = Object.entries(files) - } else { - const scoped = await readScopedShard(dir, name, bucket, ranged!.startMs, ranged!.endMs) - if (scoped?.legacy) state.legacy.add(`${provider}\0${bucket}`) - members = scoped?.members ?? null - } - // 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 (!members) { markBucketDirty(state, provider, bucket); continue } - for (const [path, entry] of members) { - const key = `${provider}\0${path}` - const seenIn = state.bucketOf.get(key) - let at = seq++ - if (seenIn !== undefined) { - markBucketDirty(state, provider, seenIn) - markBucketDirty(state, provider, bucket) - const prior = section.files[path] ?? stubs!.get(path)! - if (prior.fingerprint.mtimeMs >= entry.fingerprint.mtimeMs) continue - at = isCacheStub(prior) ? prior.seq : state.seq.get(prior) ?? at - if (isCacheStub(prior) !== isCacheStub(entry)) { - delete section.files[path] - stubs?.delete(path) - } - } - state.bucketOf.set(key, bucket) - if (isCacheStub(entry)) { - entry.seq = at - stubs!.set(path, entry) - } else { - if (stubs) state.seq.set(entry, at) - section.files[path] = entry - } - } - } - })()) - state.loaded.set(provider, loaded) - state.shards.set(provider, meta.shards) - state.fingerprints.set(provider, meta.envFingerprint) + const stubbed = !full && ranged !== null && !FULL_LOAD_PROVIDER_NAMES.has(provider) + if (stubbed) state.covered = ranged + reads.push(loadProvider(state, provider, section, meta.index, full ? null : scope!, stubbed ? ranged : null)) } await Promise.all(reads) state.scope = scopeKey @@ -1625,26 +1755,227 @@ export async function loadCache(scope?: CacheLoadScope): Promise { return cache } -// 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. +// The months of `rows` a query over `scope` reports on: a v9 shard's span was +// its oldest file's month to its newest file's `until`. +function visibleMonths(rows: Map, scope: CacheLoadScope): Set { + const until = new Map() + for (const row of rows.values()) { + const seen = until.get(row.month) + if (seen === undefined || row.until > seen) until.set(row.month, row.until) + } + const visible = new Set([UNDATED_BUCKET]) + for (const [month, last] of until) if (shardInScope(month, last, scope)) visible.add(month) + return visible +} + +async function loadProvider(state: CacheState, provider: string, section: ProviderSection, indexName: string, scope: CacheLoadScope | null, ranged: { startMs: number; endMs: number } | null): Promise { + const index = await readIndex(state.dir, indexName) + const base = index ?? emptyBase() + state.base.set(provider, base) + // Unreadable: its files re-parse, and the save publishes a fresh index. + if (!index) markPathDirty(state, provider, ALL_PATHS) + const visible = scope ? visibleMonths(base.rows, scope) : null + state.visible.set(provider, visible) + const stubs = ranged ? new Map() : undefined + if (stubs) state.stubs.set(provider, { section, byPath: stubs }) + const hidden = new Map() + state.hidden.set(provider, { section, byPath: hidden }) + + const inFull = new Map>() + const asStubs = new Map>() + for (const [path, row] of base.rows) { + if (visible && !visible.has(row.month)) hidden.set(path, row) + else if (!ranged || fileNeeded(path, row, ranged.startMs, ranged.endMs)) pushTo(inFull, row.day, [path, row]) + else pushTo(asStubs, row.day, [path, row]) + } + const members = new Map() + // A member or key list that cannot be read is dirtied without an entry: the + // save drops its line and the file re-parses like any uncached one. + // A full read takes every piece at once, as v9 read its shards; a ranged one + // goes piece by piece, so it holds one read buffer at a time. + const whole = ranged ? [] : [...inFull.keys()].map(day => base.pieces[day] ? loadShardMemoized(state.dir, base.pieces[day]!) : null) + const scratch = { buf: Buffer.allocUnsafe(0) } + for (const [i, [day, rows]] of [...inFull].entries()) { + const name = base.pieces[day] + let files: Map | Record | null = null + if (!ranged) files = await whole[i] ?? null + else if (name) files = await readMembersAt(join(state.dir, name), rows, scratch) + for (const [path] of rows) { + const file = files instanceof Map ? files.get(path) : files?.[path] + if (file) members.set(path, file) + else markPathDirty(state, provider, path) + } + } + for (const [day, rows] of asStubs) { + const lists = base.pieces[day] ? await readKeyLists(state.dir, base.pieces[day]!, rows) : null + for (const [path, row] of rows) { + const list = lists?.get(path) + if (!list) { markPathDirty(state, provider, path); continue } + members.set(path, { fingerprint: row.fingerprint, hasPr: (row.flags & ROW_PR) !== 0, keys: list, lo: row.lo, hi: row.hi, seq: 0 }) + } + } + let seq = 0 + for (const path of base.rows.keys()) { + const entry = members.get(path) + const at = seq++ + if (!entry) continue + if (isCacheStub(entry)) { + entry.seq = at + stubs!.set(path, entry) + } else { + if (stubs) state.seq.set(entry, at) + section.files[path] = entry + } + } +} + +// The piece directory is absent/unreadable. Prefer a LOSSLESS re-layout of the +// newest prior layout that is present (v9 is handled by loadCache itself, then +// v8 provider shards, then the v7 single file — all 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 piece directory +// is minted on the next save. async function afterMissingShardCache(): Promise { 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 - // file is ignored (left intact). We copy it into the shard layout once via + // file is ignored (left intact). We copy it into the piece layout once via // saveCache; the legacy file is never modified. return adoptLegacyCache() } -// 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. +// Migration appends each member to its day piece as it arrives. Only this many +// pieces are held open at once; a piece closed early is reopened for append. +const MIGRATION_OPEN_PIECES = 32 + +type PieceSink = { name: string; pos: number; count: number } + +// One-time, lossless re-layout of the v9 month shards into day pieces. It +// streams: every shard is read a member at a time (a single-line shard main +// wrote included) and each member is appended, verbatim, to its piece, so the +// peak is one member plus the open pieces, never the cache. Rows keep v9's full +// load order. Publishing the v10 envelope commits it; the v9 directory is +// removed only after that. +async function migrateMonthShardCache(dir: string): Promise { + const v9 = join(getCodeburnCacheDir(), MONTH_SHARD_DIR_NAME) + let prior: { complete?: unknown; providers: Record> } + try { + const parsed = JSON.parse(await readFile(join(v9, ENVELOPE_FILE), 'utf-8')) as Record + if (parsed['version'] !== 9 || !parsed['providers'] || typeof parsed['providers'] !== 'object') return false + prior = parsed as typeof prior + } catch { + return false + } + const written: string[] = [] + const open_ = new Map() + const closeOne = async (day: string): Promise => { + const handles = open_.get(day) + open_.delete(day) + for (const handle of handles ?? []) { await handle.sync(); await handle.close() } + } + try { + if (!existsSync(dir)) await mkdir(dir, { recursive: true, mode: 0o700 }) + const providers: Record = {} + for (const [provider, meta] of Object.entries(prior.providers)) { + const shards = meta?.['shards'] + if (typeof meta?.['envFingerprint'] !== 'string' || !shards || typeof shards !== 'object') continue + const rows = new Map() + const sinks = new Map() + const stale = new Set() + const append = async (day: string, key: string, value: string, keys: string[][]): Promise<[number, number]> => { + let sink = sinks.get(day) + if (!sink) { + sink = { name: pieceFileName(provider, day), pos: 0, count: 0 } + sinks.set(day, sink) + written.push(sink.name, keysFileName(sink.name)) + } + let handles = open_.get(day) + if (handles) { + open_.delete(day) + } else { + if (open_.size >= MIGRATION_OPEN_PIECES) await closeOne(open_.keys().next().value!) + handles = [await open(join(dir, sink.name), 'a', 0o600), await open(join(dir, keysFileName(sink.name)), 'a', 0o600)] + } + open_.set(day, handles) + const head = sink.count === 0 ? '{\n' : ',\n' + const line = `${JSON.stringify(key)}:${value}` + await handles[0]!.write(head + line) + await handles[1]!.write(head + memberText(key, keys)) + const offset = sink.pos + 2 + const length = Buffer.byteLength(line) + sink.pos = offset + length + sink.count++ + return [offset, length] + } + for (const ref of Object.values(shards as Record)) { + const name = (ref as { name?: unknown } | null)?.name + if (typeof name !== 'string') continue + try { + for await (const [key, value] of objectMembers(join(v9, name))) { + const file: unknown = JSON.parse(value) + if (!validateCachedFile(file)) continue + // A path in two shards resolves to the freshest copy, at the first + // copy's place, as a v9 load did. + const seen = rows.get(key) + if (seen && seen.fingerprint.mtimeMs >= file.fingerprint.mtimeMs) continue + if (seen) stale.add(seen.day) + const row = rowFor(key, file) + const [offset, length] = await append(row.day, key, value, keysOf(file)) + rows.set(key, { ...row, offset, length }) + } + } catch { /* an unreadable shard costs the files it still held, as on v9 */ } + } + for (const [day, sink] of sinks) { + let handles = open_.get(day) + if (!handles) handles = [await open(join(dir, sink.name), 'a', 0o600), await open(join(dir, keysFileName(sink.name)), 'a', 0o600)] + open_.set(day, handles) + await handles[0]!.write('\n}') + await handles[1]!.write('\n}') + await closeOne(day) + } + // A piece still holding a duplicate's stale line is rewritten without it. + for (const day of stale) { + const sink = sinks.get(day)! + const dayRows = [...rows].filter(([, row]) => row.day === day) + if (dayRows.length === 0) sinks.delete(day) + const piece = await writePiece(dir, provider, day, sink.name, dayRows, {}, new Map(), written) + for (const [path, row] of dayRows) { + const span = piece.spans.get(path) + if (span) rows.set(path, { ...row, offset: span[0], length: span[1] }) + } + for (const name of [sink.name, keysFileName(sink.name)]) await retryCacheFileMutation(() => unlink(join(dir, name))) + sink.name = piece.name + } + const index = indexFileName(provider) + await writeFileAtomic(join(dir, index), indexText(Object.fromEntries([...sinks].map(([day, sink]) => [day, sink.name])), rows)) + written.push(index) + providers[provider] = { + envFingerprint: meta['envFingerprint'] as string, + ...(meta['durable'] === true ? { durable: true } : {}), + ...(meta['complete'] === true ? { complete: true } : {}), + ...(isNum(meta['completeFrom']) ? { completeFrom: meta['completeFrom'] } : {}), + index, + } + await yieldToEventLoop() + } + if (await readEnvelope(dir)) throw new Error('another process published first') + const envelope: CacheEnvelope = { version: CACHE_VERSION, complete: prior.complete === true, nonce: randomBytes(8).toString('hex'), providers } + await writeFileAtomic(join(dir, ENVELOPE_FILE), JSON.stringify(envelope)) + } catch { + for (const day of [...open_.keys()]) await closeOne(day).catch(() => {}) + for (const name of written) await retryCacheFileMutation(() => unlink(join(dir, name))) + return false + } + await retryCacheFileMutation(() => rm(v9, { recursive: true, force: true })) + return true +} + +// One-time, lossless re-layout of the v8 per-provider shard directory: v8's +// turn shape is the current one, so every entry moves across verbatim and +// nothing re-parses. The v8 directory is removed only once the save has +// published. async function migrateProviderShardCache(): Promise { const dir = join(getCodeburnCacheDir(), PRIOR_SHARD_DIR_NAME) let envelope: { complete?: boolean; shards: Record } @@ -1666,8 +1997,7 @@ async function migrateProviderShardCache(): Promise { } // 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. +// shard directory, so it is migrated straight to the current layout. async function migrateSingleFileCache(): Promise { const v7Path = join(getCodeburnCacheDir(), priorCacheFile(7)) let parsed: unknown @@ -1683,8 +2013,8 @@ async function migrateSingleFileCache(): Promise { ) } -// Every section is marked dirty so the save writes each month's shard; the old -// layout is retired only once that save has published. +// Every section is marked dirty so the save writes every piece; 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) @@ -1705,18 +2035,30 @@ async function adoptLegacyCache(): Promise { } } -// Shard filenames carry a fresh nonce on every write, so a save never overwrites -// the file the currently-published envelope points at: readers keep seeing a +// File names carry a fresh nonce on every write, so a save never overwrites a +// 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, bucket: string): string { - return `${provider.replace(/[^A-Za-z0-9_-]/g, '_')}.${bucket}.${randomBytes(8).toString('hex')}.json` +// that loses the ownership fence leaves the canonical files untouched. +function providerFilePart(provider: string): string { + return provider.replace(/[^A-Za-z0-9_-]/g, '_') +} + +function pieceFileName(provider: string, day: string): string { + return `${providerFilePart(provider)}.${day}.${randomBytes(8).toString('hex')}.json` +} + +function keysFileName(piece: string): string { + return piece.replace(/\.json$/, '.keys.json') +} + +function indexFileName(provider: string): string { + return `index.${providerFilePart(provider)}.${randomBytes(8).toString('hex')}.json` } // The temp name carries a nonce: two processes writing the SAME final path // (the envelope, every save) would otherwise share one temp file and interleave // their writes into a torn or foreign payload. -async function writeFileAtomic(finalPath: string, payload: string | AsyncIterable): Promise { +async function writeFileAtomic(finalPath: string, payload: string | AsyncIterable, sync = true): Promise { const tempPath = `${finalPath}.${randomBytes(8).toString('hex')}.tmp` const handle = await open(tempPath, 'w', 0o600) try { @@ -1732,7 +2074,7 @@ async function writeFileAtomic(finalPath: string, payload: string | AsyncIterabl } if (pending) await handle.writeFile(pending, { encoding: 'utf-8' }) } - await handle.sync() + if (sync) await handle.sync() } finally { await handle.close() } @@ -1753,372 +2095,296 @@ async function writeFileAtomic(finalPath: string, payload: string | AsyncIterabl } } -// A stub opens its bucket's group (at the position a full load would) but -// never joins it: its member stays on disk until the bucket is rewritten. -function bucketFiles(entries: Array<[string, CachedFile | CacheStub]>): { groups: Map>; until: Map } { - const groups = new Map>() - const until = new Map() - for (const [path, file] of entries) { - if (isCacheStub(file)) { - if (!groups.has(file.bucket)) groups.set(file.bucket, {}) - continue - } - 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 { groups, until } -} - -function untilMonth(files: Record): string { - let until = UNDATED_BUCKET - for (const file of Object.values(files)) { - 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 - stubs: Map | undefined - /** Buckets holding a stub, which memory alone cannot rewrite. */ - partial: Set - /** Stub lines a rewrite moved, applied once the envelope publishes. */ - relocated: Array<[CacheStub, string, number, number]> - /** Partial buckets rewritten without their stubs' lines (the source shard was gone). */ - lost: Set -} - // Surrender the event loop without microtask overhead. Used inside saveCache -// between shard writes so an interactive TTY's stdin handler (Ink's useInput) +// between piece writes so an interactive TTY's stdin handler (Ink's useInput) // can run while a long save publishes a 21k-file cache (#1141). function yieldToEventLoop(): Promise { return new Promise(resolve => setImmediate(resolve)) } +// The order a v9 save would leave these files in, which is the order the next +// full load holds them in (see IndexRow). v9 kept files in month shards: a +// month the load read is rewritten from memory in load order, a month it did +// not read that gained a file (a re-parse) keeps its published order with the +// new file after it, behind every month the load read, and the months it +// neither read nor touched follow as they were. +function nextOrder(base: Map, rows: Map, loadOrder: string[], visible: Set | null): string[] { + const read = (month: string): boolean => visible === null || visible.has(month) + const before = new Map() + for (const [path, row] of base) pushTo(before, row.month, path) + const now = new Map() + for (const path of loadOrder) { + const row = rows.get(path) + if (row) pushTo(now, row.month, path) + } + const out: string[] = [] + const placed = new Set() + const put = (path: string): void => { + if (placed.has(path) || !rows.has(path)) return + placed.add(path) + out.push(path) + } + const stayed = (month: string) => (path: string): void => { if (rows.get(path)?.month === month) put(path) } + for (const [month, paths] of now) if (!before.has(month) || read(month)) paths.forEach(put) + for (const [month, paths] of now) { + if (!before.has(month) || read(month)) continue + before.get(month)!.forEach(stayed(month)) + paths.forEach(put) + } + for (const [month, paths] of before) if (!now.has(month) && !read(month)) paths.forEach(stayed(month)) + // Rows this process never held (a concurrent writer's), then anything left. + for (const path of base.keys()) put(path) + for (const path of rows.keys()) put(path) + return out +} + +// Rewrite one piece: the lines of rows that stay are copied from the published +// piece, the rest come from memory. A piece whose lines all come out the same +// keeps its published name (#1032). Returns each written row's byte span; a +// row the published piece no longer holds has none. +async function writePiece(dir: string, provider: string, day: string, from: string | undefined, rows: Array<[string, IndexRow]>, files: Record, fromMemory: Map, written: string[]): Promise<{ name: string; spans: Map }> { + const copied = new Map() + const pending = new Map() + for (const [path, row] of rows) { + if (fromMemory.has(path)) pending.set(path, files[path]!) + else copied.set(path, row) + } + const baseKeys = from && copied.size > 0 ? await readKeys(dir, from) : null + const order: string[] = [] + const keyLines: string[] = [] + let changed = !from + const members = async function* (): AsyncGenerator { + if (from) { + try { + for await (const line of shardLines(join(dir, from))) { + const file = pending.get(line.key) + const row = copied.get(line.key) + if (file) { + pending.delete(line.key) + const text = memberText(line.key, file) + if (text !== shardLineText(line)) changed = true + order.push(line.key) + keyLines.push(memberText(line.key, keysOf(file))) + yield text + } else if (row && row.offset === line.offset) { + copied.delete(line.key) + const keys = baseKeys?.[line.key] + order.push(line.key) + keyLines.push(memberText(line.key, isKeyList(keys) ? keys : keysOf(JSON.parse(line.value) as CachedFile))) + yield shardLineText(line) + } else { + changed = true + } + } + } catch { + changed = true + } + } + for (const [path, file] of pending) { + changed = true + order.push(path) + keyLines.push(memberText(path, keysOf(file))) + yield memberText(path, file) + } + } + const name = pieceFileName(provider, day) + const positions: Array<[number, number]> = [] + await writeFileAtomic(join(dir, name), shardPayload(members(), positions)) + const spans = new Map(order.map((path, i): [string, [number, number]] => [path, positions[i]!])) + if (!changed && copied.size === 0) { + await retryCacheFileMutation(() => unlink(join(dir, name))) + return { name: from!, spans } + } + written.push(name) + await writeFileAtomic(join(dir, keysFileName(name)), shardPayload(keyLines), false) + written.push(keysFileName(name)) + return { name, spans } +} + +type ProviderSave = ProviderBase & { retire: string[]; lost: Set } + +// One provider's part of a save: rows changed in memory (dirty paths) over the +// rows of `base`, the pieces those changes touch rewritten, the rest kept. +async function saveProvider(dir: string, state: CacheState, cache: SessionCache, provider: string, section: ProviderSection, base: ProviderBase, reset: boolean, written: string[]): Promise { + const dirty = state.dirtyPaths.get(provider) + const all = dirty?.has(ALL_PATHS) === true + const isDirty = (path: string): boolean => all || dirty?.has(path) === true + const stubs = stubsOf(state, cache, provider) + const hidden = hiddenOf(state, cache, provider) + const priorRows = reset ? new Map() : base.rows + const rows = new Map() + const fromMemory = new Map() + const touched = new Set() + for (const [path, row] of priorRows) { + if (!isDirty(path) || (!section.files[path] && (stubs?.has(path) || hidden?.has(path)))) rows.set(path, row) + else if (!section.files[path]) touched.add(row.day) + } + for (const [path, file] of Object.entries(section.files)) { + const prior = priorRows.get(path) + if (prior && !isDirty(path)) continue + const row = rowFor(path, file) + rows.set(path, row) + fromMemory.set(path, row) + touched.add(row.day) + if (prior) touched.add(prior.day) + } + + const order = nextOrder(priorRows, rows, entriesInLoadOrder(state, section, stubs).map(([path]) => path), reset ? null : state.visible.get(provider) ?? null) + const byDay = new Map>() + for (const path of order) pushTo(byDay, rows.get(path)!.day, [path, rows.get(path)!]) + + const priorPieces = reset ? {} : base.pieces + const pieces: Record = {} + const retire: string[] = [] + const lost = new Set() + const spansOf = new Map() + for (const [day, name] of Object.entries(base.pieces)) if (reset || !byDay.has(day)) retire.push(name) + for (const [day, dayRows] of byDay) { + const from = priorPieces[day] + if (from && !touched.has(day)) { pieces[day] = from; continue } + const piece = await writePiece(dir, provider, day, from, dayRows, section.files, fromMemory, written) + pieces[day] = piece.name + if (from && piece.name !== from) retire.push(from) + for (const [path] of dayRows) { + const span = piece.spans.get(path) + if (span) spansOf.set(path, span) + else lost.add(path) + } + await yieldToEventLoop() + } + const finalRows = new Map() + for (const path of order) { + if (lost.has(path)) continue + const row = rows.get(path)! + const span = spansOf.get(path) + finalRows.set(path, span && (span[0] !== row.offset || span[1] !== row.length) ? { ...row, offset: span[0], length: span[1] } : row) + } + if (lost.size > 0) { + const held = new Set([...finalRows.values()].map(row => row.day)) + for (const day of byDay.keys()) if (!held.has(day)) { retire.push(pieces[day]!); delete pieces[day] } + } + + const text = indexText(pieces, finalRows) + const hash = textHash(text) + let index = base.index + if (reset || !index || hash !== base.hash) { + index = indexFileName(provider) + await writeFileAtomic(join(dir, index), text) + written.push(index) + if (base.index) retire.push(base.index) + } + return { index, pieces, rows: finalRows, hash, retire, lost } +} + 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 written = new Set() - const plans = new Map() - const writeShard = async (provider: string, bucket: string, files: Record): Promise => { - const name = shardFileName(provider, bucket) - await writeFileAtomic(join(dir, name), shardPayload(memberTexts(files))) - written.add(name) - return { name, until: untilMonth(files) } - } + for (let attempt = 0; ; attempt++) { + const written: string[] = [] + try { + // Another process may have published since this cache was read. Its + // envelope is then the base: what this process did not change stays as + // that process left it, and what it did change is applied on top. + const live = await readEnvelope(dir) + const rebased = live !== null && live.nonce !== state.nonce + const saves = new Map() + for (const [provider, section] of Object.entries(cache.providers)) { + const base = rebased + ? await readIndex(dir, live!.providers[provider]?.index) ?? emptyBase() + : state.base.get(provider) ?? emptyBase() + const baseFingerprint = rebased ? live!.providers[provider]?.envFingerprint : state.fingerprints.get(provider) + // A fingerprint change discards the section outright (see + // getOrCreateProviderSection), so nothing is carried from the base. + const reset = baseFingerprint !== undefined && baseFingerprint !== section.envFingerprint + saves.set(provider, await saveProvider(dir, state, cache, provider, section, base, reset, written)) + } - // A bucket holding stubs is written as a full load would write it, members in - // load order and entries written since after them, but a stub's member is - // copied from the shard it was loaded from instead of rebuilt from memory. - // Unchanged, the published shard is kept (#1032). - const writePartialShard = async (provider: string, plan: ProviderPlan, bucket: string, from: string | undefined): Promise => { - const files = plan.groups.get(bucket)! - if (!from || !existsSync(join(dir, from))) { - plan.lost.add(bucket) - return writeShard(provider, bucket, files) - } - const stubs = plan.stubs! - let changed = state.legacy.has(`${provider}\0${bucket}`) - let until = UNDATED_BUCKET - const copied: Array<[CacheStub, number]> = [] - const members = async function* (): AsyncGenerator { - const done = new Set() - let index = 0 - for await (const line of shardLines(join(dir, from))) { - const file = files[line.key] - const stub = file ? undefined : stubs.get(line.key) - if (file && state.seq.has(file)) { - const text = `${JSON.stringify(line.key)}:${JSON.stringify(file)}` - if (text !== shardLineText(line)) changed = true - done.add(line.key) - const month = cacheFileSpan(file).until - if (month > until) until = month - index++ - yield text - } else if (stub && stub.bucket === bucket) { - copied.push([stub, index++]) - if (stub.until > until) until = stub.until - yield shardLineText(line) - } else { - // Replaced, moved or deleted since the load. - changed = true + // One optimistic retry when a publish landed while this one was being + // built: redone from that publish. The remaining window is the envelope + // rename below; a save that loses it has its changes re-derived by the + // next parse (the reconcile finds no row and re-reads the file). + if (attempt === 0 && (await readEnvelope(dir))?.nonce !== live?.nonce) { + for (const name of written) await retryCacheFileMutation(() => unlink(join(dir, name))) + continue + } + + // 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. Files written above are + // unreferenced until the envelope names them, so a lost fence publishes + // nothing. + if (verifyStillOwner && !await verifyStillOwner()) { + for (const name of written) await retryCacheFileMutation(() => unlink(join(dir, name))) + return false + } + + const providers: Record = {} + for (const [provider, save] of saves) { + const section = cache.providers[provider]! + providers[provider] = { + envFingerprint: section.envFingerprint, + ...(section.durable ? { durable: true } : {}), + ...(section.complete === true ? { complete: true } : {}), + ...(section.completeFrom !== undefined ? { completeFrom: section.completeFrom } : {}), + index: save.index!, } } - for (const [path, file] of Object.entries(files)) { - if (done.has(path)) continue - changed = true - const month = cacheFileSpan(file).until - if (month > until) until = month - yield `${JSON.stringify(path)}:${JSON.stringify(file)}` + const envelope: CacheEnvelope = { + version: CACHE_VERSION, + complete: cache.complete === true, + nonce: randomBytes(8).toString('hex'), + providers, } - } - const name = shardFileName(provider, bucket) - const positions: Array<[number, number]> = [] - await writeFileAtomic(join(dir, name), shardPayload(members(), positions)) - if (!changed) { - await retryCacheFileMutation(() => unlink(join(dir, name))) - return plan.priorRefs[bucket] ?? { name: from, until } - } - written.add(name) - for (const [stub, index] of copied) plan.relocated.push([stub, name, positions[index]![0], positions[index]![1]]) - return { name, until } - } + await writeFileAtomic(join(dir, ENVELOPE_FILE), JSON.stringify(envelope)) - // 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) - // A file whose month this run never loaded has no visible cache entry, so it - // looks uncached and is re-parsed into the same bucket — re-deriving the - // entry the shard already holds. Republishing then churns the shard's nonce - // name on every run for content that never changed (#1032), so a merge that - // neither adds, changes nor removes an entry keeps the published shard. - const adds = Object.entries(files).some(([path, file]) => - onDisk[path] === undefined || JSON.stringify(onDisk[path]) !== JSON.stringify(file)) - const removes = [...plan.moved].some(path => onDisk[path] !== undefined && files[path] === undefined) - if (!adds && !removes) return { name: from!, until: untilMonth(onDisk) } - for (const path of plan.moved) delete onDisk[path] - return writeShard(provider, bucket, { ...onDisk, ...files }) - } + // Files 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. + const retired = new Set() + for (const save of saves.values()) for (const name of save.retire) retired.add(name) + for (const [provider, base] of state.base) { + if (saves.has(provider)) continue + if (base.index) retired.add(base.index) + for (const name of Object.values(base.pieces)) retired.add(name) + } - 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 - // 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 stubs = stubsOf(state, cache, provider) - const { groups } = bucketFiles(entriesInLoadOrder(state, section, stubs)) - const partial = new Set() - for (const stub of stubs?.values() ?? []) partial.add(stub.bucket) - const plan: ProviderPlan = { section, groups, loaded, priorRefs, reset, moved: new Set(), deferred: [], mergedFrom: new Map(), refs: {}, stubs, partial, relocated: [], lost: new Set() } - 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) + state.dirty = false + state.dirtyPaths.clear() + state.base.clear() + state.fingerprints.clear() + state.nonce = envelope.nonce + for (const [provider, save] of saves) { + state.base.set(provider, { index: save.index, pieces: save.pieces, rows: save.rows, hash: save.hash }) + state.fingerprints.set(provider, envelope.providers[provider]!.envFingerprint) + const section = cache.providers[provider]! + const hidden = hiddenOf(state, cache, provider) + const stubs = stubsOf(state, cache, provider) + for (const path of save.lost) { hidden?.delete(path); stubs?.delete(path) } + if (!stubs) continue + // Load order continues as it stood in memory, as it does for v9's + // in-memory `files` after a save. + let seq = 0 + for (const [, entry] of entriesInLoadOrder(state, section, stubs)) { + if (isCacheStub(entry)) entry.seq = seq++ + else state.seq.set(entry, seq++) } } - - 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. - const legacy = state.legacy.has(`${provider}\0${bucket}`) - if (prior && !isBucketDirty(state, provider, bucket) && !legacy && existsSync(join(dir, prior.name))) { - plan.refs[bucket] = prior - continue - } - // Dirty but never loaded: memory holds only the entries this run wrote - // 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] = partial.has(bucket) - ? await writePartialShard(provider, plan, bucket, prior?.name) - : await writeShard(provider, bucket, files) - // Surrender the event loop between shard writes so an interactive - // TTY's stdin handler (Ink's useInput) can run while a long save - // publishes a 21k-file cache. The yield is BETWEEN shards - never - // inside one - so the temp+rename atomicity of writeFileAtomic is - // preserved and the in-process `written` set still tracks every - // file this save owns for the lost-fence cleanup below (#1141). - await yieldToEventLoop() + // Write-through: the object just published IS the freshest state, so the + // next loadCache in this process reuses it instead of re-reading. 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))) + if (!name.startsWith('index.')) await retryCacheFileMutation(() => unlink(join(dir, keysFileName(name)))) } - } - - // 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. Shards written above are - // unreferenced until the envelope names them, so a lost fence publishes - // nothing. - if (verifyStillOwner && !await verifyStillOwner()) { + return true + } catch (err) { for (const name of written) await retryCacheFileMutation(() => unlink(join(dir, name))) - return false + throw err } - - // ── 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) continue - if (plan.partial.has(bucket)) plan.lost.add(bucket) - shards[bucket] = await writeShard(provider, bucket, files) - } - providers[provider] = { - envFingerprint: plan.section.envFingerprint, - ...(plan.section.durable ? { durable: true } : {}), - ...(plan.section.complete === true ? { complete: true } : {}), - ...(plan.section.completeFrom !== undefined ? { completeFrom: plan.section.completeFrom } : {}), - shards, - } - } - - const envelope: CacheEnvelope = { - version: CACHE_VERSION, - complete: cache.complete === true, - nonce: randomBytes(8).toString('hex'), - providers, - } - await writeFileAtomic(join(dir, ENVELOPE_FILE), JSON.stringify(envelope)) - - // 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. - 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. - state.legacy.clear() - 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}`, cacheFileSpan(file).bucket) - } - const plan = plans.get(provider)! - if (!plan.stubs) continue - for (const [stub, name, offset, length] of plan.relocated) Object.assign(stub, { name, offset, length }) - for (const [path, stub] of plan.stubs) { - if (plan.lost.has(stub.bucket)) plan.stubs.delete(path) - else state.bucketOf.set(`${provider}\0${path}`, stub.bucket) - } - // What was just written is the new load order. - let seq = 0 - for (const [, entry] of entriesInLoadOrder(state, plan.section, plan.stubs)) { - if (isCacheStub(entry)) entry.seq = seq++ - else state.seq.set(entry, seq++) - } - } - // 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))) - throw err } } @@ -2338,21 +2604,30 @@ export async function cleanupOrphanedTempFiles(): Promise { if (!existsSync(dir)) return const referenced = new Set([ENVELOPE_FILE]) - const envelope = await readEnvelope(dir) + let 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) + referenced.add(meta.index) + const memo = cacheMemo?.nonce === envelope.nonce ? stateOf(cacheMemo.cache).base : undefined + const pieces = [...(memo?.values() ?? [])].find(base => base.index === meta.index)?.pieces + ?? (await readIndex(dir, meta.index))?.pieces + // An index that cannot be read (a concurrent save may just have retired + // it) leaves its pieces unknown, so nothing is swept this time. + if (!pieces) { envelope = null; break } + for (const name of Object.values(pieces)) { referenced.add(name); referenced.add(keysFileName(name)) } } - // A published v9 envelope means the re-layout completed. Its retirement of + // A published 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. + // same reason the piece 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 {} + for (const name of [PRIOR_SHARD_DIR_NAME, MONTH_SHARD_DIR_NAME]) { + const old = join(getCodeburnCacheDir(), name) + try { + const s = await stat(join(old, ENVELOPE_FILE)) + if (now - s.mtimeMs > UNREFERENCED_SHARD_MAX_AGE_MS) await rm(old, { recursive: true, force: true }) + } catch {} + } } try { diff --git a/tests/bounded-load-parse.test.ts b/tests/bounded-load-parse.test.ts index 9b9868a6..cad10b1e 100644 --- a/tests/bounded-load-parse.test.ts +++ b/tests/bounded-load-parse.test.ts @@ -113,16 +113,17 @@ describe('bounded load through the provider parse', () => { expect(JSON.stringify(boundedSnapshot)).toBe(JSON.stringify(await snapshot('all'))) }) - it('evicts a deleted transcript that was only a stub from its shard on disk', async () => { + it('evicts a deleted transcript that was only a stub from its piece on disk', async () => { const doomed = rollout('04-10', 'aadoomed', '2099-04-10T10:00:00.000Z', 2) - const kept = rollout('04-12', 'cckept', '2099-04-12T10:00:00.000Z', 2) + const kept = rollout('04-10', 'cckept', '2099-04-10T12:00:00.000Z', 2) rollout('05-01', 'bbtoday', '2099-05-01T10:00:00.000Z', 2) await parseAllSessions(undefined, 'codex') - const aprilShard = async (): Promise => { + const aprilPiece = async (): Promise => { const envelope = JSON.parse(await readFile(join(sessionCacheDir(), 'envelope.json'), 'utf-8')) - return readFile(join(sessionCacheDir(), envelope.providers.codex.shards['2099-04'].name), 'utf-8') + const index = JSON.parse(await readFile(join(sessionCacheDir(), envelope.providers.codex.index), 'utf-8')) + return readFile(join(sessionCacheDir(), index.pieces['2099-04-10']), 'utf-8') } - const before = await aprilShard() + const before = await aprilPiece() const line = (text: string, path: string) => text.split('\n').find(l => l.startsWith(JSON.stringify(path) + ':'))?.replace(/,$/, '') expect(line(before, doomed)).toBeDefined() @@ -132,7 +133,7 @@ describe('bounded load through the provider parse', () => { rmSync(doomed) await reparse(may1) - const after = await aprilShard() + const after = await aprilPiece() expect(after).not.toContain(doomed) expect(line(after, kept)).toBe(line(before, kept)) expect(Object.keys(JSON.parse(after))).toEqual([kept]) diff --git a/tests/copilot-aiu-serve.test.ts b/tests/copilot-aiu-serve.test.ts index 5d502a3f..1ddd301d 100644 --- a/tests/copilot-aiu-serve.test.ts +++ b/tests/copilot-aiu-serve.test.ts @@ -78,7 +78,7 @@ describe.skipIf(!isSqliteAvailable())('cachedCallToApiCall threads nanoAiu', () }) it('serves nanoAiu on a cold parse and a warm cache read without a CACHE_VERSION bump', async () => { - expect(CACHE_VERSION).toBe(9) + expect(CACHE_VERSION).toBe(10) clearSessionCache() const cold = await parseAllSessions(undefined, 'copilot') @@ -88,6 +88,6 @@ describe.skipIf(!isSqliteAvailable())('cachedCallToApiCall threads nanoAiu', () clearSessionCache() const warm = await parseAllSessions(undefined, 'copilot') expect(collectNanoAiu(warm)).toEqual([1_500_000_000]) - expect(CACHE_VERSION).toBe(9) + expect(CACHE_VERSION).toBe(10) }) }) diff --git a/tests/fixtures/session-cache-io.ts b/tests/fixtures/session-cache-io.ts index 445a769b..b7c2767e 100644 --- a/tests/fixtures/session-cache-io.ts +++ b/tests/fixtures/session-cache-io.ts @@ -10,6 +10,7 @@ import { markCacheDirty, saveCache, sessionCacheDir, + type CachedFile, type SessionCache, } from '../../src/session-cache.js' @@ -33,3 +34,20 @@ export async function cacheDirSnapshot(): Promise { const parts = await Promise.all(names.map(async name => `${name}:${await readFile(join(dir, name), 'utf-8')}`)) return parts.join('\n') } + +/** The published pieces of `provider` (UTC day -> file name), straight off disk. */ +export async function publishedPieces(provider: string): Promise> { + const dir = sessionCacheDir() + const envelope = JSON.parse(await readFile(join(dir, 'envelope.json'), 'utf-8')) as { providers: Record } + const index = envelope.providers[provider]?.index + if (!index) return {} + return (JSON.parse(await readFile(join(dir, index), 'utf-8')) as { pieces: Record }).pieces +} + +/** Every member the published pieces of `provider` hold, straight off disk. */ +export async function publishedMembers(provider: string): Promise> { + const dir = sessionCacheDir() + const files: Record = {} + for (const name of Object.values(await publishedPieces(provider))) Object.assign(files, JSON.parse(await readFile(join(dir, name), 'utf-8'))) + return files +} diff --git a/tests/incremental-sweep.test.ts b/tests/incremental-sweep.test.ts index b720c2fc..e7cb8a44 100644 --- a/tests/incremental-sweep.test.ts +++ b/tests/incremental-sweep.test.ts @@ -2,7 +2,7 @@ // watcher proved untouched, and coalesces shard publication. Both are freshness // trades, so what is pinned here is when they are NOT taken. import { describe, it, expect, beforeEach, afterEach } from 'vitest' -import { mkdtemp, mkdir, readFile, writeFile, rm, rename } from 'fs/promises' +import { mkdtemp, mkdir, writeFile, rm, rename } from 'fs/promises' import { join } from 'path' import { tmpdir } from 'os' @@ -14,7 +14,8 @@ import { setShardPublishCoalescing, setSweepWatchSource, } from '../src/parser.js' -import { clearLoadCacheMemo, fingerprintFileCount, loadCache, markCacheDirty, saveCache, sessionCacheDir } from '../src/session-cache.js' +import { clearLoadCacheMemo, fingerprintFileCount, loadCache, markCacheDirty, saveCache } from '../src/session-cache.js' +import { publishedMembers } from './fixtures/session-cache-io.js' let tmpDir: string @@ -210,15 +211,7 @@ describe('incremental discovery sweep', () => { describe('coalesced shard publication', () => { // Read the published shards straight off disk: the usual test reader drops the // load memo, and dropping it is exactly what retires a held publish. - const publishedFiles = async (provider = 'claude'): Promise> => { - const dir = sessionCacheDir() - const envelope = JSON.parse(await readFile(join(dir, 'envelope.json'), 'utf-8')) as - { providers: Record }> } - const names = Object.values(envelope.providers[provider]?.shards ?? {}).map(s => s.name) - const files: Record = {} - for (const name of names) Object.assign(files, JSON.parse(await readFile(join(dir, name), 'utf-8'))) - return files - } + const publishedFiles = (provider = 'claude'): Promise> => publishedMembers(provider) const claudeFiles = async (): Promise => Object.keys(await publishedFiles()) it('publishes the first time, holds the next, and flushes on shutdown', async () => { diff --git a/tests/parse-workers.test.ts b/tests/parse-workers.test.ts index 33e7f270..1afccc0a 100644 --- a/tests/parse-workers.test.ts +++ b/tests/parse-workers.test.ts @@ -220,14 +220,15 @@ async function writeForkedCodexPair(codexHome: string, day: string, tag: string, for (const [name, body] of order) await writeCodexRollout(codexHome, day, name, body) } -/// Cache shard file names carry a random nonce, so compare bodies keyed by -/// `.` instead of by file name. +/// Cache piece file names carry a random nonce, so compare bodies keyed by +/// `.` instead of by file name. An index names pieces by their +/// nonce, so it is compared through the pieces it names. async function shardBodies(cacheDir: string): Promise> { - const dir = join(cacheDir, 'session-cache.v9') + const dir = join(cacheDir, 'session-cache.v10') const out: Record = {} for (const name of (await readdir(dir).catch(() => []))) { - if (name === 'envelope.json' || !name.endsWith('.json')) continue - const key = name.split('.').slice(0, 2).join('.') + if (name === 'envelope.json' || name.startsWith('index.') || !name.endsWith('.json')) continue + const key = name.split('.').slice(0, 2).join('.') + (name.endsWith('.keys.json') ? '.keys' : '') out[key] = createHash('sha256').update(await readFile(join(dir, name))).digest('hex') } return out diff --git a/tests/parser.test.ts b/tests/parser.test.ts index 64c1c59e..664ca53d 100644 --- a/tests/parser.test.ts +++ b/tests/parser.test.ts @@ -8,7 +8,7 @@ // discovered source stays, flagged or not import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest' -import { mkdtemp, mkdir, writeFile, rm, unlink } from 'fs/promises' +import { mkdtemp, mkdir, readFile, writeFile, rm, unlink } from 'fs/promises' import { tmpdir } from 'os' import { join } from 'path' import { createRequire } from 'node:module' @@ -18,8 +18,8 @@ import { calculateCost } from '../src/models.js' import { aggregateProjectsIntoDays } from '../src/day-aggregator.js' import { DAILY_CACHE_VERSION, currentTzKey, ensureCacheHydrated, saveDailyCache } from '../src/daily-cache.js' import { clearSessionCache, isSessionHydrationComplete, parseAllSessions, setParseReuseValidator } from '../src/parser.js' -import { CACHE_VERSION, clearLoadCacheMemo, computeEnvFingerprint, loadCache, PROVIDER_PARSE_VERSIONS, saveCache } from '../src/session-cache.js' -import { cacheDirSnapshot, readCacheOnDisk, writeCacheOnDisk } from './fixtures/session-cache-io.js' +import { CACHE_VERSION, clearLoadCacheMemo, computeEnvFingerprint, loadCache, PROVIDER_PARSE_VERSIONS, saveCache, sessionCacheDir } from '../src/session-cache.js' +import { publishedPieces, readCacheOnDisk, writeCacheOnDisk } from './fixtures/session-cache-io.js' import type { SessionSource, SessionParser, ParsedProviderCall } from '../src/providers/types.js' import { setHome } from './setup/home.js' @@ -3555,20 +3555,19 @@ describe.skipIf(!isSqliteAvailable())('(sc) month-sharded cache integration for // retainWhilePresent: >90d rows serve and the entry survives its save. expect(sumInput(await parseAllSessions(undefined, 'copilot'))).toBe(200) - // Appending an OLDER row moves the file's oldest-turn month, which - // re-buckets the shard. The path must live in exactly ONE shard file — - // a stale copy in the old bucket would resurrect dropped data on a - // partial load. cacheDirSnapshot exposes raw shard bytes; the merged + // Appending an OLDER row moves the file's oldest-turn month. The path must + // live in exactly ONE piece file — a stale copy would resurrect dropped + // data on a partial load. The raw piece bytes show it; the merged // loadCache view (which collapses duplicates) cannot see this. insertStoreRow(dbPath, 'sess-ret', 150, 0, 0, at(130)) clearSessionCache() clearLoadCacheMemo() expect(sumInput(await parseAllSessions(undefined, 'copilot'))).toBe(350) - const snapshot = await cacheDirSnapshot() - const shardSections = snapshot.split('\n').filter(s => s.length > 0 && !s.startsWith('envelope.json:')) - // The shards are JSON, so a Windows path appears with its separators escaped. - const holding = shardSections.filter(s => s.includes(JSON.stringify(dbPath).slice(1, -1))) + const pieces = Object.values(await publishedPieces('copilot')) + const bodies = await Promise.all(pieces.map(name => readFile(join(sessionCacheDir(), name), 'utf-8'))) + // The pieces are JSON, so a Windows path appears with its separators escaped. + const holding = bodies.filter(s => s.includes(JSON.stringify(dbPath).slice(1, -1))) expect(holding).toHaveLength(1) const disk = await readCacheOnDisk() expect(disk.providers['copilot']!.files[dbPath]!.turns.flatMap(t => t.calls)).toHaveLength(3) diff --git a/tests/providers/vercel-gateway.test.ts b/tests/providers/vercel-gateway.test.ts index 3a7a0eee..4e11eb85 100644 --- a/tests/providers/vercel-gateway.test.ts +++ b/tests/providers/vercel-gateway.test.ts @@ -130,7 +130,7 @@ describe('vercel-gateway end-to-end (parseAllSessions network path)', () => { // halves of that round-trip — the total above is the read half, the stored // record is the write half. const { readdirSync, readFileSync } = await import('node:fs') - const cacheRoot = join(cacheDir, 'session-cache.v9') + const cacheRoot = join(cacheDir, 'session-cache.v10') const stored = readdirSync(cacheRoot) .map(f => readFileSync(join(cacheRoot, f), 'utf-8')) .join('') diff --git a/tests/session-cache-bounded-load.test.ts b/tests/session-cache-bounded-load.test.ts index 23f04125..0536176a 100644 --- a/tests/session-cache-bounded-load.test.ts +++ b/tests/session-cache-bounded-load.test.ts @@ -1,8 +1,7 @@ // A ranged load holds the cached files its range cannot report on as stubs. -// These pin what that must never cost: a save rewrites a bucket holding stubs -// without losing or rebuilding their members, a single-line shard from an -// older build converts on the next save, and a wider request in the same -// process loads the members it now needs. +// These pin what that must never cost: a save leaves the pieces of stubs +// alone, a single-line v9 shard re-lays out into pieces, and a wider request +// in the same process loads the members it now needs. import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { mkdir, readFile, rm, writeFile } from 'fs/promises' import { existsSync } from 'fs' @@ -70,22 +69,22 @@ async function publish(files: Record): Promise { clearLoadCacheMemo() } -async function mayShard(): Promise<{ name: string; text: string }> { - const envelope = JSON.parse(await readFile(join(sessionCacheDir(), 'envelope.json'), 'utf-8')) - const name = envelope.providers.claude.shards['2026-05'].name as string - return { name, text: await readFile(join(sessionCacheDir(), name), 'utf-8') } -} - -function memberLine(text: string, path: string): string | undefined { - return text.split('\n').find(line => line.startsWith(JSON.stringify(path) + ':'))?.replace(/,$/, '') +async function pieceOf(day: string): Promise<{ name: string; text: string }> { + const dir = sessionCacheDir() + const envelope = JSON.parse(await readFile(join(dir, 'envelope.json'), 'utf-8')) + const index = JSON.parse(await readFile(join(dir, envelope.providers.claude.index), 'utf-8')) + const name = index.pieces[`2026-05-${day}`] as string + return { name, text: await readFile(join(dir, name), 'utf-8') } } describe('bounded ranged load', () => { - it('rewrites a bucket holding stubs by copying their lines, in full-load order', async () => { + it('rewrites only the pieces that changed, and keeps full-load order', async () => { await publish({ '/p/x.jsonl': fileAt('20', 'x'), '/p/y.jsonl': fileAt('02', 'y'), '/p/w.jsonl': fileAt('03', 'w') }) - const before = await mayShard() - expect(before.text.startsWith('{\n')).toBe(true) - expect(JSON.parse(before.text)).toEqual({ '/p/x.jsonl': fileAt('20', 'x'), '/p/y.jsonl': fileAt('02', 'y'), '/p/w.jsonl': fileAt('03', 'w') }) + const y = await pieceOf('02') + const w = await pieceOf('03') + const x = await pieceOf('20') + expect(x.text.startsWith('{\n')).toBe(true) + expect(JSON.parse(x.text)).toEqual({ '/p/x.jsonl': fileAt('20', 'x') }) const cache = await loadCache(day20) const section = cache.providers['claude']! @@ -100,53 +99,46 @@ describe('bounded ranged load', () => { markCacheDirty(cache, 'claude', '/p/z.jsonl') expect(await saveCache(cache)).toBe(true) - const after = await mayShard() - expect(after.name).not.toBe(before.name) - expect(memberLine(after.text, '/p/y.jsonl')).toBe(memberLine(before.text, '/p/y.jsonl')) - expect(memberLine(after.text, '/p/w.jsonl')).toBe(memberLine(before.text, '/p/w.jsonl')) - const parsed = JSON.parse(after.text) as Record - // A full load would hold the re-parsed file after the loaded ones. - expect(Object.keys(parsed)).toEqual(['/p/y.jsonl', '/p/w.jsonl', '/p/x.jsonl', '/p/z.jsonl']) - expect(parsed['/p/x.jsonl']).toEqual(fileAt('20', 'x', 1)) - expect(parsed['/p/y.jsonl']).toEqual(fileAt('02', 'y')) + // The stubs' pieces are not touched at all. + expect(await pieceOf('02')).toEqual(y) + expect(await pieceOf('03')).toEqual(w) + const after = await pieceOf('20') + expect(after.name).not.toBe(x.name) + expect(JSON.parse(after.text)).toEqual({ '/p/x.jsonl': fileAt('20', 'x', 1) }) - // Dirty but unchanged: the published shard is kept. + // Dirty but unchanged: the published piece is kept. markCacheDirty(cache, 'claude', '/p/x.jsonl') expect(await saveCache(cache)).toBe(true) - expect((await mayShard()).name).toBe(after.name) + expect((await pieceOf('20')).name).toBe(after.name) + // A full load would hold the re-parsed file after the loaded ones. clearLoadCacheMemo() const full = await loadCache() - expect(full.providers['claude']!.files).toEqual(parsed) + expect(Object.keys(full.providers['claude']!.files)).toEqual(['/p/y.jsonl', '/p/w.jsonl', '/p/x.jsonl', '/p/z.jsonl']) + expect(full.providers['claude']!.files['/p/x.jsonl']).toEqual(fileAt('20', 'x', 1)) }) - it('converts a single-line shard from an older build on the next save and still loads stubs from it', async () => { + it('re-lays a single-line v9 shard out and still loads stubs from its pieces', async () => { const files = { '/p/x.jsonl': fileAt('20', 'x'), '/p/y.jsonl': fileAt('02', 'y') } - const dir = sessionCacheDir() - await mkdir(dir, { recursive: true }) - await writeFile(join(dir, 'claude.2026-05.0123456789abcdef.json'), JSON.stringify(files)) - await writeFile(join(dir, 'envelope.json'), JSON.stringify({ - version: CACHE_VERSION, complete: true, nonce: 'legacy', + const v9 = join(TMP_DIR, 'session-cache.v9') + await mkdir(v9, { recursive: true }) + await writeFile(join(v9, 'claude.2026-05.0123456789abcdef.json'), JSON.stringify(files)) + await writeFile(join(v9, 'envelope.json'), JSON.stringify({ + version: 9, complete: true, nonce: 'legacy', providers: { claude: { envFingerprint: computeEnvFingerprint('claude'), complete: true, shards: { '2026-05': { name: 'claude.2026-05.0123456789abcdef.json', until: '2026-05' } } } }, })) const cache = await loadCache(day20) + expect(existsSync(v9)).toBe(false) + expect(Object.keys(cache.providers['claude']!.files)).toEqual(['/p/x.jsonl']) expect([...cacheStubs(cache, 'claude')!.keys()]).toEqual(['/p/y.jsonl']) - // Any save rewrites it, even one whose only change is elsewhere. - cache.providers['codex'] = { envFingerprint: computeEnvFingerprint('codex'), files: { '/c/r.jsonl': fileAt('20', 'r') } } - markCacheDirty(cache, 'codex', '/c/r.jsonl') - expect(await saveCache(cache)).toBe(true) + expect(JSON.parse((await pieceOf('02')).text)).toEqual({ '/p/y.jsonl': fileAt('02', 'y') }) - const after = await mayShard() - expect(after.name).not.toBe('claude.2026-05.0123456789abcdef.json') - expect(after.text.startsWith('{\n')).toBe(true) - expect(JSON.parse(after.text)).toEqual(files) - - // A wider request in this process loads the stub from its new line. + // A wider request in this process loads the stub from its piece. const wider = await loadCache(monthScopeForRange(new Date('2026-05-01T00:00:00.000Z'), new Date('2026-05-20T23:59:59.999Z'))) expect(wider).toBe(cache) expect(cacheStubs(cache, 'claude')!.size).toBe(0) - expect(cache.providers['claude']!.files['/p/y.jsonl']).toEqual(fileAt('02', 'y')) + expect(cache.providers['claude']!.files).toEqual(files) }) it('reads the same members as a full load for every range', async () => { diff --git a/tests/session-cache-day-pieces.test.ts b/tests/session-cache-day-pieces.test.ts new file mode 100644 index 00000000..4827a8c1 --- /dev/null +++ b/tests/session-cache-day-pieces.test.ts @@ -0,0 +1,267 @@ +// Day pieces (CACHE_VERSION 10): a ranged load reads the index, the key files, +// and only the members the range reports on; an append that crosses a day +// moves the file between pieces; v9 month shards re-lay out losslessly; and a +// narrow query no longer re-parses cached transcripts outside its months. +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { mkdir, readFile, readdir, rm, writeFile } from 'fs/promises' +import { existsSync } from 'fs' +import { tmpdir } from 'os' +import { join } from 'path' + +import { + CACHE_VERSION, + cacheStubs, + clearLoadCacheMemo, + computeEnvFingerprint, + loadCache, + loadCacheStubs, + markCacheDirty, + monthScopeForRange, + saveCache, + sessionCacheDir, + type CachedFile, + type SessionCache, +} from '../src/session-cache.js' +import { clearSessionCache, parseAllSessions } from '../src/parser.js' + +let TMP_DIR: string + +beforeEach(async () => { + TMP_DIR = join(tmpdir(), `codeburn-pieces-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`) + process.env['CODEBURN_CACHE_DIR'] = TMP_DIR + await mkdir(TMP_DIR, { recursive: true }) + clearLoadCacheMemo() +}) + +afterEach(async () => { + clearLoadCacheMemo() + if (existsSync(TMP_DIR)) await rm(TMP_DIR, { recursive: true }) +}) + +function fileAt(days: string[], key: string): CachedFile { + return { + fingerprint: { dev: 1, ino: key.length, mtimeMs: 3, sizeBytes: 4 }, + lastCompleteLineOffset: 9, + mcpInventory: [], + turns: days.map((day, i) => { + const timestamp = `2026-05-${day}T10:00:00.000Z` + return { + timestamp, + sessionId: key, + userMessage: 'go', + calls: [{ + provider: 'claude', + model: 'claude-sonnet-4-20250514', + usage: { inputTokens: 10, outputTokens: 5, cacheCreationInputTokens: 0, cacheReadInputTokens: 0, cachedInputTokens: 0, reasoningTokens: 0, webSearchRequests: 0, cacheCreationOneHourTokens: 0 }, + costUSD: 0.01, + speed: 'standard', + timestamp, + tools: [], + bashCommands: [], + skills: [], + subagentTypes: [], + deduplicationKey: `${key}-${i}`, + }], + } + }), + } +} + +const may = (day: string) => monthScopeForRange(new Date(`2026-05-${day}T00:00:00.000Z`), new Date(`2026-05-${day}T23:59:59.999Z`)) + +async function publish(files: Record): Promise { + const cache: SessionCache = { version: CACHE_VERSION, complete: true, providers: { claude: { envFingerprint: computeEnvFingerprint('claude'), files } } } + markCacheDirty(cache, 'claude') + expect(await saveCache(cache)).toBe(true) + clearLoadCacheMemo() +} + +async function pieces(): Promise> { + const dir = sessionCacheDir() + const envelope = JSON.parse(await readFile(join(dir, 'envelope.json'), 'utf-8')) + return JSON.parse(await readFile(join(dir, envelope.providers.claude.index), 'utf-8')).pieces +} + +describe('piece selection', () => { + it('reads members only from the pieces the range reports on; the rest come from key files', async () => { + const files = { '/p/a.jsonl': fileAt(['01'], 'a'), '/p/b.jsonl': fileAt(['05', '06'], 'b'), '/p/c.jsonl': fileAt(['20'], 'c') } + await publish(files) + const named = await pieces() + expect(Object.keys(named).sort()).toEqual(['2026-05-01', '2026-05-06', '2026-05-20']) + // Wreck the members of every piece but the 20th's: a load that read them + // would lose those files. + for (const day of ['2026-05-01', '2026-05-06']) await writeFile(join(sessionCacheDir(), named[day]!), 'not json') + + const cache = await loadCache(may('20')) + expect(cache.providers['claude']!.files).toEqual({ '/p/c.jsonl': files['/p/c.jsonl'] }) + const stubs = cacheStubs(cache, 'claude')! + expect([...stubs.keys()]).toEqual(['/p/a.jsonl', '/p/b.jsonl']) + expect(stubs.get('/p/b.jsonl')!.keys).toEqual([['b-0'], ['b-1']]) + expect(stubs.get('/p/b.jsonl')!.fingerprint).toEqual(files['/p/b.jsonl'].fingerprint) + }) + + it('rebuilds stub keys from the piece when its key file is gone (key files are not synced)', async () => { + const files = { '/p/a.jsonl': fileAt(['01'], 'a'), '/p/c.jsonl': fileAt(['20'], 'c') } + await publish(files) + await rm(join(sessionCacheDir(), (await pieces())['2026-05-01']!.replace(/\.json$/, '.keys.json'))) + const cache = await loadCache(may('20')) + expect(cacheStubs(cache, 'claude')!.get('/p/a.jsonl')!.keys).toEqual([['a-0']]) + }) +}) + +describe('an append that crosses a day', () => { + it('moves the file to its new piece and rewrites the old one without it', async () => { + await publish({ '/p/y.jsonl': fileAt(['02'], 'y'), '/p/v.jsonl': fileAt(['02'], 'v'), '/p/x.jsonl': fileAt(['20'], 'x') }) + const before = await pieces() + const oldPiece = await readFile(join(sessionCacheDir(), before['2026-05-02']!), 'utf-8') + + const cache = await loadCache(may('20')) + await loadCacheStubs(cache, 'claude', ['/p/y.jsonl']) + cache.providers['claude']!.files['/p/y.jsonl'] = fileAt(['02', '20'], 'y') + markCacheDirty(cache, 'claude', '/p/y.jsonl') + expect(await saveCache(cache)).toBe(true) + + const after = await pieces() + const dir = sessionCacheDir() + const day02 = await readFile(join(dir, after['2026-05-02']!), 'utf-8') + const day20 = JSON.parse(await readFile(join(dir, after['2026-05-20']!), 'utf-8')) + expect(after['2026-05-02']).not.toBe(before['2026-05-02']) + expect(existsSync(join(dir, before['2026-05-02']!))).toBe(false) + expect(Object.keys(JSON.parse(day02))).toEqual(['/p/v.jsonl']) + // The neighbour's line is copied, byte for byte. + const line = (text: string) => text.split('\n').find(l => l.startsWith('"/p/v.jsonl":'))!.replace(/,$/, '') + expect(line(day02)).toBe(line(oldPiece)) + expect(Object.keys(day20).sort()).toEqual(['/p/x.jsonl', '/p/y.jsonl']) + + clearLoadCacheMemo() + const full = await loadCache() + // The rewritten entry follows the loaded ones, where a re-parse puts it on v9. + expect(Object.keys(full.providers['claude']!.files)).toEqual(['/p/v.jsonl', '/p/x.jsonl', '/p/y.jsonl']) + expect(full.providers['claude']!.files['/p/y.jsonl']).toEqual(fileAt(['02', '20'], 'y')) + }) +}) + +describe('v9 -> v10 migration', () => { + const v9 = () => join(TMP_DIR, 'session-cache.v9') + + async function writeV9(): Promise> { + await mkdir(v9(), { recursive: true }) + const may = { '/p/b.jsonl': fileAt(['09'], 'b'), '/p/a.jsonl': fileAt(['01', '21'], 'a') } + const undated: Record = { '/p/failed.jsonl': { fingerprint: { dev: 1, ino: 9, mtimeMs: 3, sizeBytes: 4 }, mcpInventory: [], turns: [], failed: true } } + // The month shard as stage 1 wrote it (one member per line), the undated + // one as main wrote it (one line). + const lines = Object.entries(may).map(([k, v]) => `${JSON.stringify(k)}:${JSON.stringify(v)}`) + await writeFile(join(v9(), 'claude.2026-05.aa.json'), `{\n${lines.join(',\n')}\n}`) + await writeFile(join(v9(), 'claude.0000-00.bb.json'), JSON.stringify(undated)) + await writeFile(join(v9(), 'envelope.json'), JSON.stringify({ version: 9, complete: true, nonce: 'n', providers: { + claude: { envFingerprint: computeEnvFingerprint('claude'), complete: true, completeFrom: 5, shards: { + '0000-00': { name: 'claude.0000-00.bb.json', until: '0000-00' }, + '2026-05': { name: 'claude.2026-05.aa.json', until: '2026-05' }, + } }, + } })) + return { ...undated, ...may } + } + + it('is lossless, keeps v9 load order, and retires v9 once v10 is published', async () => { + const expected = await writeV9() + const cache = await loadCache() + expect(existsSync(v9())).toBe(false) + expect(cache.complete).toBe(true) + expect(cache.providers['claude']!.complete).toBe(true) + expect(cache.providers['claude']!.completeFrom).toBe(5) + expect(Object.keys(cache.providers['claude']!.files)).toEqual(['/p/failed.jsonl', '/p/b.jsonl', '/p/a.jsonl']) + expect(cache.providers['claude']!.files).toEqual(expected) + expect(Object.keys(await pieces()).sort()).toEqual(['0000-00-00', '2026-05-09', '2026-05-21']) + // Members moved verbatim. + const a = await readFile(join(sessionCacheDir(), (await pieces())['2026-05-21']!), 'utf-8') + expect(a).toBe(`{\n${JSON.stringify('/p/a.jsonl')}:${JSON.stringify(expected['/p/a.jsonl'])}\n}`) + }) + + it('leaves v9 in place when the v10 directory cannot be written', async () => { + await writeV9() + await writeFile(join(TMP_DIR, `session-cache.v${CACHE_VERSION}`), 'in the way') + await loadCache().catch(() => null) + expect(existsSync(join(v9(), 'envelope.json'))).toBe(true) + expect(await readdir(v9())).toHaveLength(3) + }) +}) + +describe('a narrow query and an older transcript', () => { + const home = () => join(TMP_DIR, 'home') + const saved = { dir: process.env['CLAUDE_CONFIG_DIR'], desktop: process.env['CODEBURN_DESKTOP_SESSIONS_DIR'] } + + beforeEach(() => { + process.env['CLAUDE_CONFIG_DIR'] = home() + process.env['CODEBURN_DESKTOP_SESSIONS_DIR'] = join(TMP_DIR, 'no-desktop') + clearSessionCache() + }) + afterEach(() => { + if (saved.dir === undefined) delete process.env['CLAUDE_CONFIG_DIR']; else process.env['CLAUDE_CONFIG_DIR'] = saved.dir + if (saved.desktop === undefined) delete process.env['CODEBURN_DESKTOP_SESSIONS_DIR']; else process.env['CODEBURN_DESKTOP_SESSIONS_DIR'] = saved.desktop + delete process.env['CODEBURN_PROGRESS'] + clearSessionCache() + }) + + async function session(name: string, iso: string): Promise { + const dir = join(home(), 'projects', 'proj') + await mkdir(dir, { recursive: true }) + await writeFile(join(dir, `${name}.jsonl`), [ + { type: 'user', uuid: `u-${name}`, sessionId: name, timestamp: iso, cwd: '/tmp/proj', message: { role: 'user', content: 'go' } }, + { type: 'assistant', uuid: `a-${name}`, sessionId: name, timestamp: iso, cwd: '/tmp/proj', + message: { id: `msg-${name}`, type: 'message', role: 'assistant', model: 'claude-sonnet-4-5', content: [], usage: { input_tokens: 100, output_tokens: 50 } } }, + ].map(line => JSON.stringify(line)).join('\n') + '\n') + } + + it('does not re-parse a cached June file on a September day, and writes nothing', async () => { + await session('june', '2099-06-10T10:00:00Z') + await session('today', '2099-09-23T10:00:00Z') + await parseAllSessions() + const day = { start: new Date('2099-09-23T00:00:00.000Z'), end: new Date('2099-09-23T23:59:59.999Z') } + const narrow = async (): Promise => { + clearSessionCache() + clearLoadCacheMemo() + const totals: number[] = [] + process.env['CODEBURN_PROGRESS'] = '1' + const spy = vi.spyOn(process.stderr, 'write').mockImplementation(((chunk: unknown) => { + const m = /"kind":"tick","provider":"claude","done":0,"total":(\d+)/.exec(String(chunk)) + if (m) totals.push(Number(m[1])) + return true + }) as typeof process.stderr.write) + try { + const projects = await parseAllSessions(day, 'claude') + expect(projects.reduce((n, p) => n + p.totalApiCalls, 0)).toBe(1) + } finally { + spy.mockRestore() + delete process.env['CODEBURN_PROGRESS'] + } + return totals + } + await narrow() + const envelope = await readFile(join(sessionCacheDir(), 'envelope.json'), 'utf-8') + expect(await narrow()).toEqual([0]) + expect(await readFile(join(sessionCacheDir(), 'envelope.json'), 'utf-8')).toBe(envelope) + }) + + it('lets the cached June file keep the calls a fork made today replays', async () => { + await session('june', '2099-06-10T10:00:00Z') + await parseAllSessions() + // A fork of the June session: its message id restated under today's date, then a new call. + const dir = join(home(), 'projects', 'proj') + await writeFile(join(dir, 'fork.jsonl'), [ + { type: 'user', uuid: 'u-fork-0', sessionId: 'fork', timestamp: '2099-09-23T10:00:00Z', cwd: '/tmp/proj', message: { role: 'user', content: 'go' } }, + { type: 'assistant', uuid: 'a-fork-0', sessionId: 'fork', timestamp: '2099-09-23T10:00:00Z', cwd: '/tmp/proj', + message: { id: 'msg-june', type: 'message', role: 'assistant', model: 'claude-sonnet-4-5', content: [], usage: { input_tokens: 100, output_tokens: 50 } } }, + { type: 'user', uuid: 'u-fork-1', sessionId: 'fork', timestamp: '2099-09-23T10:01:00Z', cwd: '/tmp/proj', message: { role: 'user', content: 'more' } }, + { type: 'assistant', uuid: 'a-fork-1', sessionId: 'fork', timestamp: '2099-09-23T10:01:00Z', cwd: '/tmp/proj', + message: { id: 'msg-fork', type: 'message', role: 'assistant', model: 'claude-sonnet-4-5', content: [], usage: { input_tokens: 100, output_tokens: 50 } } }, + ].map(line => JSON.stringify(line)).join('\n') + '\n') + const calls = async (range?: { start: Date; end: Date }): Promise => { + clearSessionCache() + clearLoadCacheMemo() + return (await parseAllSessions(range, 'claude')).reduce((n, p) => n + p.totalApiCalls, 0) + } + expect(await calls({ start: new Date('2099-09-23T00:00:00.000Z'), end: new Date('2099-09-23T23:59:59.999Z') })).toBe(1) + expect(await calls({ start: new Date('2099-06-01T00:00:00.000Z'), end: new Date('2099-06-30T23:59:59.999Z') })).toBe(1) + expect(await calls()).toBe(2) + }) +}) diff --git a/tests/session-cache-shards.test.ts b/tests/session-cache-shards.test.ts index 4855e3d9..958fa89b 100644 --- a/tests/session-cache-shards.test.ts +++ b/tests/session-cache-shards.test.ts @@ -1,9 +1,10 @@ -// 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. +// Provider x day piece layout (CACHE_VERSION 10): the on-disk cache is a +// directory holding one envelope, one index per provider and one piece (plus +// its key file) per provider-day. What matters here is that the move off the +// older layouts loses nothing, that a save rewrites only the pieces that +// changed (including when the load was scoped to a subset of the months), that +// a file keeps its v9 month for scoping whichever piece it is in, and that one +// unreadable piece costs exactly the files it held. import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { mkdir, readFile, readdir, rm, stat, utimes, writeFile } from 'fs/promises' import { existsSync } from 'fs' @@ -116,10 +117,18 @@ async function shardNames(): Promise { return (await readdir(sessionCacheDir())).sort() } -async function envelope(): Promise<{ providers: Record }> }> { +async function envelope(): Promise<{ providers: Record }> { return JSON.parse(await readFile(join(sessionCacheDir(), 'envelope.json'), 'utf-8')) } +async function index(provider: string): Promise<{ pieces: Record; rows: unknown[][] }> { + return JSON.parse(await readFile(join(sessionCacheDir(), (await envelope()).providers[provider]!.index), 'utf-8')) +} + +async function pieces(provider: string): Promise> { + return (await index(provider)).pieces +} + /** name -> bytes, for every shard on disk. */ async function shardBytes(): Promise> { const dir = sessionCacheDir() @@ -142,9 +151,9 @@ describe('v7 -> shard migration', () => { expect(existsSync(v7Path)).toBe(false) const names = await shardNames() expect(names).toContain('envelope.json') - // 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']) + // claude's three entries split by day: two called on 2026-05-15, one turn-less. + expect(Object.keys(await pieces('claude')).sort()).toEqual(['0000-00-00', '2026-05-15']) + expect(Object.keys(await pieces('codex'))).toEqual(['2026-05-15']) // A second load reads only the shards and produces the same cache. clearLoadCacheMemo() @@ -191,7 +200,7 @@ describe('per-provider dirty tracking', () => { }) describe('corrupt shard isolation', () => { - it('drops only the unreadable month, keeping every other month and provider', async () => { + it('drops only the unreadable piece, keeping every other piece and provider', async () => { const cache: SessionCache = { version: CACHE_VERSION, complete: true, @@ -208,14 +217,14 @@ describe('corrupt shard isolation', () => { await saveCache(cache) const dir = sessionCacheDir() - await writeFile(join(dir, (await envelope()).providers['claude']!.shards['2026-05']!.name), '{"/x":{"turns":') + await writeFile(join(dir, (await pieces('claude'))['2026-05-15']!), '{"/x":{"turns":') clearLoadCacheMemo() const reloaded = await loadCache() 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 + // Self-heals: the unreadable piece 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') @@ -304,9 +313,10 @@ describe('concurrent writers', () => { async function assertReferentialIntegrity(expected: string[]): Promise { const dir = sessionCacheDir() - 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) + for (const [provider, meta] of Object.entries((await envelope()).providers)) { + expect(existsSync(join(dir, meta.index)), `envelope names a missing index: ${meta.index}`).toBe(true) + for (const name of Object.values(await pieces(provider))) { + expect(existsSync(join(dir, name)), `index names a missing piece: ${name}`).toBe(true) } } clearLoadCacheMemo() @@ -366,8 +376,8 @@ describe('concurrent writers', () => { }) }) -describe('month buckets', () => { - it('keeps a file in its first-turn month when the session is appended to', async () => { +describe('day pieces', () => { + it('moves an appended file to the day of its newest call and keeps its first-turn month', async () => { const cache: SessionCache = { version: CACHE_VERSION, complete: true, @@ -375,26 +385,27 @@ describe('month buckets', () => { } markCacheDirty(cache, 'claude') await saveCache(cache) - expect(Object.keys((await envelope()).providers['claude']!.shards)).toEqual(['2026-05']) + expect(Object.keys(await pieces('claude'))).toEqual(['2026-05-15']) - // Two months of appends later the bucket is unchanged; only `until` moves, - // which is what lets a ranged load still find this session. + // Two months of appends later the file lives in July's piece, but its v9 + // month 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') + expect(Object.keys(await pieces('claude'))).toEqual(['2026-07-02']) + const [row] = (await index('claude')).rows + expect(row!.slice(0, 4)).toEqual(['/live/long.jsonl', '2026-07-02', '2026-05', '2026-07']) - // ...and a July query still loads it, despite the May bucket key. + // ...and a July query still loads it. 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 () => { + it('rewrites only the day that changed', async () => { const cache: SessionCache = { version: CACHE_VERSION, complete: true, @@ -412,10 +423,7 @@ describe('month buckets', () => { 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, - ] + const untouched = [(await pieces('claude'))['2026-03-10']!, (await pieces('claude'))['2026-04-10']!] cache.providers['claude']!.files['/live/may.jsonl'] = cachedFile({ turns: [turnAt('2026-05-10T10:00:00Z', 'a')], mcpInventory: ['changed'] }) markCacheDirty(cache, 'claude', '/live/may.jsonl') @@ -423,10 +431,10 @@ describe('month buckets', () => { 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) + expect(after.has((await pieces('claude'))['2026-05-10']!)).toBe(true) }) - it('dirties the month a deleted file was in', async () => { + it('dirties the piece a deleted file was in', async () => { const cache: SessionCache = { version: CACHE_VERSION, complete: true, @@ -483,10 +491,7 @@ describe('scoped load', () => { 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, - ] + const kept = [(await pieces('claude'))['2026-03-10']!, (await pieces('claude'))['2026-04-10']!] clearLoadCacheMemo() const scoped = await loadCache(juneScope) @@ -503,12 +508,12 @@ describe('scoped load', () => { .toEqual(['/live/apr.jsonl', '/live/jun.jsonl', '/live/jun2.jsonl', '/live/mar.jsonl']) }) - it('keeps the unloaded month\'s shard name when a re-parse re-derives the same entry', async () => { + it('keeps the unloaded month\'s piece name when a re-parse re-derives the same entry', async () => { await seedThreeMonths() - // The March entry is invisible to a June-scoped run, so the reconcile - // re-parses that file and writes the identical entry straight back. Nothing - // changed, so the March shard must keep its name run after run (#1032). - const nameOf = async (): Promise => (await envelope()).providers['claude']!.shards['2026-03']!.name + // A changed March file is re-parsed from a June-scoped run and can write + // the identical entry straight back. Nothing changed, so the March piece + // must keep its name run after run (#1032). + const nameOf = async (): Promise => (await pieces('claude'))['2026-03-10']! const before = await nameOf() for (let run = 0; run < 2; run++) { clearLoadCacheMemo() @@ -582,8 +587,8 @@ describe('scoped load', () => { }) }) -describe('v8 -> v9 migration', () => { - it('re-buckets the v8 provider shards losslessly and retires the v8 directory', async () => { +describe('v8 migration', () => { + it('re-lays the v8 provider shards out losslessly and retires the v8 directory', async () => { const v8Dir = join(TMP_DIR, 'session-cache.v8') await mkdir(v8Dir, { recursive: true }) const section = { @@ -603,7 +608,7 @@ describe('v8 -> v9 migration', () => { 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']) + expect(Object.keys(await pieces('claude')).sort()).toEqual(['2026-03-10', '2026-05-15', '2026-06-10']) clearLoadCacheMemo() expect(await loadCache()).toEqual(loaded) @@ -631,7 +636,7 @@ describe('out-of-order turns', () => { } markCacheDirty(cache, 'claude') await saveCache(cache) - expect((await envelope()).providers['claude']!.shards['2026-03']!.until).toBe('2026-08') + expect((await index('claude')).rows[0]!.slice(1, 4)).toEqual(['2026-08-10', '2026-03', '2026-08']) for (const [from, to] of [['2026-03-01', '2026-03-31'], ['2026-08-01', '2026-08-31']] as const) { clearLoadCacheMemo() @@ -641,9 +646,8 @@ describe('out-of-order turns', () => { }) }) -// 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. +// A path must never end up in two pieces at once: a stale copy could be read +// back, and nothing would sweep it. 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')) @@ -666,13 +670,13 @@ describe('re-bucketing out of an unloaded month', () => { await saveCache(cache) } - /** Every shard's view of `path`, so a duplicate is visible directly. */ + /** Every piece'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) + for (const [day, name] of Object.entries(await pieces('claude'))) { + const files = JSON.parse(await readFile(join(dir, name), 'utf-8')) + if (files[path]) found.push(day) } return found.sort() } @@ -687,7 +691,7 @@ describe('re-bucketing out of an unloaded month', () => { markCacheDirty(scoped, 'claude', '/live/moving.jsonl') await saveCache(scoped) - expect(await copiesOf('/live/moving.jsonl')).toEqual(['2026-05']) + expect(await copiesOf('/live/moving.jsonl')).toEqual(['2026-05-02']) clearLoadCacheMemo() const full = await loadCache() expect(full.providers['claude']!.files['/live/moving.jsonl']!.mcpInventory).toEqual(['reparsed']) @@ -703,31 +707,30 @@ describe('re-bucketing out of an unloaded month', () => { markCacheDirty(scoped, 'claude', '/live/moving.jsonl') await saveCache(scoped) - expect(await copiesOf('/live/moving.jsonl')).toEqual(['0000-00']) + expect(await copiesOf('/live/moving.jsonl')).toEqual(['0000-00-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)) + it('resolves a v9 duplicate to the freshest copy when it re-lays the shards out', async () => { + const v9 = join(TMP_DIR, 'session-cache.v9') + await mkdir(v9, { recursive: true }) + const stale = fileSpanning('2026-01-10T10:00:00Z') + const fresh = cachedFile({ turns: [turnAt('2026-06-02T10:00:00Z', 'fresh')], fingerprint: { dev: 1, ino: 2, mtimeMs: 999, sizeBytes: 4 }, mcpInventory: ['fresh'] }) + await writeFile(join(v9, 'claude.2026-01.a.json'), JSON.stringify({ '/live/moving.jsonl': stale, '/live/stay.jsonl': fileSpanning('2026-01-11T10:00:00Z') })) + await writeFile(join(v9, 'claude.2026-06.b.json'), JSON.stringify({ '/live/moving.jsonl': fresh })) + await writeFile(join(v9, 'envelope.json'), JSON.stringify({ version: 9, complete: true, nonce: 'n', providers: { + claude: { envFingerprint: computeEnvFingerprint('claude'), shards: { '2026-01': { name: 'claude.2026-01.a.json', until: '2026-01' }, '2026-06': { name: 'claude.2026-06.b.json', until: '2026-06' } } }, + } })) - 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) + // Newest fingerprint wins, at the first copy's place in load order. + expect(Object.keys(full.providers['claude']!.files)).toEqual(['/live/moving.jsonl', '/live/stay.jsonl']) + expect(full.providers['claude']!.files['/live/moving.jsonl']).toEqual(fresh) + expect(await copiesOf('/live/moving.jsonl')).toEqual(['2026-06-02']) + expect(existsSync(v9)).toBe(false) }) }) @@ -765,11 +768,11 @@ describe('carried months under a concurrent writer', () => { clearLoadCacheMemo() const final = await loadCache() - // March survived under A's name, with both orphans; June has both files. + // March survived 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) + for (const name of Object.values(await pieces('claude'))) { + expect(existsSync(join(sessionCacheDir(), name))).toBe(true) } }) @@ -809,8 +812,8 @@ describe('carried months under a concurrent writer', () => { 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) + for (const name of Object.values(await pieces('claude'))) { + expect(existsSync(join(sessionCacheDir(), name))).toBe(true) } expect(final.providers['claude']!.files['/live/mar.jsonl']).toBeDefined() expect(final.providers['claude']!.files['/live/jun.jsonl']).toBeDefined() @@ -820,7 +823,7 @@ describe('carried months under a concurrent writer', () => { }) describe('retiring an orphaned prior layout', () => { - it('sweeps a v8 directory and a v7 file left behind by an interrupted re-layout', async () => { + it('sweeps v9 and v8 directories 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') } }, } }) @@ -830,17 +833,23 @@ describe('retiring an orphaned prior layout', () => { await writeFile(join(v8Dir, 'claude.abc.json'), '{}') const v7 = join(TMP_DIR, 'session-cache.v7.json') await writeFile(v7, '{}') + const v9Dir = join(TMP_DIR, 'session-cache.v9') + await mkdir(v9Dir, { recursive: true }) + await writeFile(join(v9Dir, 'envelope.json'), JSON.stringify({ version: 9, nonce: 'n', providers: {} })) // Fresh: an in-flight write by an older binary must be left alone. await cleanupOrphanedTempFiles() expect(existsSync(v8Dir)).toBe(true) + expect(existsSync(v9Dir)).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(join(v9Dir, 'envelope.json'), old, old) await utimes(v7, old, old) await cleanupOrphanedTempFiles() expect(existsSync(v8Dir)).toBe(false) + expect(existsSync(v9Dir)).toBe(false) expect(existsSync(v7)).toBe(false) }) })