mirror of
https://github.com/AgentSeal/codeburn.git
synced 2026-10-20 11:44:39 +00:00
perf: load only the session-cache members a date range can use (#1542)
* Write session shards one member per line A shard is still one JSON object, so older builds read it unchanged, but it is written through a stream and never exists as one string. Past V8's string limit the save used to throw and the cache silently never persisted; reads of such a shard now go member by member. * Hold cached files a ranged query cannot report on as stubs A ranged load streams its in-scope shards and keeps in full only the files with a call in the range, spawn anchors, failure markers, turn-less files and WSL paths. Every other file becomes a stub: its fingerprint, bucket, PR flag, dedup keys per turn and where its member line lives. A stub stands in for its entry everywhere but the summaries it cannot contribute to: reconcile, the seenMsgIds and parser dedup pre-seeds, the query-time key replay in serve order, orphan surfacing, eviction and the emptiness checks. A stub whose file changed loads its member first, so the append or re-parse runs as before, and a wider request in the same process loads the stubs it now needs. A bucket holding stubs is rewritten as a full load would write it, copying the stubs' lines from the shard they came from; unchanged, the published shard is kept. A single-line shard from an older build is rewritten on the next save. Providers whose parse reads cached entries no report shows (hermes, quickdesk, antigravity, devin, gemini) still load in full. * Discover codex projects from a small index Discovery only needs each rollout's project, but it loaded the whole codex results file (hundreds of MB on a large corpus) on every run to get it. flushCodexCache now also writes codex-projects.v15.json, keyed by the same fingerprint, and discovery reads that. A missing index is built once from the results file. * Load the session cache once for status today and month status asks for today, then the month. A command can now declare the union of its ranges, so the first load already holds the month in full and the second request finds it there; each request keeps its own month scope and summaries. * Release the pre-lock cache view before reloading Once the refresh lock is acquired or another process completed the refresh, the snapshot loaded before the wait is never used again, but it stayed referenced while the reload read a second view. * Test the first-paint snapshot serves stubs in load order * Parse a shard whole when the range covers all its months A range that spans a shard's months needs nearly every member, and streaming it one member at a time only added allocation churn (about 200MB more RSS on report -p all over a 1GB cache). The streaming reader also reuses one read buffer instead of allocating a fresh one per chunk. * Parse a shard whole once the range reaches its oldest month The previous rule also required the range to reach past the shard's newest month, which a range ending now never does, so report -p all still streamed every shard. * Carry a shard line that spans reads as decoded text Members of a megabyte and more span the 1MB reads, and joining their pieces as Buffers allocated off-heap memory that only a GC returns: report -p all over a 1GB cache peaked about 170MB above main in RSS. Decoding the pieces with a StringDecoder keeps them on the heap and brings every streamed range below main, so the whole-file parse for wide ranges goes again. * Changelog: bounded session-cache load
This commit is contained in:
Resham Joshi
GitHub
parent
e84889762c
commit
22ebfd463e
8 changed files with 1066 additions and 67 deletions
|
|
@ -5,6 +5,7 @@
|
|||
### Fixed
|
||||
- **Windows Store installs: no update offer the Store cannot deliver, and `codeburn menubar` opens the Store copy.** The desktop app read GitHub releases for its update banner, so a Store copy was told about a version still waiting on Store certification and sent to a Store page that did not have it. Store installs now skip the GitHub check and the About update button, like the tray app already did; the Store updates them. `codeburn menubar` spawned the tray exe inside `WindowsApps`, which Windows refuses from outside the package (`spawn EPERM`); it now opens the Store app by its app id, and the desktop app starts the tray. Reported in #1520, thanks @Brandin.
|
||||
- **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.
|
||||
- **`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.
|
||||
- Cache-write tokens on a non-Anthropic model with no published write price are billed at the input rate, not an invented 1.25x. Daily cache moves to v35 so settled days re-derive.
|
||||
|
|
|
|||
|
|
@ -53,6 +53,12 @@ export const CODEX_LEGACY_CACHE_FILE = 'codex-results.json'
|
|||
export function codexCacheFileName(version = CODEX_CACHE_VERSION): string {
|
||||
return `codex-results.v${version}.json`
|
||||
}
|
||||
// Discovery only needs each rollout's project, so it reads this small index
|
||||
// instead of the whole results file. Keyed by the same fingerprint, so an
|
||||
// index that lags the results file only misses, never answers wrong.
|
||||
export function codexProjectsFileName(version = CODEX_CACHE_VERSION): string {
|
||||
return `codex-projects.v${version}.json`
|
||||
}
|
||||
|
||||
export type CodexFileFingerprint = { dev: number; ino: number; mtimeMs: number; sizeBytes: number }
|
||||
type FileFingerprint = CodexFileFingerprint
|
||||
|
|
@ -83,6 +89,8 @@ type ResultCache = {
|
|||
files: Record<string, FileEntry>
|
||||
}
|
||||
|
||||
type ProjectEntry = Pick<FileEntry, 'dev' | 'ino' | 'mtimeMs' | 'sizeBytes' | 'project'>
|
||||
|
||||
const cacheDirContext = new AsyncLocalStorage<string>()
|
||||
|
||||
function currentCacheDir(): string {
|
||||
|
|
@ -119,6 +127,57 @@ const memCaches = new Map<string, ResultCache>()
|
|||
export function clearCodexMemCaches(): void {
|
||||
memCaches.clear()
|
||||
inFlightLoads.clear()
|
||||
projectIndexes.clear()
|
||||
}
|
||||
|
||||
const projectIndexes = new Map<string, Promise<Record<string, ProjectEntry>>>()
|
||||
|
||||
function loadProjectIndex(cacheDir: string): Promise<Record<string, ProjectEntry>> {
|
||||
let index = projectIndexes.get(cacheDir)
|
||||
if (!index) {
|
||||
index = readProjectIndex(cacheDir)
|
||||
projectIndexes.set(cacheDir, index)
|
||||
}
|
||||
return index
|
||||
}
|
||||
|
||||
// Missing (a cache written before the index existed): built once from the
|
||||
// results file, which is what discovery read before.
|
||||
async function readProjectIndex(cacheDir: string): Promise<Record<string, ProjectEntry>> {
|
||||
try {
|
||||
const parsed = JSON.parse(await readFile(join(cacheDir, codexProjectsFileName()), 'utf-8')) as { version?: unknown; files?: unknown }
|
||||
if (parsed.version === CODEX_CACHE_VERSION && parsed.files && typeof parsed.files === 'object') return parsed.files as Record<string, ProjectEntry>
|
||||
} catch {}
|
||||
const cache = await loadCache(cacheDir)
|
||||
if (Object.keys(cache.files).length > 0) await writeProjectIndex(cacheDir, cache)
|
||||
return cache.files
|
||||
}
|
||||
|
||||
async function writeProjectIndex(cacheDir: string, cache: ResultCache): Promise<void> {
|
||||
const files: Record<string, ProjectEntry> = {}
|
||||
for (const [path, e] of Object.entries(cache.files)) {
|
||||
files[path] = { dev: e.dev, ino: e.ino, mtimeMs: e.mtimeMs, sizeBytes: e.sizeBytes, project: e.project }
|
||||
}
|
||||
try {
|
||||
await writeFileAtomic(join(cacheDir, codexProjectsFileName()), JSON.stringify({ version: CODEX_CACHE_VERSION, files }))
|
||||
} catch {}
|
||||
}
|
||||
|
||||
async function writeFileAtomic(finalPath: string, payload: string): Promise<void> {
|
||||
const tempPath = `${finalPath}.${randomBytes(8).toString('hex')}.tmp`
|
||||
const handle = await open(tempPath, 'w', 0o600)
|
||||
try {
|
||||
await handle.writeFile(payload, { encoding: 'utf-8' })
|
||||
await handle.sync()
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
try {
|
||||
await rename(tempPath, finalPath)
|
||||
} catch (err) {
|
||||
try { await unlink(tempPath) } catch {}
|
||||
throw err
|
||||
}
|
||||
}
|
||||
|
||||
// Concurrent callers must share one load. The memo below is only populated
|
||||
|
|
@ -168,9 +227,9 @@ async function loadCacheFromDisk(cacheDir: string): Promise<ResultCache> {
|
|||
return empty
|
||||
}
|
||||
|
||||
function getEntry(cache: ResultCache, filePath: string, fp: FileFingerprint): FileEntry | null {
|
||||
if (!Object.hasOwn(cache.files, filePath)) return null
|
||||
const entry = cache.files[filePath]
|
||||
function getEntry<T extends ProjectEntry>(files: Record<string, T>, filePath: string, fp: FileFingerprint): T | null {
|
||||
if (!Object.hasOwn(files, filePath)) return null
|
||||
const entry = files[filePath]
|
||||
if (
|
||||
entry
|
||||
&& entry.dev === fp.dev
|
||||
|
|
@ -218,7 +277,7 @@ export async function readCachedCodexResults(
|
|||
const s = await stat(filePath)
|
||||
const cache = await loadCache(currentCacheDir())
|
||||
const fp = fingerprintFromStat(filePath, s)
|
||||
const entry = getEntry(cache, filePath, fp)
|
||||
const entry = getEntry(cache.files, filePath, fp)
|
||||
if (entry) return { kind: 'exact', calls: entry.calls }
|
||||
// Rollouts are append-only: the same inode, grown past a boundary we
|
||||
// recorded, can be picked up from that boundary instead of re-read whole.
|
||||
|
|
@ -245,8 +304,9 @@ export async function getCachedCodexProject(
|
|||
): Promise<string | null> {
|
||||
try {
|
||||
const s = await stat(filePath)
|
||||
const cache = await loadCache(currentCacheDir())
|
||||
const entry = getEntry(cache, filePath, fingerprintFromStat(filePath, s))
|
||||
const cacheDir = currentCacheDir()
|
||||
const files = memCaches.get(cacheDir)?.files ?? await loadProjectIndex(cacheDir)
|
||||
const entry = getEntry(files, filePath, fingerprintFromStat(filePath, s))
|
||||
return entry?.project ?? null
|
||||
} catch {}
|
||||
return null
|
||||
|
|
@ -323,21 +383,8 @@ export async function flushCodexCache(): Promise<void> {
|
|||
}
|
||||
|
||||
if (!existsSync(cacheDir)) await mkdir(cacheDir, { recursive: true })
|
||||
const finalPath = getCachePath(cacheDir)
|
||||
const tempPath = `${finalPath}.${randomBytes(8).toString('hex')}.tmp`
|
||||
const payload = JSON.stringify(memCache)
|
||||
const handle = await open(tempPath, 'w', 0o600)
|
||||
try {
|
||||
await handle.writeFile(payload, { encoding: 'utf-8' })
|
||||
await handle.sync()
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
try {
|
||||
await rename(tempPath, finalPath)
|
||||
} catch (err) {
|
||||
try { await unlink(tempPath) } catch {}
|
||||
throw err
|
||||
}
|
||||
await writeFileAtomic(getCachePath(cacheDir), JSON.stringify(memCache))
|
||||
await writeProjectIndex(cacheDir, memCache)
|
||||
projectIndexes.delete(cacheDir)
|
||||
} catch {}
|
||||
}
|
||||
|
|
|
|||
14
src/main.ts
14
src/main.ts
|
|
@ -7,7 +7,7 @@ import { cachedProjectIdentitiesForRange } from './daily-cache.js'
|
|||
import { reportUnmatchedProjectPatterns } from './project-filter-warnings.js'
|
||||
import { getVercelGatewayApiKey } from './providers/vercel-gateway.js'
|
||||
import { BILLING_FILTER_VALUES, ROUTE_FILTER_VALUES, filterProjectsByBillingRoute } from './billing-filter.js'
|
||||
import { AGGREGATE_ONLY_PROVIDER, aggregateOnlyCostUSD, excludesAggregateOnlyProviders, parseAllSessions, filterProjectsByName, filterProjectsByDateRange, clearSessionCache, setInteractiveScanUI, computeCorpusFingerprint, isSessionHydrationComplete } from './parser.js'
|
||||
import { AGGREGATE_ONLY_PROVIDER, aggregateOnlyCostUSD, excludesAggregateOnlyProviders, parseAllSessions, filterProjectsByName, filterProjectsByDateRange, clearSessionCache, setInteractiveScanUI, computeCorpusFingerprint, isSessionHydrationComplete, withLoadWindow } from './parser.js'
|
||||
import { allProviderNames, getAllProviders } from './providers/index.js'
|
||||
import { getProvider } from './providers/index.js'
|
||||
import { getClaudeConfigDirs, getDesktopSessionsDirs } from './providers/claude.js'
|
||||
|
|
@ -1354,10 +1354,12 @@ program
|
|||
|
||||
if (opts.format === 'json') {
|
||||
// Durable totals so the compact status matches the menubar / report.
|
||||
const todayDurable = await buildDurablePeriod(getDateRange('today'), { provider: pf, project: opts.project, exclude: opts.exclude })
|
||||
const [todayDurable, monthDurable] = await withLoadWindow(getDateRange('month').range, async () => [
|
||||
await buildDurablePeriod(getDateRange('today'), { provider: pf, project: opts.project, exclude: opts.exclude }),
|
||||
await buildDurablePeriod(getDateRange('month'), { provider: pf, project: opts.project, exclude: opts.exclude }),
|
||||
] as const)
|
||||
const todayData = todayDurable.data
|
||||
const todayProjects = todayDurable.liveProjects
|
||||
const monthDurable = await buildDurablePeriod(getDateRange('month'), { provider: pf, project: opts.project, exclude: opts.exclude })
|
||||
await reportUnmatchedProjectPatterns([...todayDurable.knownProjects, ...monthDurable.knownProjects], opts.project, opts.exclude)
|
||||
const monthData = monthDurable.data
|
||||
const monthProjects = monthDurable.liveProjects
|
||||
|
|
@ -1391,8 +1393,10 @@ program
|
|||
return
|
||||
}
|
||||
|
||||
const todayDurable = await buildDurablePeriod(getDateRange('today'), { provider: pf, project: opts.project, exclude: opts.exclude })
|
||||
const monthDurable = await buildDurablePeriod(getDateRange('month'), { provider: pf, project: opts.project, exclude: opts.exclude })
|
||||
const [todayDurable, monthDurable] = await withLoadWindow(getDateRange('month').range, async () => [
|
||||
await buildDurablePeriod(getDateRange('today'), { provider: pf, project: opts.project, exclude: opts.exclude }),
|
||||
await buildDurablePeriod(getDateRange('month'), { provider: pf, project: opts.project, exclude: opts.exclude }),
|
||||
] as const)
|
||||
await reportUnmatchedProjectPatterns([...todayDurable.knownProjects, ...monthDurable.knownProjects], opts.project, opts.exclude)
|
||||
console.log(renderStatusBar([], {
|
||||
today: { cost: todayDurable.data.cost, calls: todayDurable.data.calls },
|
||||
|
|
|
|||
124
src/parser.ts
124
src/parser.ts
|
|
@ -28,18 +28,25 @@ import {
|
|||
type ProviderSection,
|
||||
type SessionCache,
|
||||
beginColdHydration,
|
||||
cacheEntriesInLoadOrder,
|
||||
cacheStubs,
|
||||
cleanupOrphanedTempFiles,
|
||||
computeEnvFingerprint,
|
||||
DURABLE_PROVIDER_NAMES,
|
||||
evictCacheStub,
|
||||
fingerprintFile,
|
||||
hasCachedEntries,
|
||||
hasDirtyDurableProvider,
|
||||
isCacheComplete,
|
||||
isCacheCurrent,
|
||||
isCacheDirty,
|
||||
isCacheStub,
|
||||
loadCache,
|
||||
loadCacheStubs,
|
||||
markCacheDirty,
|
||||
markProviderComplete,
|
||||
monthScopeForRange,
|
||||
type CacheLoadScope,
|
||||
reconcileFile,
|
||||
saveCache,
|
||||
sourcePathStatCandidates,
|
||||
|
|
@ -2045,15 +2052,30 @@ async function scanProjectDirs(
|
|||
rememberSweepFingerprint(e.filePath, fp, sweptAt)
|
||||
return fp
|
||||
})
|
||||
// A stub stands in for its entry. One whose file changed loads its member
|
||||
// first, so the append or re-parse below runs exactly as on a full load.
|
||||
const stubs = cacheStubs(diskCache, 'claude')
|
||||
if (stubs?.size && !readOnly) {
|
||||
await loadCacheStubs(diskCache, 'claude', discovered.flatMap(({ filePath }, i) => {
|
||||
const stub = stubs.get(filePath)
|
||||
return stub && fingerprints[i] && reconcileFile(fingerprints[i]!, stub).action !== 'unchanged' ? [filePath] : []
|
||||
}))
|
||||
}
|
||||
const servedStubs: string[][][] = []
|
||||
|
||||
for (const [i, { filePath, dirName, source }] of discovered.entries()) {
|
||||
allDiscoveredFiles.add(filePath)
|
||||
const fp = fingerprints[i]
|
||||
if (!fp) continue
|
||||
|
||||
const cached = section.files[filePath]
|
||||
const action = reconcileFile(fp, cached)
|
||||
if (!readOnly && deferToBackgroundFill(filePath, fp, cached)) {
|
||||
const stub = cached ? undefined : stubs?.get(filePath)
|
||||
const action = reconcileFile(fp, cached ?? stub)
|
||||
if (!readOnly && deferToBackgroundFill(filePath, fp, cached ?? stub)) {
|
||||
continue
|
||||
} else if (stub && (readOnly || action.action === 'unchanged')) {
|
||||
if (readOnly && action.action !== 'unchanged') readOnlyServedStale = true
|
||||
servedStubs.push(stub.keys)
|
||||
} else if (cached && (readOnly || action.action === 'unchanged')) {
|
||||
if (readOnly && action.action !== 'unchanged') readOnlyServedStale = true
|
||||
unchangedFiles.push({ filePath, dirName, source, cached: section.files[filePath]! })
|
||||
|
|
@ -2083,8 +2105,13 @@ async function scanProjectDirs(
|
|||
// is gone and can never re-parse, but they carry attributable PR spend the by-PR
|
||||
// report must keep (as a legacy even-split); the eviction below preserves the
|
||||
// same set so `section.files` still holds them when summaries are built.
|
||||
for (const [filePath, cached] of Object.entries(section.files)) {
|
||||
for (const [filePath, cached] of cacheEntriesInLoadOrder(diskCache, 'claude')) {
|
||||
if (allDiscoveredFiles.has(filePath)) continue
|
||||
if (isCacheStub(cached)) {
|
||||
// Never a WSL path (those always load in full).
|
||||
if (readOnly || cached.hasPr) servedStubs.push(cached.keys)
|
||||
continue
|
||||
}
|
||||
const wslStatus = classifyWslCachePath(filePath, wslHomesForOrphans)
|
||||
// A WSL path missing from discovery is retained only when its root is
|
||||
// currently offline (or WSL discovery is explicitly disabled). If the
|
||||
|
|
@ -2106,6 +2133,7 @@ async function scanProjectDirs(
|
|||
}
|
||||
}
|
||||
}
|
||||
for (const keys of servedStubs) 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
|
||||
|
|
@ -2377,6 +2405,10 @@ async function scanProjectDirs(
|
|||
delete section.files[cachedPath]
|
||||
markCacheDirty(diskCache, 'claude', cachedPath)
|
||||
}
|
||||
for (const [cachedPath, stub] of stubs ?? []) {
|
||||
if (allDiscoveredFiles.has(cachedPath) || stub.hasPr || dirs.length === 0) continue
|
||||
evictCacheStub(diskCache, 'claude', cachedPath)
|
||||
}
|
||||
}
|
||||
|
||||
const projectMap = new Map<string, { project: string; projectPath: string; sessions: SessionSummary[]; anchors: SessionSummary[]; dirNames: Set<string> }>()
|
||||
|
|
@ -3426,6 +3458,20 @@ export async function parseProviderSources(
|
|||
return fp
|
||||
})
|
||||
|
||||
// As in scanProjectDirs: a changed stub (and every network source, which
|
||||
// re-fetches on a write run) loads its member so the parse below runs as on
|
||||
// a full load. An unchanged one only ever contributes its dedup keys.
|
||||
const stubs = cacheStubs(diskCache, providerName)
|
||||
const stubServed = new Set<string>()
|
||||
if (stubs?.size && !readOnly) {
|
||||
await loadCacheStubs(diskCache, providerName, sources.flatMap((s, i) => {
|
||||
const stub = stubs.get(s.path)
|
||||
if (!stub) return []
|
||||
const fp = sourceFingerprints[i]
|
||||
return skipFingerprint || (fp && reconcileFile(fp, stub).action !== 'unchanged') ? [s.path] : []
|
||||
}))
|
||||
}
|
||||
|
||||
for (const [sourceIndex, source] of sources.entries()) {
|
||||
allDiscoveredFiles.add(source.path)
|
||||
|
||||
|
|
@ -3465,12 +3511,16 @@ export async function parseProviderSources(
|
|||
}
|
||||
|
||||
const cached = section.files[source.path]
|
||||
const action = reconcileFile(fp, cached)
|
||||
const stub = cached ? undefined : stubs?.get(source.path)
|
||||
const action = reconcileFile(fp, cached ?? stub)
|
||||
// A cached parse failure at this same fingerprint stays skipped — don't
|
||||
// re-read a file that already threw and hasn't changed. It re-parses only
|
||||
// when the file changes (then `reconcileFile` reports non-'unchanged').
|
||||
if (!readOnly && deferToBackgroundFill(source.path, fp, cached)) {
|
||||
if (!readOnly && deferToBackgroundFill(source.path, fp, cached ?? stub)) {
|
||||
continue
|
||||
} else if (stub && (readOnly || action.action === 'unchanged')) {
|
||||
if (readOnly && action.action !== 'unchanged') readOnlyServedStale = true
|
||||
stubServed.add(source.path)
|
||||
} else if (cached && (readOnly || (action.action === 'unchanged' && (cached.failed || !cachedFileNeedsProviderReparse(providerName, source.path, cached))))) {
|
||||
if (readOnly && action.action !== 'unchanged') readOnlyServedStale = true
|
||||
unchangedSources.push({ source, cached })
|
||||
|
|
@ -3485,8 +3535,14 @@ export async function parseProviderSources(
|
|||
const wslHomesForOrphans = refreshWslHomesForOrphans(Object.keys(section.files), allDiscoveredFiles)
|
||||
|
||||
if (readOnly) {
|
||||
for (const [path, cached] of Object.entries(section.files)) {
|
||||
for (const [path, cached] of cacheEntriesInLoadOrder(diskCache, providerName)) {
|
||||
if (allDiscoveredFiles.has(path)) continue
|
||||
if (isCacheStub(cached)) {
|
||||
servedSources.push({ provider: providerName, path, project: providerName })
|
||||
allDiscoveredFiles.add(path)
|
||||
stubServed.add(path)
|
||||
continue
|
||||
}
|
||||
servedSources.push({
|
||||
provider: providerName,
|
||||
path,
|
||||
|
|
@ -3522,6 +3578,7 @@ export async function parseProviderSources(
|
|||
}
|
||||
}
|
||||
}
|
||||
for (const path of stubServed) for (const turn of stubs?.get(path)?.keys ?? []) for (const key of turn) parserDedup.add(key)
|
||||
|
||||
// Codex rollouts are the bulk of a cold parse (multi-GB against Claude's
|
||||
// hundreds of MB), so whole-file decodes go to worker threads. A file the
|
||||
|
|
@ -3814,6 +3871,11 @@ export async function parseProviderSources(
|
|||
delete section.files[cachedPath]
|
||||
markCacheDirty(diskCache, providerName, cachedPath)
|
||||
}
|
||||
// Never a WSL path (those always load in full).
|
||||
for (const [cachedPath] of stubs ?? []) {
|
||||
if (allDiscoveredFiles.has(cachedPath) || sources.length === 0) continue
|
||||
evictCacheStub(diskCache, providerName, cachedPath)
|
||||
}
|
||||
}
|
||||
if (evictedCodexResultCache) await flushCodexCache()
|
||||
|
||||
|
|
@ -4081,7 +4143,14 @@ export async function parseProviderSources(
|
|||
|
||||
for (const source of servedSources) {
|
||||
const cachedFile = section.files[source.path]
|
||||
if (!cachedFile) continue
|
||||
if (!cachedFile) {
|
||||
// Out of range: it adds nothing but the keys a full entry would claim here.
|
||||
for (const keys of stubs?.get(source.path)?.keys ?? []) {
|
||||
if (keys.some(k => seenKeys.has(k))) continue
|
||||
for (const k of keys) seenKeys.add(k)
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
for (const rawTurn of cachedFile.turns) {
|
||||
const turn = reconcileCopilotCalls(rawTurn)
|
||||
|
|
@ -5558,6 +5627,33 @@ const dateFloorSkippedProviders = new Set<string>()
|
|||
type SinglePassScope = { range: DateRange; parses: Map<string, Promise<ProjectSummary[]>> }
|
||||
let singlePassScope: SinglePassScope | null = null
|
||||
|
||||
// A command that asks for several ranges one after another (status: today,
|
||||
// then the month) declares their union here, so the session cache loads once
|
||||
// and every later request finds its range already held in full. Only the load
|
||||
// range widens: each request keeps its own month scope and summaries, and a
|
||||
// cached file the wider range holds in full instead of as a stub contributes
|
||||
// exactly what its stub would.
|
||||
let loadWindow: DateRange | null = null
|
||||
|
||||
export async function withLoadWindow<T>(range: DateRange, fn: () => Promise<T>): Promise<T> {
|
||||
const outer = loadWindow
|
||||
loadWindow = range
|
||||
try {
|
||||
return await fn()
|
||||
} finally {
|
||||
loadWindow = outer
|
||||
}
|
||||
}
|
||||
|
||||
function withinLoadWindow(scope: CacheLoadScope): CacheLoadScope {
|
||||
if (!loadWindow) return scope
|
||||
return {
|
||||
...scope,
|
||||
startMs: Math.min(scope.startMs!, loadWindow.start.getTime()),
|
||||
endMs: Math.max(scope.endMs!, loadWindow.end.getTime()),
|
||||
}
|
||||
}
|
||||
|
||||
export async function withSinglePassParse<T>(range: DateRange, fn: () => Promise<T>): Promise<T> {
|
||||
const outer = singlePassScope
|
||||
singlePassScope = { range, parses: new Map() }
|
||||
|
|
@ -5780,7 +5876,7 @@ function canServeCompleteSnapshot(cache: SessionCache, providerFilter?: string,
|
|||
const sections = providerFilter && providerFilter !== 'all'
|
||||
? ([[providerFilter, cache.providers[providerFilter]]] as const).filter((entry): entry is readonly [string, ProviderSection] => entry[1] != null)
|
||||
: Object.entries(cache.providers)
|
||||
return sections.some(([, section]) => Object.keys(section.files).length > 0)
|
||||
return sections.some(([name]) => hasCachedEntries(cache, name))
|
||||
&& sections.every(([name, section]) => section.envFingerprint === computeEnvFingerprint(name))
|
||||
}
|
||||
|
||||
|
|
@ -5836,10 +5932,10 @@ async function parseAllSessionsInCacheScope(dateRange?: DateRange, providerFilte
|
|||
// unaffected (a suppressed duplicate contributes nothing either way), but for
|
||||
// a proxied key emitted under two providers the attribution can land on a
|
||||
// different provider than a full load would pick.
|
||||
const loadScope = dateRange ? monthScopeForRange(dateRange.start, dateRange.end) : undefined
|
||||
const loadScope = dateRange ? withinLoadWindow(monthScopeForRange(dateRange.start, dateRange.end)) : undefined
|
||||
const rangeStartMs = dateRange?.start.getTime()
|
||||
const cacheLoadStarted = performance.now()
|
||||
let diskCache = await loadCache(loadScope)
|
||||
let diskCache: SessionCache | null = await loadCache(loadScope)
|
||||
await cleanupOrphanedTempFiles()
|
||||
if (process.env['CODEBURN_VERBOSE'] === '1') {
|
||||
process.stderr.write(`codeburn: startup timing cache-load=${(performance.now() - cacheLoadStarted).toFixed(1)}ms complete=${isCacheComplete(diskCache, providerFilter, rangeStartMs)}\n`)
|
||||
|
|
@ -5876,7 +5972,6 @@ async function parseAllSessionsInCacheScope(dateRange?: DateRange, providerFilte
|
|||
// A complete cache refresh is a strict read/reconcile/parse/save transaction.
|
||||
// Keep the snapshot loaded before acquisition: timeout/unavailable paths serve
|
||||
// exactly this complete snapshot and never mutate or invalidate the holder.
|
||||
const priorSnapshot = diskCache
|
||||
// Heartbeat the WAIT too, not just the parse behind it. This is the one place
|
||||
// a healthy process is deliberately idle for a long stretch, and the desktop
|
||||
// and menubar watchdogs read silence as a dead child - which is how a waiter
|
||||
|
|
@ -5894,8 +5989,11 @@ async function parseAllSessionsInCacheScope(dateRange?: DateRange, providerFilte
|
|||
process.stderr.write(`codeburn: startup timing refresh-lock=${(performance.now() - refreshWaitStarted).toFixed(1)}ms outcome=${refresh.outcome}\n`)
|
||||
}
|
||||
if (refresh.outcome === 'timed-out' || refresh.outcome === 'unavailable') {
|
||||
return runParse(key, priorSnapshot, dateRange, providerFilter, { readOnly: true, burstSig, parseStartedAt })
|
||||
return runParse(key, diskCache, dateRange, providerFilter, { readOnly: true, burstSig, parseStartedAt })
|
||||
}
|
||||
// Every other outcome reloads, so the snapshot goes first: two views of a
|
||||
// large cache are never held at once.
|
||||
diskCache = null
|
||||
if (refresh.outcome === 'completed-by-other') {
|
||||
return runParse(key, await loadCache(loadScope), dateRange, providerFilter, { readOnly: true, burstSig, parseStartedAt })
|
||||
}
|
||||
|
|
@ -6077,7 +6175,7 @@ async function runParseInner(
|
|||
// Skip if filtered to a different provider
|
||||
if (providerFilter && providerFilter !== 'all' && providerFilter !== providerName) continue
|
||||
const section = diskCache.providers[providerName]
|
||||
if (!section || Object.keys(section.files).length === 0) continue
|
||||
if (!section || !hasCachedEntries(diskCache, providerName)) continue
|
||||
const hasWslCachedSources = Object.keys(section.files).some(isWslUncPath)
|
||||
// Use the persisted durable flag (set by parseProviderSources when it first
|
||||
// processes a durableSources provider) OR the static DURABLE_PROVIDER_NAMES
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ import { readFile, stat, open, rename, unlink, readdir, mkdir, rm } from 'fs/pro
|
|||
import { existsSync, readFileSync, unlinkSync } from 'fs'
|
||||
import { createHash, randomBytes } from 'crypto'
|
||||
import { join } from 'path'
|
||||
import { StringDecoder } from 'string_decoder'
|
||||
|
||||
import { getCodeburnCacheDir } from './cache-dir.js'
|
||||
import { acquireCacheRefreshLock, releaseOwnedRefreshLocksForExit } from './cache-refresh-lock.js'
|
||||
|
|
@ -329,6 +330,13 @@ export const PROVIDER_ENV_VARS: Record<string, string[]> = {
|
|||
// disappear — they are preserved so month-to-date totals never drop.
|
||||
export const DURABLE_PROVIDER_NAMES: ReadonlySet<string> = new Set(['copilot'])
|
||||
|
||||
// Read in full by a ranged load, like the durable providers: their parse reads
|
||||
// cached entries no report shows. hermes seeds its ledger cursors from every
|
||||
// cached call, quickdesk has durable sources before its section is stamped
|
||||
// durable, and antigravity, devin and gemini decide a re-parse from the cached
|
||||
// entry itself (cachedFileNeedsProviderReparse in parser.ts).
|
||||
const FULL_LOAD_PROVIDER_NAMES: ReadonlySet<string> = new Set(['hermes', 'quickdesk', 'antigravity', 'devin', 'gemini'])
|
||||
|
||||
// Estimated-cost surfacing (#639): providers that set `costIsEstimated` carry a
|
||||
// `-est-cost` suffix (or a new entry) so their already-cached sessions reparse
|
||||
// once and the flag lands, instead of silently reading as measured. Copilot
|
||||
|
|
@ -580,6 +588,28 @@ export function cacheBucketMonth(file: CachedFile): string {
|
|||
return cacheFileSpan(file).bucket
|
||||
}
|
||||
|
||||
/** 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. */
|
||||
export type CacheStub = {
|
||||
fingerprint: FileFingerprint
|
||||
bucket: string
|
||||
until: string
|
||||
hasPr: boolean
|
||||
/** Dedup keys per turn, in turn order. */
|
||||
keys: string[][]
|
||||
/** Oldest and newest call timestamp (ms). */
|
||||
lo: number
|
||||
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.
|
||||
type CacheState = {
|
||||
|
|
@ -597,6 +627,19 @@ type CacheState = {
|
|||
bucketOf: Map<string, string>
|
||||
/** The load scope this cache was read under, for the cross-request memo. */
|
||||
scope: string
|
||||
/** The shard directory this cache was read from. */
|
||||
dir: string
|
||||
/** provider -> stubbed paths, valid only while `section` is still the provider's section. */
|
||||
stubs: Map<string, { section: ProviderSection; byPath: Map<string, CacheStub> }>
|
||||
/** 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. */
|
||||
seq: WeakMap<CachedFile, number>
|
||||
/** 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<string>
|
||||
}
|
||||
const cacheStates = new WeakMap<SessionCache, CacheState>()
|
||||
|
||||
|
|
@ -611,6 +654,11 @@ function stateOf(cache: SessionCache): CacheState {
|
|||
fingerprints: new Map(),
|
||||
bucketOf: new Map(),
|
||||
scope: 'all',
|
||||
dir: sessionCacheDir(),
|
||||
stubs: new Map(),
|
||||
seq: new WeakMap(),
|
||||
covered: null,
|
||||
legacy: new Set(),
|
||||
}
|
||||
cacheStates.set(cache, state)
|
||||
}
|
||||
|
|
@ -651,6 +699,107 @@ export function isCacheDirty(cache: SessionCache): boolean {
|
|||
return stateOf(cache).dirty
|
||||
}
|
||||
|
||||
function stubsOf(state: CacheState, cache: SessionCache, provider: string): Map<string, CacheStub> | undefined {
|
||||
const held = state.stubs.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<string, CacheStub> | undefined {
|
||||
return stubsOf(stateOf(cache), cache, provider)
|
||||
}
|
||||
|
||||
/** True when `provider` has any cached entry, full or stubbed. */
|
||||
export function hasCachedEntries(cache: SessionCache, provider: string): boolean {
|
||||
const section = cache.providers[provider]
|
||||
if (!section) return false
|
||||
for (const _ in section.files) return true
|
||||
return (stubsOf(stateOf(cache), cache, provider)?.size ?? 0) > 0
|
||||
}
|
||||
|
||||
/** Drop a stub whose source is gone, exactly as deleting its entry would. */
|
||||
export function evictCacheStub(cache: SessionCache, provider: string, path: string): void {
|
||||
stubsOf(stateOf(cache), cache, provider)?.delete(path)
|
||||
markCacheDirty(cache, provider, path)
|
||||
}
|
||||
|
||||
function entriesInLoadOrder(state: CacheState, section: ProviderSection, stubs: Map<string, CacheStub> | undefined): Array<[string, CachedFile | CacheStub]> {
|
||||
if (!stubs) return Object.entries(section.files)
|
||||
const loaded: Array<[string, CachedFile | CacheStub, number]> = []
|
||||
const later: Array<[string, CachedFile]> = []
|
||||
for (const [path, file] of Object.entries(section.files)) {
|
||||
const seq = state.seq.get(file)
|
||||
if (seq === undefined) later.push([path, file])
|
||||
else loaded.push([path, file, seq])
|
||||
}
|
||||
for (const [path, stub] of stubs) loaded.push([path, stub, stub.seq])
|
||||
loaded.sort((a, b) => a[2] - b[2])
|
||||
return [...loaded.map(([path, entry]): [string, CachedFile | CacheStub] => [path, entry]), ...later]
|
||||
}
|
||||
|
||||
/** 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
|
||||
* written since in the order they were written. */
|
||||
export function cacheEntriesInLoadOrder(cache: SessionCache, provider: string): Array<[string, CachedFile | CacheStub]> {
|
||||
const section = cache.providers[provider]
|
||||
if (!section) return []
|
||||
const state = stateOf(cache)
|
||||
return entriesInLoadOrder(state, section, stubsOf(state, cache, provider))
|
||||
}
|
||||
|
||||
export function isCacheStub(entry: CachedFile | CacheStub): entry is CacheStub {
|
||||
return 'keys' in entry
|
||||
}
|
||||
|
||||
/** Load stubbed members in full, in place, at their load position. A member
|
||||
* that can no longer be read loses its stub, so the file re-parses like any
|
||||
* uncached one and the save drops its line. */
|
||||
export async function loadCacheStubs(cache: SessionCache, provider: string, paths: Iterable<string>): Promise<void> {
|
||||
const state = stateOf(cache)
|
||||
const stubs = stubsOf(state, cache, provider)
|
||||
const section = cache.providers[provider]
|
||||
if (!stubs || !section) return
|
||||
const byName = new Map<string, Array<[string, CacheStub]>>()
|
||||
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]])
|
||||
}
|
||||
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) {
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Widen what a memoized ranged load holds in full to [startMs, endMs] (and
|
||||
// whatever it already covered), loading the stubs that range can report on.
|
||||
async function coverRange(cache: SessionCache, startMs: number, endMs: number): Promise<void> {
|
||||
const state = stateOf(cache)
|
||||
const covered = state.covered
|
||||
if (!covered || (startMs >= covered.startMs && endMs <= covered.endMs)) return
|
||||
const from = Math.min(startMs, covered.startMs)
|
||||
const to = Math.max(endMs, covered.endMs)
|
||||
for (const provider of Object.keys(cache.providers)) {
|
||||
const stubs = stubsOf(state, cache, provider)
|
||||
if (!stubs) continue
|
||||
const wanted: string[] = []
|
||||
for (const [path, stub] of stubs) if (stub.hi >= from && stub.lo <= to) wanted.push(path)
|
||||
await loadCacheStubs(cache, provider, wanted)
|
||||
}
|
||||
state.covered = { startMs: from, endMs: to }
|
||||
}
|
||||
|
||||
/** True when a dirty bucket belongs to a provider whose cache entry is the ONLY
|
||||
* surviving record of that spend (see {@link DURABLE_PROVIDER_NAMES}): the
|
||||
* source can be pruned before the next publish, so such a window must never be
|
||||
|
|
@ -1024,11 +1173,12 @@ export async function isCacheCurrent(cache: SessionCache): Promise<boolean> {
|
|||
* widens this by one month BELOW `fromMonth` and none above (see
|
||||
* shardInScope): every cross-range carry in the report reads BACKWARDS from the
|
||||
* first in-range turn, never forwards, so there is nothing above the range to
|
||||
* reach for. */
|
||||
export type CacheLoadScope = { fromMonth: string; toMonth: string }
|
||||
* reach for. `startMs`/`endMs` are the range itself: a cached file the range
|
||||
* cannot report on is then held as a {@link CacheStub}, not in full. */
|
||||
export type CacheLoadScope = { fromMonth: string; toMonth: string; startMs?: number; endMs?: number }
|
||||
|
||||
export function monthScopeForRange(start: Date, end: Date): CacheLoadScope {
|
||||
return { fromMonth: monthKey(start.toISOString())!, toMonth: monthKey(end.toISOString())! }
|
||||
return { fromMonth: monthKey(start.toISOString())!, toMonth: monthKey(end.toISOString())!, startMs: start.getTime(), endMs: end.getTime() }
|
||||
}
|
||||
|
||||
function previousMonth(month: string): string {
|
||||
|
|
@ -1082,18 +1232,200 @@ async function readEnvelope(dir: string): Promise<CacheEnvelope | null> {
|
|||
}
|
||||
}
|
||||
|
||||
// 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 }
|
||||
|
||||
const SHARD_READ_CHUNK = 1 << 20
|
||||
|
||||
function splitMemberLine(line: string): [string, string] | null {
|
||||
if (line.charCodeAt(0) !== 34) return null
|
||||
let i = 1
|
||||
for (; i < line.length; i++) {
|
||||
const c = line.charCodeAt(i)
|
||||
if (c === 92) i++
|
||||
else if (c === 34) break
|
||||
}
|
||||
if (line.charCodeAt(i + 1) !== 58) return null
|
||||
const end = line.charCodeAt(line.length - 1) === 44 ? line.length - 1 : line.length
|
||||
return [JSON.parse(line.slice(0, i + 1)) as string, line.slice(i + 2, end)]
|
||||
}
|
||||
|
||||
function shardLine(text: string, offset: number, length: number): ShardLine {
|
||||
const split = splitMemberLine(text)
|
||||
if (!split) throw new Error('malformed shard line')
|
||||
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
|
||||
// malformed, as JSON.parse would.
|
||||
async function* shardLines(path: string): AsyncGenerator<ShardLine> {
|
||||
const handle = await open(path, 'r')
|
||||
try {
|
||||
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
|
||||
}
|
||||
// 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
|
||||
// RSS when members run to megabytes. A newline is always a character
|
||||
// boundary, so the decoder never holds bytes across one.
|
||||
const chunk = Buffer.allocUnsafe(SHARD_READ_CHUNK)
|
||||
const decoder = new StringDecoder('utf-8')
|
||||
let pos = 2
|
||||
let lineStart = 2
|
||||
let pending = ''
|
||||
for (;;) {
|
||||
const { bytesRead } = await handle.read(chunk, 0, SHARD_READ_CHUNK, pos)
|
||||
if (bytesRead === 0) break
|
||||
const view = chunk.subarray(0, bytesRead)
|
||||
let from = 0
|
||||
for (let nl = view.indexOf(10); nl !== -1; nl = view.indexOf(10, from)) {
|
||||
const text = pending + decoder.write(view.subarray(from, nl))
|
||||
pending = ''
|
||||
if (text === '}') return
|
||||
yield shardLine(text, lineStart, pos + nl - lineStart)
|
||||
from = nl + 1
|
||||
lineStart = pos + from
|
||||
}
|
||||
if (from < bytesRead) pending += decoder.write(view.subarray(from))
|
||||
pos += bytesRead
|
||||
}
|
||||
if (pending + decoder.end() !== '}') throw new Error('truncated shard')
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
}
|
||||
|
||||
function shardLineFile(line: ShardLine): unknown {
|
||||
return line.file ?? JSON.parse(line.value!)
|
||||
}
|
||||
|
||||
function shardLineText(line: ShardLine): string {
|
||||
return `${JSON.stringify(line.key)}:${line.value ?? JSON.stringify(line.file)}`
|
||||
}
|
||||
|
||||
// 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.
|
||||
async function* shardPayload(members: AsyncIterable<string> | Iterable<string>, positions?: Array<[number, number]>): AsyncGenerator<string> {
|
||||
let pos = 0
|
||||
let first = true
|
||||
for await (const member of members) {
|
||||
const head = first ? '{\n' : ',\n'
|
||||
first = false
|
||||
if (positions) {
|
||||
const length = Buffer.byteLength(member)
|
||||
positions.push([pos + 2, length])
|
||||
pos += 2 + length
|
||||
}
|
||||
yield head + member
|
||||
}
|
||||
yield first ? '{}' : '\n}'
|
||||
}
|
||||
|
||||
function* memberTexts(files: Record<string, CachedFile>): Generator<string> {
|
||||
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.
|
||||
async function loadShard(path: string): Promise<Record<string, CachedFile> | null> {
|
||||
try {
|
||||
const parsed = JSON.parse(await readFile(path, 'utf-8'))
|
||||
return validateFiles(parsed) ? parsed : null
|
||||
const files: Record<string, unknown> = {}
|
||||
for await (const line of shardLines(path)) files[line.key] = shardLineFile(line)
|
||||
return validateFiles(files) ? files : null
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
async function readMembersAt(path: string, wanted: Array<[string, number, number]>): Promise<Map<string, CachedFile>> {
|
||||
const out = new Map<string, CachedFile>()
|
||||
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)
|
||||
}
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
} catch { /* unreadable: the caller treats every wanted member as gone */ }
|
||||
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
|
||||
|
|
@ -1148,8 +1480,10 @@ export async function loadShardMemoized(dir: string, name: string): Promise<Reco
|
|||
let raw: string
|
||||
try {
|
||||
raw = await readFile(join(dir, name), 'utf-8')
|
||||
} catch {
|
||||
return null
|
||||
} catch (err) {
|
||||
// Past the string limit a shard can still be read a member at a time. It is
|
||||
// far over the memo budget, so it is not memoized.
|
||||
return err instanceof RangeError ? loadShard(join(dir, name)) : null
|
||||
}
|
||||
let files: Record<string, CachedFile>
|
||||
try {
|
||||
|
|
@ -1188,10 +1522,19 @@ export async function loadCache(scope?: CacheLoadScope): Promise<SessionCache> {
|
|||
if (!envelope) return afterMissingShardCache()
|
||||
const scopeKey = scope ? `${scope.fromMonth}..${scope.toMonth}` : 'all'
|
||||
if (cacheMemo && cacheMemo.dir === dir && cacheMemo.nonce === envelope.nonce
|
||||
&& (cacheMemo.scope === 'all' || cacheMemo.scope === scopeKey)) return cacheMemo.cache
|
||||
&& (cacheMemo.scope === 'all' || cacheMemo.scope === scopeKey)) {
|
||||
if (scope) await coverRange(cacheMemo.cache, scope.startMs ?? -Infinity, scope.endMs ?? Infinity)
|
||||
return cacheMemo.cache
|
||||
}
|
||||
// Released before the new view is read, so the two are never held at once.
|
||||
cacheMemo = null
|
||||
|
||||
const cache: SessionCache = { version: CACHE_VERSION, providers: {}, complete: envelope.complete === true }
|
||||
const state = stateOf(cache)
|
||||
state.dir = dir
|
||||
const ranged = scope?.startMs !== undefined && scope.endMs !== undefined
|
||||
? { startMs: scope.startMs, endMs: scope.endMs }
|
||||
: null
|
||||
const reads: Promise<void>[] = []
|
||||
for (const [provider, meta] of Object.entries(envelope.providers)) {
|
||||
const section: ProviderSection = {
|
||||
|
|
@ -1213,33 +1556,62 @@ export async function loadCache(scope?: CacheLoadScope): Promise<SessionCache> {
|
|||
// 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<string> | null = full ? null : new Set()
|
||||
const stubs = loaded && ranged && !FULL_LOAD_PROVIDER_NAMES.has(provider) ? new Map<string, CacheStub>() : 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.
|
||||
const pending: { bucket: string; files: Promise<Record<string, CachedFile> | null> }[] = []
|
||||
// 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<Record<string, CachedFile> | 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, files: loadShardMemoized(dir, ref.name) })
|
||||
pending.push({ bucket, name: ref.name, files: stubs ? null : loadShardMemoized(dir, ref.name) })
|
||||
}
|
||||
reads.push((async () => {
|
||||
for (const { bucket, files: read } of pending) {
|
||||
const files = await read
|
||||
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 (!files) { markBucketDirty(state, provider, bucket); continue }
|
||||
for (const [path, file] of Object.entries(files)) {
|
||||
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)
|
||||
if (section.files[path]!.fingerprint.mtimeMs >= file.fingerprint.mtimeMs) continue
|
||||
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)
|
||||
section.files[path] = file
|
||||
if (isCacheStub(entry)) {
|
||||
entry.seq = at
|
||||
stubs!.set(path, entry)
|
||||
} else {
|
||||
if (stubs) state.seq.set(entry, at)
|
||||
section.files[path] = entry
|
||||
}
|
||||
}
|
||||
}
|
||||
})())
|
||||
|
|
@ -1344,11 +1716,22 @@ function shardFileName(provider: string, bucket: string): string {
|
|||
// 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): Promise<void> {
|
||||
async function writeFileAtomic(finalPath: string, payload: string | AsyncIterable<string>): Promise<void> {
|
||||
const tempPath = `${finalPath}.${randomBytes(8).toString('hex')}.tmp`
|
||||
const handle = await open(tempPath, 'w', 0o600)
|
||||
try {
|
||||
await handle.writeFile(payload, { encoding: 'utf-8' })
|
||||
if (typeof payload === 'string') {
|
||||
await handle.writeFile(payload, { encoding: 'utf-8' })
|
||||
} else {
|
||||
let pending = ''
|
||||
for await (const chunk of payload) {
|
||||
pending += chunk
|
||||
if (pending.length < SHARD_READ_CHUNK) continue
|
||||
await handle.writeFile(pending, { encoding: 'utf-8' })
|
||||
pending = ''
|
||||
}
|
||||
if (pending) await handle.writeFile(pending, { encoding: 'utf-8' })
|
||||
}
|
||||
await handle.sync()
|
||||
} finally {
|
||||
await handle.close()
|
||||
|
|
@ -1370,10 +1753,16 @@ async function writeFileAtomic(finalPath: string, payload: string): Promise<void
|
|||
}
|
||||
}
|
||||
|
||||
function bucketFiles(section: ProviderSection): { groups: Map<string, Record<string, CachedFile>>; until: Map<string, string> } {
|
||||
// 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<string, Record<string, CachedFile>>; until: Map<string, string> } {
|
||||
const groups = new Map<string, Record<string, CachedFile>>()
|
||||
const until = new Map<string, string>()
|
||||
for (const [path, file] of Object.entries(section.files)) {
|
||||
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) }
|
||||
|
|
@ -1409,6 +1798,13 @@ type ProviderPlan = {
|
|||
/** bucket -> the shard name the merge was built from, for the retry below. */
|
||||
mergedFrom: Map<string, string | undefined>
|
||||
refs: Record<string, ShardRef>
|
||||
stubs: Map<string, CacheStub> | undefined
|
||||
/** Buckets holding a stub, which memory alone cannot rewrite. */
|
||||
partial: Set<string>
|
||||
/** 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<string>
|
||||
}
|
||||
|
||||
// Surrender the event loop without microtask overhead. Used inside saveCache
|
||||
|
|
@ -1428,11 +1824,68 @@ export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Pr
|
|||
|
||||
const writeShard = async (provider: string, bucket: string, files: Record<string, CachedFile>): Promise<ShardRef> => {
|
||||
const name = shardFileName(provider, bucket)
|
||||
await writeFileAtomic(join(dir, name), JSON.stringify(files))
|
||||
await writeFileAtomic(join(dir, name), shardPayload(memberTexts(files)))
|
||||
written.add(name)
|
||||
return { name, until: untilMonth(files) }
|
||||
}
|
||||
|
||||
// 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<ShardRef> => {
|
||||
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<string> {
|
||||
const done = new Set<string>()
|
||||
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
|
||||
}
|
||||
}
|
||||
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 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 }
|
||||
}
|
||||
|
||||
// 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<ShardRef> => {
|
||||
|
|
@ -1464,8 +1917,11 @@ export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Pr
|
|||
// null here in practice; the guard is what makes that safe to rely on.
|
||||
const priorFingerprint = state.fingerprints.get(provider)
|
||||
const reset = priorFingerprint !== undefined && priorFingerprint !== section.envFingerprint
|
||||
const { groups } = bucketFiles(section)
|
||||
const plan: ProviderPlan = { section, groups, loaded, priorRefs, reset, moved: new Set(), deferred: [], mergedFrom: new Map(), refs: {} }
|
||||
const stubs = stubsOf(state, cache, provider)
|
||||
const { groups } = bucketFiles(entriesInLoadOrder(state, section, stubs))
|
||||
const partial = new Set<string>()
|
||||
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
|
||||
|
|
@ -1489,7 +1945,8 @@ export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Pr
|
|||
// ANOTHER process may have republished that shard since, unlinking the
|
||||
// file we are about to name — so reuse is conditional on the file still
|
||||
// being there, and a vanished one is rewritten from memory.
|
||||
if (prior && !isBucketDirty(state, provider, bucket) && existsSync(join(dir, prior.name))) {
|
||||
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
|
||||
}
|
||||
|
|
@ -1498,7 +1955,9 @@ export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Pr
|
|||
// merged back in or the save would drop them. Deferred to phase two so
|
||||
// the read happens against the CURRENT shard, not a stale name.
|
||||
if (loaded && !loaded.has(bucket) && prior) { plan.deferred.push(bucket); continue }
|
||||
plan.refs[bucket] = await writeShard(provider, bucket, files)
|
||||
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
|
||||
|
|
@ -1591,7 +2050,9 @@ export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Pr
|
|||
// A carried month whose file vanished and whose content was never in
|
||||
// memory cannot be rewritten; dropping the reference is the only honest
|
||||
// option, and the sweep retires the name.
|
||||
if (files) shards[bucket] = await writeShard(provider, bucket, files)
|
||||
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,
|
||||
|
|
@ -1628,12 +2089,26 @@ export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Pr
|
|||
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
|
||||
|
|
@ -1779,7 +2254,7 @@ export type ReconcileAction =
|
|||
|
||||
export function reconcileFile(
|
||||
current: FileFingerprint,
|
||||
cached: CachedFile | undefined,
|
||||
cached: Pick<CachedFile, 'fingerprint' | 'lastCompleteLineOffset'> | undefined,
|
||||
): ReconcileAction {
|
||||
if (!cached) return { action: 'new' }
|
||||
|
||||
|
|
|
|||
140
tests/bounded-load-parse.test.ts
Normal file
140
tests/bounded-load-parse.test.ts
Normal file
|
|
@ -0,0 +1,140 @@
|
|||
// End to end through parseProviderSources: a ranged parse holds out-of-range
|
||||
// rollouts as stubs, and a stub must still do everything its entry would
|
||||
// outside the summaries (claim its dedup keys in serve order, be evicted from
|
||||
// disk when its source is deleted). Codex captures CODEX_HOME when its module
|
||||
// is first imported, hence the env before the dynamic imports.
|
||||
import { afterAll, beforeEach, describe, expect, it } from 'vitest'
|
||||
import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from 'fs'
|
||||
import { readFile } from 'fs/promises'
|
||||
import { join } from 'path'
|
||||
import { tmpdir } from 'os'
|
||||
import type { DateRange } from '../src/types.js'
|
||||
|
||||
const tmpDir = mkdtempSync(join(tmpdir(), 'bounded-load-'))
|
||||
const savedCodexHome = process.env['CODEX_HOME']
|
||||
process.env['CODEX_HOME'] = join(tmpDir, 'codex')
|
||||
|
||||
const { cacheStubs, clearLoadCacheMemo, loadCache, markCacheDirty, monthScopeForRange, saveCache, sessionCacheDir } = await import('../src/session-cache.js')
|
||||
const { clearSessionCache, parseAllSessions, withColdFirstPaintFloor } = await import('../src/parser.js')
|
||||
const { clearCodexMemCaches } = await import('../src/codex-cache.js')
|
||||
|
||||
const ENV_KEYS = ['CODEBURN_CACHE_DIR', 'CLAUDE_CONFIG_DIR', 'CLAUDE_CONFIG_DIRS', 'CODEBURN_DESKTOP_SESSIONS_DIR', 'CODEBURN_CACHE_SCOPE'] as const
|
||||
const saved = Object.fromEntries(ENV_KEYS.map(k => [k, process.env[k]]))
|
||||
let run = 0
|
||||
|
||||
function rollout(day: string, sid: string, startIso: string, n: number): string {
|
||||
const dir = join(tmpDir, 'codex', 'sessions', '2099', day.slice(0, 2), day.slice(3))
|
||||
mkdirSync(dir, { recursive: true })
|
||||
const path = join(dir, `rollout-2099-${day}T10-00-00-${sid}.jsonl`)
|
||||
const lines = [
|
||||
{ type: 'session_meta', timestamp: startIso, payload: { id: sid, session_id: sid, cwd: `/work/${sid}`, originator: 'codex_cli_rs', model: 'gpt-5.3-codex', cli_version: '0.50.0' } },
|
||||
{ type: 'turn_context', timestamp: startIso, payload: { cwd: `/work/${sid}`, model: 'gpt-5.3-codex' } },
|
||||
]
|
||||
const total = { input_tokens: 0, cached_input_tokens: 0, output_tokens: 0, reasoning_output_tokens: 0, total_tokens: 0 }
|
||||
for (let i = 0; i < n; i++) {
|
||||
const ts = new Date(Date.parse(startIso) + i * 60_000).toISOString()
|
||||
const last = { input_tokens: 1000 + i, cached_input_tokens: 100, output_tokens: 50 + i, reasoning_output_tokens: 0, total_tokens: 1050 + 2 * i }
|
||||
for (const k of Object.keys(total) as Array<keyof typeof total>) total[k] += last[k]
|
||||
lines.push({ type: 'response_item', timestamp: ts, payload: { type: 'message', role: 'user', content: [{ type: 'input_text', text: `step ${i}` }] } } as never)
|
||||
lines.push({ type: 'event_msg', timestamp: ts, payload: { type: 'token_count', info: { last_token_usage: last, total_token_usage: { ...total } } } } as never)
|
||||
}
|
||||
writeFileSync(path, lines.map(l => JSON.stringify(l)).join('\n') + '\n')
|
||||
return path
|
||||
}
|
||||
|
||||
const may1: DateRange = { start: new Date('2099-05-01T00:00:00.000Z'), end: new Date('2099-05-01T23:59:59.999Z') }
|
||||
const calls = (projects: Awaited<ReturnType<typeof parseAllSessions>>): number => projects.reduce((n, p) => n + p.totalApiCalls, 0)
|
||||
const cost = (projects: Awaited<ReturnType<typeof parseAllSessions>>): number => projects.reduce((n, p) => n + p.totalCostUSD, 0)
|
||||
|
||||
beforeEach(() => {
|
||||
rmSync(join(tmpDir, 'codex'), { recursive: true, force: true })
|
||||
process.env['CODEBURN_CACHE_DIR'] = join(tmpDir, `cache-${run++}`)
|
||||
process.env['CLAUDE_CONFIG_DIR'] = join(tmpDir, 'no-claude')
|
||||
process.env['CODEBURN_DESKTOP_SESSIONS_DIR'] = join(tmpDir, 'no-desktop')
|
||||
delete process.env['CLAUDE_CONFIG_DIRS']
|
||||
delete process.env['CODEBURN_CACHE_SCOPE']
|
||||
clearSessionCache()
|
||||
clearLoadCacheMemo()
|
||||
clearCodexMemCaches()
|
||||
})
|
||||
|
||||
afterAll(() => {
|
||||
for (const k of ENV_KEYS) { if (saved[k] === undefined) delete process.env[k]; else process.env[k] = saved[k] }
|
||||
if (savedCodexHome === undefined) delete process.env['CODEX_HOME']
|
||||
else process.env['CODEX_HOME'] = savedCodexHome
|
||||
rmSync(tmpDir, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
async function reparse(range: DateRange, scope?: 'all'): Promise<Awaited<ReturnType<typeof parseAllSessions>>> {
|
||||
clearSessionCache()
|
||||
clearLoadCacheMemo()
|
||||
clearCodexMemCaches()
|
||||
if (scope) process.env['CODEBURN_CACHE_SCOPE'] = scope
|
||||
else delete process.env['CODEBURN_CACHE_SCOPE']
|
||||
return parseAllSessions(range, 'codex')
|
||||
}
|
||||
|
||||
describe('bounded load through the provider parse', () => {
|
||||
it('replays a stub\'s dedup keys in serve order, exactly as its full entry would', async () => {
|
||||
const old = rollout('04-10', 'aaold', '2099-04-10T10:00:00.000Z', 3)
|
||||
const today = rollout('05-01', 'bbtoday', '2099-05-01T10:00:00.000Z', 2)
|
||||
await parseAllSessions(undefined, 'codex')
|
||||
|
||||
// Give the in-range file a turn that carries one of the old file's keys, as
|
||||
// a file parsed before the other was cached can.
|
||||
clearLoadCacheMemo()
|
||||
const full = await loadCache()
|
||||
const section = full.providers['codex']!
|
||||
const oldKey = section.files[old]!.turns[0]!.calls[0]!.deduplicationKey
|
||||
section.files[today]!.turns[1]!.calls[0]!.deduplicationKey = oldKey
|
||||
markCacheDirty(full, 'codex', today)
|
||||
expect(await saveCache(full)).toBe(true)
|
||||
|
||||
clearLoadCacheMemo()
|
||||
const scoped = await loadCache(monthScopeForRange(may1.start, may1.end))
|
||||
expect([...cacheStubs(scoped, 'codex')!.keys()]).toEqual([old])
|
||||
|
||||
const bounded = await reparse(may1)
|
||||
const reference = await reparse(may1, 'all')
|
||||
expect(calls(bounded)).toBe(1)
|
||||
expect(calls(bounded)).toBe(calls(reference))
|
||||
expect(cost(bounded)).toBe(cost(reference))
|
||||
|
||||
// The first-paint snapshot serves every cached entry as an orphan, in load
|
||||
// order: the stub has to take its full entry's place in that order too.
|
||||
const snapshot = async (scope?: 'all') => {
|
||||
clearSessionCache(); clearLoadCacheMemo(); clearCodexMemCaches()
|
||||
if (scope) process.env['CODEBURN_CACHE_SCOPE'] = scope
|
||||
else delete process.env['CODEBURN_CACHE_SCOPE']
|
||||
return (await withColdFirstPaintFloor(may1.start, () => parseAllSessions(may1, 'codex'), true, true)).result
|
||||
}
|
||||
const boundedSnapshot = await snapshot()
|
||||
expect(calls(boundedSnapshot)).toBe(1)
|
||||
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 () => {
|
||||
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)
|
||||
rollout('05-01', 'bbtoday', '2099-05-01T10:00:00.000Z', 2)
|
||||
await parseAllSessions(undefined, 'codex')
|
||||
const aprilShard = async (): Promise<string> => {
|
||||
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 before = await aprilShard()
|
||||
const line = (text: string, path: string) => text.split('\n').find(l => l.startsWith(JSON.stringify(path) + ':'))?.replace(/,$/, '')
|
||||
expect(line(before, doomed)).toBeDefined()
|
||||
|
||||
clearLoadCacheMemo()
|
||||
const scoped = await loadCache(monthScopeForRange(may1.start, may1.end))
|
||||
expect([...cacheStubs(scoped, 'codex')!.keys()].sort()).toEqual([doomed, kept].sort())
|
||||
|
||||
rmSync(doomed)
|
||||
await reparse(may1)
|
||||
const after = await aprilShard()
|
||||
expect(after).not.toContain(doomed)
|
||||
expect(line(after, kept)).toBe(line(before, kept))
|
||||
expect(Object.keys(JSON.parse(after))).toEqual([kept])
|
||||
})
|
||||
})
|
||||
67
tests/codex-projects-index.test.ts
Normal file
67
tests/codex-projects-index.test.ts
Normal file
|
|
@ -0,0 +1,67 @@
|
|||
// Codex discovery needs only each rollout's project, so it reads a small
|
||||
// index flushCodexCache writes beside the results file instead of the whole
|
||||
// (hundreds-of-MB) results cache.
|
||||
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
|
||||
import { mkdtempSync, rmSync, writeFileSync, existsSync, readdirSync } from 'fs'
|
||||
import { tmpdir } from 'os'
|
||||
import { join } from 'path'
|
||||
|
||||
import {
|
||||
clearCodexMemCaches,
|
||||
codexCacheFileName,
|
||||
codexProjectsFileName,
|
||||
fingerprintFile,
|
||||
flushCodexCache,
|
||||
getCachedCodexProject,
|
||||
withCodexCacheDirectory,
|
||||
writeCachedCodexResults,
|
||||
} from '../src/codex-cache.js'
|
||||
|
||||
let dir: string
|
||||
|
||||
beforeEach(() => {
|
||||
dir = mkdtempSync(join(tmpdir(), 'codex-projects-'))
|
||||
clearCodexMemCaches()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
clearCodexMemCaches()
|
||||
rmSync(dir, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
describe('codex projects index', () => {
|
||||
it('serves discovery without the results file', async () => {
|
||||
const rollout = join(dir, 'rollout-2099-05-01T10-00-00-a.jsonl')
|
||||
writeFileSync(rollout, '{}\n')
|
||||
const fp = (await fingerprintFile(rollout))!
|
||||
await withCodexCacheDirectory(dir, async () => {
|
||||
await writeCachedCodexResults(rollout, 'indexed-project', [], fp)
|
||||
await flushCodexCache()
|
||||
})
|
||||
expect(existsSync(join(dir, codexProjectsFileName()))).toBe(true)
|
||||
|
||||
clearCodexMemCaches()
|
||||
writeFileSync(join(dir, codexCacheFileName()), 'not the results cache')
|
||||
expect(await withCodexCacheDirectory(dir, () => getCachedCodexProject(rollout))).toBe('indexed-project')
|
||||
|
||||
// An index that lags the file (it grew since) only misses.
|
||||
writeFileSync(rollout, '{}\n{}\n')
|
||||
clearCodexMemCaches()
|
||||
expect(await withCodexCacheDirectory(dir, () => getCachedCodexProject(rollout))).toBeNull()
|
||||
expect(readdirSync(dir).filter(f => f.endsWith('.tmp'))).toEqual([])
|
||||
})
|
||||
|
||||
it('builds the index from the results file when it is missing', async () => {
|
||||
const rollout = join(dir, 'rollout-2099-05-01T10-00-00-b.jsonl')
|
||||
writeFileSync(rollout, '{}\n')
|
||||
const fp = (await fingerprintFile(rollout))!
|
||||
await withCodexCacheDirectory(dir, async () => {
|
||||
await writeCachedCodexResults(rollout, 'from-results', [], fp)
|
||||
await flushCodexCache()
|
||||
})
|
||||
rmSync(join(dir, codexProjectsFileName()))
|
||||
clearCodexMemCaches()
|
||||
expect(await withCodexCacheDirectory(dir, () => getCachedCodexProject(rollout))).toBe('from-results')
|
||||
expect(existsSync(join(dir, codexProjectsFileName()))).toBe(true)
|
||||
})
|
||||
})
|
||||
167
tests/session-cache-bounded-load.test.ts
Normal file
167
tests/session-cache-bounded-load.test.ts
Normal file
|
|
@ -0,0 +1,167 @@
|
|||
// 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.
|
||||
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
|
||||
import { mkdir, readFile, rm, writeFile } from 'fs/promises'
|
||||
import { existsSync } from 'fs'
|
||||
import { tmpdir } from 'os'
|
||||
import { join } from 'path'
|
||||
|
||||
import {
|
||||
CACHE_VERSION,
|
||||
cacheStubs,
|
||||
clearLoadCacheMemo,
|
||||
computeEnvFingerprint,
|
||||
loadCache,
|
||||
markCacheDirty,
|
||||
monthScopeForRange,
|
||||
saveCache,
|
||||
sessionCacheDir,
|
||||
type CachedFile,
|
||||
type SessionCache,
|
||||
} from '../src/session-cache.js'
|
||||
|
||||
let TMP_DIR: string
|
||||
|
||||
beforeEach(async () => {
|
||||
TMP_DIR = join(tmpdir(), `codeburn-bounded-${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 () => {
|
||||
if (existsSync(TMP_DIR)) await rm(TMP_DIR, { recursive: true })
|
||||
})
|
||||
|
||||
function fileAt(day: string, key: string, extraTurns = 0): CachedFile {
|
||||
const turn = (i: number): CachedFile['turns'][number] => {
|
||||
const timestamp = `2026-05-${day}T1${i}: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}`,
|
||||
}],
|
||||
}
|
||||
}
|
||||
return { fingerprint: { dev: 1, ino: key.length, mtimeMs: 3, sizeBytes: 4 }, lastCompleteLineOffset: 9, mcpInventory: [], turns: Array.from({ length: 1 + extraTurns }, (_, i) => turn(i)) }
|
||||
}
|
||||
|
||||
const day20 = monthScopeForRange(new Date('2026-05-20T00:00:00.000Z'), new Date('2026-05-20T23:59:59.999Z'))
|
||||
|
||||
async function publish(files: Record<string, CachedFile>): Promise<void> {
|
||||
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 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(/,$/, '')
|
||||
}
|
||||
|
||||
describe('bounded ranged load', () => {
|
||||
it('rewrites a bucket holding stubs by copying their lines, in 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 cache = await loadCache(day20)
|
||||
const section = cache.providers['claude']!
|
||||
expect(Object.keys(section.files)).toEqual(['/p/x.jsonl'])
|
||||
expect([...cacheStubs(cache, 'claude')!.keys()]).toEqual(['/p/y.jsonl', '/p/w.jsonl'])
|
||||
|
||||
// What a re-parse does: delete, then install; plus one new file.
|
||||
delete section.files['/p/x.jsonl']
|
||||
section.files['/p/x.jsonl'] = fileAt('20', 'x', 1)
|
||||
section.files['/p/z.jsonl'] = fileAt('21', 'z')
|
||||
markCacheDirty(cache, 'claude', '/p/x.jsonl')
|
||||
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<string, CachedFile>
|
||||
// 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'))
|
||||
|
||||
// Dirty but unchanged: the published shard is kept.
|
||||
markCacheDirty(cache, 'claude', '/p/x.jsonl')
|
||||
expect(await saveCache(cache)).toBe(true)
|
||||
expect((await mayShard()).name).toBe(after.name)
|
||||
|
||||
clearLoadCacheMemo()
|
||||
const full = await loadCache()
|
||||
expect(full.providers['claude']!.files).toEqual(parsed)
|
||||
})
|
||||
|
||||
it('converts a single-line shard from an older build on the next save and still loads stubs from it', 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',
|
||||
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([...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)
|
||||
|
||||
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.
|
||||
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'))
|
||||
})
|
||||
|
||||
it('reads the same members as a full load for every range', async () => {
|
||||
const files = { '/p/a.jsonl': fileAt('01', 'a'), '/p/b.jsonl': fileAt('15', 'b', 2), '/p/c.jsonl': fileAt('20', 'c') }
|
||||
await publish(files)
|
||||
for (const [from, to] of [['01', '01'], ['15', '15'], ['02', '14'], ['01', '31']]) {
|
||||
clearLoadCacheMemo()
|
||||
const cache = await loadCache(monthScopeForRange(new Date(`2026-05-${from}T00:00:00.000Z`), new Date(`2026-05-${to}T23:59:59.999Z`)))
|
||||
const held = { ...cache.providers['claude']!.files }
|
||||
for (const [path, stub] of cacheStubs(cache, 'claude') ?? []) {
|
||||
expect(stub.keys).toEqual(files[path as keyof typeof files].turns.map(t => t.calls.map(c => c.deduplicationKey)))
|
||||
expect(stub.fingerprint).toEqual(files[path as keyof typeof files].fingerprint)
|
||||
held[path] = files[path as keyof typeof files]
|
||||
}
|
||||
expect(held).toEqual(files)
|
||||
}
|
||||
})
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue