mirror of
https://github.com/AgentSeal/codeburn.git
synced 2026-08-21 14:34:32 +00:00
fix(cache): make the shard layout safe against a second live writer
Two live processes share one cache directory routinely (a one-shot CLI beside the resident serve child, two menubar polls), and the shard layout had two ways to lose data there. - The atomic write used a FIXED temp name, so two writers publishing the envelope — every save does — shared one `envelope.json.tmp` and interleaved into a torn payload, or one deleted the shards the other's envelope named. 39 of 40 rounds ended in a total cache loss. The temp name carries a nonce again, as it did before the shard layout. - A save reused a shard filename from its own load snapshot without checking the file was still there. Another process republishing that provider unlinks the old shard, so the stale writer published an envelope naming a deleted file — read back as a corrupt provider and dropped whole, including PR-linked orphans no re-parse can recover. A reused shard is now existence-checked, and re-verified once more immediately before the envelope is published; a vanished one is rewritten from memory. Also: - Progress saves take a 30s floor beside the file counter. Only the claude scan reports per file; every other provider calls saveProgress once at its own boundary, so the counter alone never fired there. - The unreferenced-shard sweep waits an hour (temps still 5 minutes): an unreferenced shard may belong to a concurrent save whose envelope has not landed yet. - The sweep also retires the pre-v8 single-file temps in the parent directory, which nothing writes anymore. - The shard directory is created 0o700. - The claude and provider paths mark the cache dirty where they DELETE a stale entry, not only where they replace it: an unreadable file skips the replace, and the deletion would otherwise live only in memory. - Codex only treats a grown file as an append when the recorded boundary still lands just after a newline, so a same-inode rewrite that happens to end up larger re-parses instead of resuming mid-line.
This commit is contained in:
iamtoruk
parent
e7f9fbc8ce
commit
0e11b51516
6 changed files with 307 additions and 35 deletions
|
|
@ -104,6 +104,26 @@ function getEntry(cache: ResultCache, filePath: string, fp: FileFingerprint): Fi
|
|||
return null
|
||||
}
|
||||
|
||||
// A grown file is only assumed to be an APPEND if the recorded boundary still
|
||||
// falls right after a newline. A same-inode rewrite (truncate + refill, or an
|
||||
// in-place edit) that happens to end up larger would otherwise resume into the
|
||||
// middle of an unrelated line. Reading one byte is cheaper than being wrong.
|
||||
async function endsLineAt(filePath: string, offset: number): Promise<boolean> {
|
||||
if (offset === 0) return true
|
||||
try {
|
||||
const handle = await open(filePath, 'r')
|
||||
try {
|
||||
const buf = Buffer.alloc(1)
|
||||
const { bytesRead } = await handle.read(buf, 0, 1, offset - 1)
|
||||
return bytesRead === 1 && buf[0] === 0x0a
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
export async function readCachedCodexResults(
|
||||
filePath: string,
|
||||
): Promise<CodexCacheHit | null> {
|
||||
|
|
@ -125,6 +145,7 @@ export async function readCachedCodexResults(
|
|||
&& stale.resumeCallCount !== undefined
|
||||
&& fp.sizeBytes > stale.sizeBytes
|
||||
&& stale.resumeOffset <= fp.sizeBytes
|
||||
&& await endsLineAt(filePath, stale.resumeOffset)
|
||||
) {
|
||||
return { kind: 'resume', calls: stale.calls, offset: stale.resumeOffset, state: stale.resumeState, callCount: stale.resumeCallCount }
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2012,7 +2012,10 @@ async function scanProjectDirs(
|
|||
let filesDone = 0
|
||||
emitScanProgress({ kind: 'tick', provider: 'claude', done: 0, total: progressTotal })
|
||||
for (const { filePath, info, append } of changedFiles) {
|
||||
// Marked here, not after the re-parse: an unreadable file `continue`s out
|
||||
// below, and the deletion would otherwise live only in memory.
|
||||
delete section.files[filePath]
|
||||
markCacheDirty(diskCache, 'claude')
|
||||
|
||||
try {
|
||||
if (append) {
|
||||
|
|
@ -2810,6 +2813,11 @@ export function emitScanProgress(event: ScanProgressEvent): void {
|
|||
// that an interrupted long run loses little work, high enough that repeated
|
||||
// cache writes never dominate the parse.
|
||||
const PROGRESS_SAVE_FILE_INTERVAL = 2000
|
||||
// Only the claude scan reports per file; every other provider calls saveProgress
|
||||
// once, at its own boundary. Without a time floor the counter would never reach
|
||||
// the interval during a long non-claude phase and progress saves would simply
|
||||
// stop happening there.
|
||||
const PROGRESS_SAVE_MAX_INTERVAL_MS = 30_000
|
||||
|
||||
export function createScanProgress(label: string, total: number) {
|
||||
const show = !interactiveScanUI && total > 20 && process.stderr.isTTY === true
|
||||
|
|
@ -2980,6 +2988,7 @@ async function parseProviderSources(
|
|||
// that pruned-away data is preserved for monotonic monthly totals.
|
||||
if (!provider.durableSources && !clearedPaths.has(source.path)) {
|
||||
delete section.files[source.path]
|
||||
markCacheDirty(diskCache, providerName)
|
||||
clearedPaths.add(source.path)
|
||||
}
|
||||
|
||||
|
|
@ -3872,11 +3881,14 @@ async function runParse(
|
|||
// atomic (temp + rename) and writes only the dirty provider shards, so this
|
||||
// never races the final save below.
|
||||
let filesSinceSave = 0
|
||||
let lastSaveAt = Date.now()
|
||||
const saveProgress = async (): Promise<void> => {
|
||||
if (!isCold || readOnly) return
|
||||
if (!isCacheDirty(diskCache)) return
|
||||
if (++filesSinceSave < PROGRESS_SAVE_FILE_INTERVAL) return
|
||||
filesSinceSave++
|
||||
if (filesSinceSave < PROGRESS_SAVE_FILE_INTERVAL && Date.now() - lastSaveAt < PROGRESS_SAVE_MAX_INTERVAL_MS) return
|
||||
filesSinceSave = 0
|
||||
lastSaveAt = Date.now()
|
||||
try { await saveCache(diskCache) } catch { /* best-effort partial save */ }
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -175,6 +175,12 @@ const ENVELOPE_FILE = 'envelope.json'
|
|||
// versioned file is absent and the legacy file's version matches ours.
|
||||
const LEGACY_CACHE_FILE = 'session-cache.json'
|
||||
const TEMP_FILE_MAX_AGE_MS = 5 * 60 * 1000
|
||||
// A shard the published envelope does not name is either superseded garbage or
|
||||
// a CONCURRENT writer's shard that its envelope has not published yet. The
|
||||
// second case is why this guard is an order of magnitude above the temp-file
|
||||
// one: sweeping a live save's shard out from under it would publish an envelope
|
||||
// naming a file that no longer exists. No save takes an hour.
|
||||
const UNREFERENCED_SHARD_MAX_AGE_MS = 60 * 60 * 1000
|
||||
|
||||
// Env vars that change what a provider discovers or how its sessions parse.
|
||||
// computeEnvFingerprint hashes exactly these to decide when a provider's cache
|
||||
|
|
@ -695,8 +701,11 @@ function shardFileName(provider: string): string {
|
|||
return `${provider.replace(/[^A-Za-z0-9_-]/g, '_')}.${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): Promise<void> {
|
||||
const tempPath = `${finalPath}.tmp`
|
||||
const tempPath = `${finalPath}.${randomBytes(8).toString('hex')}.tmp`
|
||||
const handle = await open(tempPath, 'w', 0o600)
|
||||
try {
|
||||
await handle.writeFile(payload, { encoding: 'utf-8' })
|
||||
|
|
@ -723,21 +732,32 @@ async function writeFileAtomic(finalPath: string, payload: string): Promise<void
|
|||
|
||||
export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Promise<boolean>): Promise<boolean> {
|
||||
const dir = sessionCacheDir()
|
||||
if (!existsSync(dir)) await mkdir(dir, { recursive: true })
|
||||
if (!existsSync(dir)) await mkdir(dir, { recursive: true, mode: 0o700 })
|
||||
|
||||
const state = stateOf(cache)
|
||||
const priorShards = state.shards
|
||||
const shards: Record<string, string> = {}
|
||||
const written: string[] = []
|
||||
|
||||
const writeShard = async (provider: string): Promise<void> => {
|
||||
const name = shardFileName(provider)
|
||||
await writeFileAtomic(join(dir, name), JSON.stringify(cache.providers[provider]))
|
||||
written.push(name)
|
||||
shards[provider] = name
|
||||
}
|
||||
|
||||
try {
|
||||
for (const [provider, section] of Object.entries(cache.providers)) {
|
||||
for (const provider of Object.keys(cache.providers)) {
|
||||
const prior = priorShards[provider]
|
||||
if (prior && !state.dirtyProviders.has(provider)) { shards[provider] = prior; continue }
|
||||
const name = shardFileName(provider)
|
||||
await writeFileAtomic(join(dir, name), JSON.stringify(section))
|
||||
written.push(name)
|
||||
shards[provider] = name
|
||||
// `priorShards` is this process's snapshot from its last load or save.
|
||||
// ANOTHER process may have republished that provider since, unlinking the
|
||||
// file we are about to name — so reuse is conditional on the file still
|
||||
// being there, and a vanished one is rewritten from memory.
|
||||
if (prior && !state.dirtyProviders.has(provider) && existsSync(join(dir, prior))) {
|
||||
shards[provider] = prior
|
||||
continue
|
||||
}
|
||||
await writeShard(provider)
|
||||
}
|
||||
|
||||
// The warm refresh transaction passes an ownership fence. It must be the
|
||||
|
|
@ -750,6 +770,15 @@ export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Pr
|
|||
return false
|
||||
}
|
||||
|
||||
// Last look before publishing: a concurrent save may have unlinked a reused
|
||||
// shard while this one was writing its own. An envelope must never name a
|
||||
// file that is already gone — that reads back as a corrupt provider and
|
||||
// drops its history, including orphans no re-parse can recover.
|
||||
for (const [provider, name] of Object.entries(shards)) {
|
||||
if (written.includes(name) || existsSync(join(dir, name))) continue
|
||||
await writeShard(provider)
|
||||
}
|
||||
|
||||
const envelope: CacheEnvelope = {
|
||||
version: CACHE_VERSION,
|
||||
complete: cache.complete === true,
|
||||
|
|
@ -928,11 +957,29 @@ export function mergeCallByDedupKey(
|
|||
|
||||
// ── Temp Cleanup ───────────────────────────────────────────────────────
|
||||
|
||||
async function unlinkIfOlderThan(path: string, maxAgeMs: number, now: number): Promise<void> {
|
||||
try {
|
||||
const s = await stat(path)
|
||||
if (now - s.mtimeMs > maxAgeMs) await unlink(path)
|
||||
} catch {}
|
||||
}
|
||||
|
||||
// Sweeps our own shard directory: interrupted temp writes, plus shards the
|
||||
// published envelope no longer references (a save supersedes a provider's shard
|
||||
// rather than overwriting it). Temps from OTHER cache versions belong to old
|
||||
// binaries mid-write and are never touched.
|
||||
// published envelope no longer references. Also retires the single-file layout's
|
||||
// leftover temps in the parent directory, which nothing writes anymore.
|
||||
export async function cleanupOrphanedTempFiles(): Promise<void> {
|
||||
const now = Date.now()
|
||||
const parent = getCodeburnCacheDir()
|
||||
|
||||
// `session-cache.v<n>.json.<nonce>.tmp` from a pre-v8 binary interrupted
|
||||
// mid-write. Age-guarded, so an old binary's in-flight write is left alone.
|
||||
try {
|
||||
for (const entry of await readdir(parent)) {
|
||||
if (!/^session-cache\.v\d+\.json\..*\.tmp$/.test(entry)) continue
|
||||
await unlinkIfOlderThan(join(parent, entry), TEMP_FILE_MAX_AGE_MS, now)
|
||||
}
|
||||
} catch {}
|
||||
|
||||
const dir = sessionCacheDir()
|
||||
if (!existsSync(dir)) return
|
||||
|
||||
|
|
@ -947,17 +994,13 @@ export async function cleanupOrphanedTempFiles(): Promise<void> {
|
|||
} catch {}
|
||||
|
||||
try {
|
||||
const entries = await readdir(dir)
|
||||
const now = Date.now()
|
||||
for (const entry of entries) {
|
||||
if (!entry.endsWith('.tmp') && (!envelopeRead || referenced.has(entry))) continue
|
||||
try {
|
||||
const fullPath = join(dir, entry)
|
||||
const s = await stat(fullPath)
|
||||
if (now - s.mtimeMs > TEMP_FILE_MAX_AGE_MS) {
|
||||
await unlink(fullPath)
|
||||
}
|
||||
} catch {}
|
||||
for (const entry of await readdir(dir)) {
|
||||
if (entry.endsWith('.tmp')) {
|
||||
await unlinkIfOlderThan(join(dir, entry), TEMP_FILE_MAX_AGE_MS, now)
|
||||
continue
|
||||
}
|
||||
if (!envelopeRead || referenced.has(entry)) continue
|
||||
await unlinkIfOlderThan(join(dir, entry), UNREFERENCED_SHARD_MAX_AGE_MS, now)
|
||||
}
|
||||
} catch {}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -119,8 +119,9 @@ describe('parseAllSessions hydration lock', () => {
|
|||
expect(totalOutput(result)).toBe(50)
|
||||
// Lock released in the finally.
|
||||
expect(existsSync(lockPath())).toBe(false)
|
||||
// The parse warmed the versioned cache.
|
||||
expect(existsSync(sessionCacheDir())).toBe(true)
|
||||
// The parse warmed the versioned cache: the envelope is what publishes it,
|
||||
// so the directory merely existing proves nothing.
|
||||
expect(existsSync(join(sessionCacheDir(), 'envelope.json'))).toBe(true)
|
||||
})
|
||||
|
||||
it('ignores a fresh lock whose pid is dead', async () => {
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@ vi.mock('../../src/fs-utils.js', async (importOriginal) => {
|
|||
}
|
||||
})
|
||||
|
||||
import { flushCodexCache, withCodexCacheDirectory } from '../../src/codex-cache.js'
|
||||
import { clearCodexMemCaches, flushCodexCache, withCodexCacheDirectory } from '../../src/codex-cache.js'
|
||||
import { createCodexProvider } from '../../src/providers/codex.js'
|
||||
import type { ParsedProviderCall } from '../../src/providers/types.js'
|
||||
|
||||
|
|
@ -111,9 +111,10 @@ async function writeRollout(lines: string[]): Promise<string> {
|
|||
return path
|
||||
}
|
||||
|
||||
async function parse(cacheDir: string): Promise<ParsedProviderCall[]> {
|
||||
async function parse(cacheDir: string, codexDir = tmpDir): Promise<ParsedProviderCall[]> {
|
||||
clearCodexMemCaches()
|
||||
return withCodexCacheDirectory(cacheDir, async () => {
|
||||
const provider = createCodexProvider(tmpDir)
|
||||
const provider = createCodexProvider(codexDir)
|
||||
const sources = await provider.discoverSessions()
|
||||
const seenKeys = new Set<string>()
|
||||
const calls: ParsedProviderCall[] = []
|
||||
|
|
@ -121,6 +122,7 @@ async function parse(cacheDir: string): Promise<ParsedProviderCall[]> {
|
|||
for await (const call of provider.createSessionParser!(source, seenKeys).parse()) calls.push(call)
|
||||
}
|
||||
await flushCodexCache()
|
||||
clearCodexMemCaches()
|
||||
return calls
|
||||
})
|
||||
}
|
||||
|
|
@ -198,3 +200,88 @@ describe('codex incremental resume', () => {
|
|||
expect(JSON.stringify(resumed)).toBe(JSON.stringify(full))
|
||||
})
|
||||
})
|
||||
|
||||
// The resume snapshot has to carry EVERY field the decode reads from an earlier
|
||||
// line. Missing one shows up only when the split lands between the line that
|
||||
// sets it and the line that reads it — so split at every line boundary and
|
||||
// require the resumed decode to equal a full one each time.
|
||||
const J = (o: unknown) => JSON.stringify(o)
|
||||
const ts = (n: number, s: number) => `2026-04-14T${String(10 + Math.floor(n / 60)).padStart(2, '0')}:${String(n % 60).padStart(2, '0')}:${String(s).padStart(2, '0')}Z`
|
||||
|
||||
function richTask(n: number, opts: { tokens?: false | 'empty'; reasoning?: number; model?: string; noComplete?: boolean } = {}): string[] {
|
||||
const lines: string[] = [J({ type: 'event_msg', timestamp: ts(n, 0), payload: { type: 'task_started' } })]
|
||||
if (opts.model) lines.push(J({ type: 'turn_context', timestamp: ts(n, 0), payload: { model: opts.model } }))
|
||||
lines.push(
|
||||
J({ type: 'response_item', timestamp: ts(n, 1), payload: { type: 'message', role: 'user', content: [{ type: 'input_text', text: `please do task ${n} with some length` }] } }),
|
||||
J({ type: 'response_item', timestamp: ts(n, 2), payload: { type: 'function_call', name: 'shell', call_id: `c${n}`, arguments: J({ command: `ls ${n}` }) } }),
|
||||
J({ type: 'response_item', timestamp: ts(n, 3), payload: { type: 'function_call_output', call_id: `c${n}` } }),
|
||||
J({ type: 'event_msg', timestamp: ts(n, 4), payload: { type: 'patch_apply_end', success: n % 3 !== 0, changes: { [`/Users/test/proj/f${n}.ts`]: { unified_diff: '@@ -1 +1,2 @@\n-old\n+new\n+extra\n' } } } }),
|
||||
J({ type: 'event_msg', timestamp: ts(n, 5), payload: { type: 'mcp_tool_call_end', call_id: `m${n}`, invocation: { server: 'github', tool: 'list' }, duration_ms: 120, result: { Ok: {} } } }),
|
||||
J({ type: 'response_item', timestamp: ts(n, 6), payload: { type: 'message', role: 'assistant', content: [{ type: 'output_text', text: 'y'.repeat(120) }] } }),
|
||||
)
|
||||
if (opts.tokens === 'empty') {
|
||||
// No `info`: the estimated-usage path, which advances estCounter.
|
||||
lines.push(J({ type: 'event_msg', timestamp: ts(n, 7), payload: { type: 'token_count' } }))
|
||||
} else if (opts.tokens !== false) {
|
||||
const c = { input: 100 * n, cached: 20 * n, output: 50 * n, reasoning: (opts.reasoning ?? 10) * n }
|
||||
lines.push(J({ type: 'event_msg', timestamp: ts(n, 7), payload: { type: 'token_count', info: {
|
||||
last_token_usage: { input_tokens: 100, cached_input_tokens: 20, output_tokens: 50, reasoning_output_tokens: opts.reasoning ?? 10, total_tokens: 180 },
|
||||
total_token_usage: { input_tokens: c.input, cached_input_tokens: c.cached, output_tokens: c.output, reasoning_output_tokens: c.reasoning, total_tokens: c.input + c.output + c.reasoning },
|
||||
} } }))
|
||||
}
|
||||
if (!opts.noComplete) lines.push(J({ type: 'event_msg', timestamp: ts(n, 8), payload: { type: 'task_complete', duration_ms: 5000 } }))
|
||||
return lines
|
||||
}
|
||||
|
||||
const RICH_LINES = [
|
||||
J({ type: 'session_meta', timestamp: ts(0, 0), payload: { cwd: '/Users/test/proj', originator: 'codex-cli', session_id: 'sess-1', model: 'gpt-5.3-codex' } }),
|
||||
...richTask(1),
|
||||
...richTask(2, { tokens: 'empty' }),
|
||||
...richTask(7, { tokens: 'empty' }),
|
||||
...richTask(3, { model: 'gpt-5.3-codex-mini' }),
|
||||
...richTask(4, { reasoning: 33 }),
|
||||
...richTask(5, { noComplete: true }),
|
||||
...richTask(6),
|
||||
]
|
||||
|
||||
const FORK_LINES = [
|
||||
J({ type: 'session_meta', timestamp: ts(0, 0), payload: { cwd: '/Users/test/proj', originator: 'codex-cli', session_id: 'sess-2', forked_from_id: 'sess-1', model: 'gpt-5.3-codex' } }),
|
||||
// Parent history replayed inside the 5s fork cutoff: must stay skipped across a split.
|
||||
...richTask(0).map(l => l.replace(/2026-04-14T10:00:0\d/g, '2026-04-14T10:00:01')),
|
||||
...richTask(11),
|
||||
...richTask(12, { tokens: 'empty' }),
|
||||
]
|
||||
|
||||
describe('codex resume differential', () => {
|
||||
async function rollout(lines: string[]): Promise<{ codexDir: string; path: string }> {
|
||||
const codexDir = await mkdtemp(join(tmpdir(), 'codex-split-'))
|
||||
const dir = join(codexDir, 'sessions', '2026', '04', '14')
|
||||
await mkdir(dir, { recursive: true })
|
||||
const path = join(dir, 'rollout-sess-1.jsonl')
|
||||
await writeFile(path, lines.join('\n') + '\n')
|
||||
return { codexDir, path }
|
||||
}
|
||||
|
||||
async function assertEverySplitMatches(lines: string[]): Promise<void> {
|
||||
const base = await rollout(lines)
|
||||
const full = await parse(await mkdtemp(join(tmpdir(), 'codex-c-')), base.codexDir)
|
||||
expect(full.length).toBeGreaterThan(0)
|
||||
|
||||
for (let k = 1; k < lines.length; k++) {
|
||||
const split = await rollout(lines.slice(0, k))
|
||||
const cacheDir = await mkdtemp(join(tmpdir(), 'codex-c-'))
|
||||
await parse(cacheDir, split.codexDir)
|
||||
await appendFile(split.path, lines.slice(k).join('\n') + '\n')
|
||||
const resumed = await parse(cacheDir, split.codexDir)
|
||||
expect(JSON.stringify(resumed), `split after line ${k}: ${lines[k - 1]!.slice(0, 80)}`).toBe(JSON.stringify(full))
|
||||
}
|
||||
}
|
||||
|
||||
it('matches a full re-parse at every line boundary of a rich session', async () => {
|
||||
await assertEverySplitMatches(RICH_LINES)
|
||||
})
|
||||
|
||||
it('matches a full re-parse at every line boundary of a forked session', async () => {
|
||||
await assertEverySplitMatches(FORK_LINES)
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -183,26 +183,134 @@ describe('cleanupOrphanedTempFiles', () => {
|
|||
const dir = sessionCacheDir()
|
||||
const live = (await shardNames()).find(n => n.startsWith('claude.'))!
|
||||
|
||||
const stale = new Date(Date.now() - 10 * 60 * 1000)
|
||||
const backdate = async (path: string, minutes: number) => {
|
||||
const at = new Date(Date.now() - minutes * 60 * 1000)
|
||||
await utimes(path, at, at)
|
||||
}
|
||||
|
||||
const oldTemp = join(dir, 'claude.deadbeef.json.tmp')
|
||||
await writeFile(oldTemp, 'partial')
|
||||
await utimes(oldTemp, stale, stale)
|
||||
await backdate(oldTemp, 10)
|
||||
const orphanShard = join(dir, 'codex.deadbeef.json')
|
||||
await writeFile(orphanShard, '{}')
|
||||
await utimes(orphanShard, stale, stale)
|
||||
await backdate(orphanShard, 90)
|
||||
// Unreferenced but fresh: this is what a CONCURRENT save's shard looks like
|
||||
// before its envelope lands, so the sweep must leave it alone.
|
||||
const inFlightShard = join(dir, 'codex.c0ffee00.json')
|
||||
await writeFile(inFlightShard, '{}')
|
||||
const recentTemp = join(dir, 'claude.feedface.json.tmp')
|
||||
await writeFile(recentTemp, 'in flight')
|
||||
// The live shard is far older than the temp cutoff; being referenced is what
|
||||
// protects it, not its age.
|
||||
await backdate(join(dir, live), 120)
|
||||
|
||||
await cleanupOrphanedTempFiles()
|
||||
|
||||
expect(existsSync(oldTemp)).toBe(false)
|
||||
expect(existsSync(orphanShard)).toBe(false)
|
||||
expect(existsSync(inFlightShard)).toBe(true)
|
||||
expect(existsSync(recentTemp)).toBe(true)
|
||||
expect(existsSync(join(dir, live))).toBe(true)
|
||||
expect(existsSync(join(dir, 'envelope.json'))).toBe(true)
|
||||
// The live shard is untouched by the sweep even though it is older than the
|
||||
// temp-file age cutoff.
|
||||
const info = await stat(join(dir, live))
|
||||
expect(info.size).toBeGreaterThan(0)
|
||||
})
|
||||
|
||||
it('retires the pre-v8 single-file layout temps left in the parent directory', async () => {
|
||||
await saveCache({ version: CACHE_VERSION, complete: true, providers: {} })
|
||||
const legacyTemp = join(TMP_DIR, 'session-cache.v7.json.abc123.tmp')
|
||||
await writeFile(legacyTemp, 'orphan from an older build')
|
||||
const at = new Date(Date.now() - 10 * 60 * 1000)
|
||||
await utimes(legacyTemp, at, at)
|
||||
const freshLegacyTemp = join(TMP_DIR, 'session-cache.v7.json.def456.tmp')
|
||||
await writeFile(freshLegacyTemp, 'an old binary mid-write')
|
||||
|
||||
await cleanupOrphanedTempFiles()
|
||||
|
||||
expect(existsSync(legacyTemp)).toBe(false)
|
||||
expect(existsSync(freshLegacyTemp)).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
// Two live processes share one cache directory routinely: a one-shot CLI beside
|
||||
// the resident serve child, or two menubar polls. Neither may publish an
|
||||
// envelope naming a file that is not there — that reads back as a corrupt
|
||||
// provider and silently drops its history.
|
||||
describe('concurrent writers', () => {
|
||||
function seed(provider: string, tag: string, files: number): SessionCache {
|
||||
const cache: SessionCache = {
|
||||
version: CACHE_VERSION,
|
||||
complete: true,
|
||||
providers: {
|
||||
[provider]: {
|
||||
envFingerprint: tag,
|
||||
files: Object.fromEntries(
|
||||
Array.from({ length: files }, (_, i) => [`/f/${provider}/${i}.jsonl`, cachedFile()]),
|
||||
),
|
||||
},
|
||||
},
|
||||
}
|
||||
markCacheDirty(cache, provider)
|
||||
return cache
|
||||
}
|
||||
|
||||
async function assertReferentialIntegrity(expected: string[]): Promise<void> {
|
||||
const dir = sessionCacheDir()
|
||||
const envelope = JSON.parse(await readFile(join(dir, 'envelope.json'), 'utf-8'))
|
||||
for (const name of Object.values(envelope.shards) as string[]) {
|
||||
expect(existsSync(join(dir, name)), `envelope names a missing shard: ${name}`).toBe(true)
|
||||
}
|
||||
clearLoadCacheMemo()
|
||||
const loaded = await loadCache()
|
||||
expect(Object.keys(loaded.providers).length).toBeGreaterThan(0)
|
||||
for (const provider of expected) expect(loaded.providers[provider]).toBeDefined()
|
||||
}
|
||||
|
||||
it('never publishes a dangling envelope when two saves race', async () => {
|
||||
for (let round = 0; round < 15; round++) {
|
||||
await Promise.allSettled([
|
||||
saveCache(seed('claude', `a${round}`, 40)),
|
||||
saveCache(seed('codex', `b${round}`, 40)),
|
||||
])
|
||||
// Whichever won, the published set has to be internally consistent and
|
||||
// hold at least the provider that got there last.
|
||||
await assertReferentialIntegrity([])
|
||||
}
|
||||
})
|
||||
|
||||
it('a stale writer rewrites a shard another process retired instead of orphaning it', async () => {
|
||||
// Seed: claude holds an expired-source PR orphan no re-parse can recover.
|
||||
const initial: SessionCache = {
|
||||
version: CACHE_VERSION,
|
||||
complete: true,
|
||||
providers: {
|
||||
claude: { envFingerprint: 'fp', files: { '/gone/pruned.jsonl': cachedFile({ prLinks: ['https://github.com/o/r/pull/1'] }) } },
|
||||
codex: { envFingerprint: 'fp', durable: true, files: { '/live/r.jsonl': cachedFile() } },
|
||||
},
|
||||
}
|
||||
markCacheDirty(initial, 'claude')
|
||||
markCacheDirty(initial, 'codex')
|
||||
await saveCache(initial)
|
||||
|
||||
// Process B loads now, recording claude's current shard name.
|
||||
clearLoadCacheMemo()
|
||||
const b = await loadCache()
|
||||
|
||||
// Process A independently touches ONLY claude and republishes, retiring the
|
||||
// shard file B is still holding a name for.
|
||||
clearLoadCacheMemo()
|
||||
const a = await loadCache()
|
||||
a.providers['claude']!.files['/live/new.jsonl'] = cachedFile()
|
||||
markCacheDirty(a, 'claude')
|
||||
await saveCache(a)
|
||||
|
||||
// B now saves an unrelated codex change.
|
||||
b.providers['codex']!.files['/live/r2.jsonl'] = cachedFile()
|
||||
markCacheDirty(b, 'codex')
|
||||
await saveCache(b)
|
||||
|
||||
await assertReferentialIntegrity(['claude', 'codex'])
|
||||
clearLoadCacheMemo()
|
||||
const final = await loadCache()
|
||||
expect(final.providers['claude']!.files['/gone/pruned.jsonl']).toBeDefined()
|
||||
expect(final.providers['codex']!.files['/live/r2.jsonl']).toBeDefined()
|
||||
})
|
||||
})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue