diff --git a/src/codex-cache.ts b/src/codex-cache.ts index 967af2154..9120f77b6 100644 --- a/src/codex-cache.ts +++ b/src/codex-cache.ts @@ -1,4 +1,4 @@ -import { readFile, mkdir, stat, open, rename, unlink } from 'fs/promises' +import { readFile, readdir, mkdir, stat, open, rename, unlink } from 'fs/promises' import { existsSync } from 'fs' import { randomBytes } from 'crypto' import { join, resolve } from 'path' @@ -137,6 +137,47 @@ function isCurrentCache(cache: ResultCache): boolean { return cache.version === CODEX_CACHE_VERSION && !!cache.files && typeof cache.files === 'object' } +// Legacy temporary files have no owner information. Give recent writes a full day +// before reclaiming them; new temporary files also identify their live writer. +const ORPHAN_AGE_MS = 24 * 60 * 60 * 1000 + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error) +} + +async function cleanOrphanedTemps(dir: string): Promise { + try { + const names = await readdir(dir) + for (const name of names) { + const match = /^codex-(?:results(?:\.v\d+)?|projects\.v\d+)\.json\.[a-f0-9]{16}(?:\.(\d+))?\.tmp$/.exec(name) + if (!match) continue + if (match[1]) { + try { + process.kill(Number(match[1]), 0) + continue // Never unlink a live writer's file, even if it is old. + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ESRCH') continue + } + } + const path = join(dir, name) + try { + const info = await stat(path) + if (!info.isFile()) continue + console.warn(`[codeburn] Found unfinished Codex cache write: ${path} (${info.size} bytes). The disk cache may be stale; source fingerprints will be checked.`) + if (match[1] || Date.now() - info.mtimeMs >= ORPHAN_AGE_MS) await unlink(path) + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') { + console.warn(`[codeburn] Could not clean Codex cache temporary file ${path}: ${errorMessage(error)}`) + } + } + } + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') { + console.warn(`[codeburn] Could not inspect unfinished Codex cache writes in ${dir}: ${errorMessage(error)}`) + } + } +} + // Embedded consumers can change CODEBURN_CACHE_DIR without reloading this // module. Keep each directory's in-memory state separate so a warm cache (or an // unflushed update) from A can never be read from or written into B. @@ -180,23 +221,34 @@ async function writeProjectIndex(cacheDir: string, cache: ResultCache): Promise< } try { await writeFileAtomic(join(cacheDir, codexProjectsFileName()), JSON.stringify({ version: CODEX_CACHE_VERSION, files })) - } catch {} + } catch (error) { + console.warn(`[codeburn] Failed to update Codex project index ${join(cacheDir, codexProjectsFileName())}; the disk cache may be stale: ${errorMessage(error)}`) + } } async function writeFileAtomic(finalPath: string, payload: string): Promise { - 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() - } + const tempPath = `${finalPath}.${randomBytes(8).toString('hex')}.${process.pid}.tmp` + let ownsTemp = false try { + const handle = await open(tempPath, 'wx', 0o600) + ownsTemp = true + try { + await handle.writeFile(payload, { encoding: 'utf-8' }) + await handle.sync() + } finally { + await handle.close() + } await rename(tempPath, finalPath) - } catch (err) { - try { await unlink(tempPath) } catch {} - throw err + } finally { + // Cover write, fsync, close, and rename failures. Never remove a file we + // did not create (for example, after an exclusive-open collision). + if (ownsTemp) { + try { await unlink(tempPath) } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') { + console.warn(`[codeburn] Could not remove Codex cache temporary file ${tempPath}: ${errorMessage(error)}`) + } + } + } } } @@ -216,6 +268,7 @@ function loadCache(cacheDir: string): Promise { } async function loadCacheFromDisk(cacheDir: string): Promise { + await cleanOrphanedTemps(cacheDir) const empty = { version: CODEX_CACHE_VERSION, files: {} } const versioned = await readExistingTextFile(getCachePath(cacheDir)) if (versioned.status === 'ok') { @@ -387,6 +440,7 @@ export async function flushCodexCache(): Promise { const memCache = memCaches.get(cacheDir) if (!memCache) return try { + await cleanOrphanedTemps(cacheDir) // Evict entries for files that no longer exist on disk const paths = Object.keys(memCache.files) for (const p of paths) { @@ -406,5 +460,8 @@ export async function flushCodexCache(): Promise { await writeFileAtomic(getCachePath(cacheDir), JSON.stringify(memCache)) await writeProjectIndex(cacheDir, memCache) projectIndexes.delete(cacheDir) - } catch {} + } catch (error) { + // Persistence is optional; reports can still use the parsed results. + console.warn(`[codeburn] Failed to update Codex cache ${getCachePath(cacheDir)}; the disk cache may be stale: ${errorMessage(error)}`) + } } diff --git a/tests/codex-cache-write.test.ts b/tests/codex-cache-write.test.ts new file mode 100644 index 000000000..e68bc8b5c --- /dev/null +++ b/tests/codex-cache-write.test.ts @@ -0,0 +1,170 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { mkdir, mkdtemp, readFile, readdir, rm, stat, utimes, writeFile } from 'fs/promises' +import { join, resolve } from 'path' +import type { ParsedProviderCall } from '../src/providers/types.js' + +const faults = vi.hoisted(() => ({ stage: '', bytes: 0, target: 'codex-results.' })) +vi.mock('fs/promises', async (importOriginal) => { + const fs = await importOriginal() + return { + ...fs, + open: async (...args: Parameters) => { + const handle = await fs.open(...args) + const inject = String(args[0]).includes(faults.target) + return { + writeFile: async (payload: string, options: { encoding: BufferEncoding }) => { + if (inject && faults.stage === 'write') { + await handle.writeFile(payload.slice(0, faults.bytes), options) + throw new Error('injected write EIO') + } + await handle.writeFile(payload, options) + }, + sync: async () => { + if (inject && faults.stage === 'sync') throw new Error('injected sync EIO') + await handle.sync() + }, + close: async () => { + await handle.close() + if (inject && faults.stage === 'close') throw new Error('injected close EIO') + }, + } + }, + rename: async (...args: Parameters) => { + if (String(args[0]).includes(faults.target) && faults.stage === 'rename') throw new Error('injected rename EACCES') + return fs.rename(...args) + }, + } +}) + +let dir: string +let session: string +let finalPath: string +let previous: string +let mod: typeof import('../src/codex-cache.js') +let warning: ReturnType +const calls = (costUSD: number) => [{ costUSD }] as ParsedProviderCall[] + +beforeEach(async () => { + vi.resetModules() + faults.stage = '' + faults.bytes = 0 + faults.target = 'codex-results.' + await mkdir(resolve('.cache'), { recursive: true }) + dir = await mkdtemp(resolve('.cache/codex-write-')) + vi.stubEnv('CODEBURN_CACHE_DIR', dir) + warning = vi.spyOn(console, 'warn').mockImplementation(() => {}) + session = join(dir, 'session.jsonl') + await writeFile(session, '{}\n') + mod = await import('../src/codex-cache.js') + finalPath = join(dir, mod.codexCacheFileName()) + const fp = (await mod.fingerprintFile(session))! + await mod.writeCachedCodexResults(session, 'project', calls(1), fp) + await mod.flushCodexCache() + previous = await readFile(finalPath, 'utf8') + await mod.writeCachedCodexResults(session, 'project', calls(2), fp) +}) + +afterEach(async () => { + vi.restoreAllMocks() + vi.unstubAllEnvs() + await rm(dir, { recursive: true, force: true }) +}) + +describe('Codex cache write failures', () => { + it.each([ + ['write', 0], + ['write', 64], + ['sync', 0], + ['close', 0], + ['rename', 0], + ])('reports %s failure after %i bytes and preserves the complete cache', async (stage, bytes) => { + faults.stage = stage + faults.bytes = bytes + await mod.flushCodexCache() + expect(warning.mock.calls.flat().join(' ')).toContain(`injected ${stage}`) + expect(warning.mock.calls.flat().join(' ')).toContain('stale') + expect(warning.mock.calls.flat().join(' ')).toContain(finalPath) + expect(await readFile(finalPath, 'utf8')).toBe(previous) + expect((await readdir(dir)).filter(name => name.endsWith('.tmp'))).toEqual([]) + faults.stage = '' + await mod.flushCodexCache() + expect(JSON.parse(await readFile(finalPath, 'utf8')).files[session].calls[0].costUSD).toBe(2) + }) + + it.each(['write', 'sync', 'close', 'rename'])('reports project-index %s failures and cleans up', async (stage) => { + const indexPath = join(dir, mod.codexProjectsFileName()) + const previousIndex = await readFile(indexPath, 'utf8') + faults.target = 'codex-projects.' + faults.stage = stage + await mod.flushCodexCache() + expect(warning.mock.calls.flat().join(' ')).toContain(`injected ${stage}`) + expect(warning.mock.calls.flat().join(' ')).toContain(indexPath) + expect(await readFile(indexPath, 'utf8')).toBe(previousIndex) + expect(JSON.parse(await readFile(finalPath, 'utf8')).files[session].calls[0].costUSD).toBe(2) + expect((await readdir(dir)).filter(name => name.endsWith('.tmp'))).toEqual([]) + faults.stage = '' + await mod.flushCodexCache() + expect(warning.mock.calls.flat().join(' ')).toContain('stale') + }) + + it('reports and removes old empty and partial orphans on a cache read', async () => { + const empty = join(dir, 'codex-results.json.0123456789abcdef.tmp') + const partial = join(dir, 'codex-results.json.fedcba9876543210.tmp') + const recent = join(dir, 'codex-results.json.1111111111111111.tmp') + const unrelated = join(dir, 'other.tmp') + for (const [path, content] of [[empty, ''], [partial, '{"version":'], [recent, 'in progress'], [unrelated, 'keep']]) { + await writeFile(path, content) + } + const old = new Date(Date.now() - 48 * 60 * 60 * 1000) + await utimes(empty, old, old) + await utimes(partial, old, old) + vi.resetModules() + mod = await import('../src/codex-cache.js') + expect(await mod.readCachedCodexResults(session)).toEqual({ kind: 'exact', calls: calls(1) }) + expect(warning.mock.calls.flat().join(' ')).toContain('unfinished') + expect(await readdir(dir)).not.toContain(empty.split('/').at(-1)) + expect(await readdir(dir)).not.toContain(partial.split('/').at(-1)) + expect(await readFile(recent, 'utf8')).toBe('in progress') + expect(await readFile(unrelated, 'utf8')).toBe('keep') + expect(await readFile(finalPath, 'utf8')).toBe(previous) + }) + + it('reclaims a dead writer immediately but leaves a live writer alone', async () => { + const dead = join(dir, `${mod.codexProjectsFileName()}.2222222222222222.2147483647.tmp`) + const live = join(dir, `${mod.codexCacheFileName()}.3333333333333333.${process.pid}.tmp`) + await writeFile(dead, 'partial') + await writeFile(live, 'active') + const old = new Date(Date.now() - 48 * 60 * 60 * 1000) + await utimes(live, old, old) + vi.resetModules() + mod = await import('../src/codex-cache.js') + await mod.readCachedCodexResults(session) + expect(await readdir(dir)).not.toContain(dead.split('/').at(-1)) + expect(await readFile(live, 'utf8')).toBe('active') + expect(warning.mock.calls.flat().join(' ')).toContain('unfinished') + }) + + it('does not serve the old entry when a session changes after a failed flush', async () => { + faults.stage = 'write' + faults.bytes = 64 + await mod.flushCodexCache() + await writeFile(session, '{"new":"session data"}\n') + vi.resetModules() + mod = await import('../src/codex-cache.js') + expect(await mod.readCachedCodexResults(session)).toBeNull() + expect(await mod.getCachedCodexProject(session)).toBeNull() + expect((await stat(finalPath)).size).toBe(Buffer.byteLength(previous)) + }) + + it('writes a cache larger than 60 MB completely', async () => { + await mod.writeCachedCodexResults(session, 'project', [ + { costUSD: 3, userMessage: 'x'.repeat(60_543_405) } as ParsedProviderCall, + ], (await mod.fingerprintFile(session))!) + await mod.flushCodexCache() + const raw = await readFile(finalPath, 'utf8') + expect(raw.length).toBeGreaterThan(60_543_405) + expect(JSON.parse(raw).files[session].calls[0].userMessage.length).toBe(60_543_405) + expect((await readdir(dir)).filter(name => name.endsWith('.tmp'))).toEqual([]) + expect(warning).not.toHaveBeenCalled() + }) +})