diff --git a/src/main/codex/codex-session-backfill-audit-pass.ts b/src/main/codex/codex-session-backfill-audit-pass.ts index 8958cf689b6..7751fa01614 100644 --- a/src/main/codex/codex-session-backfill-audit-pass.ts +++ b/src/main/codex/codex-session-backfill-audit-pass.ts @@ -2,6 +2,8 @@ import type { Stats } from 'node:fs' import { lstat } from 'node:fs/promises' import { appendCodexSessionHealAuditRecord, + addOwnedCopyEventId, + createCodexSessionBackfillCopyOwnershipKey, createCodexSessionBackfillAuditWriter, createCodexSessionBackfillDiagnosticEventId, createCodexSessionBackfillFileEventId, @@ -13,6 +15,7 @@ import type { CodexSessionBackfillSummary } from './codex-session-backfill-types export type CodexSessionBackfillAuditPass = { appendRecord: CodexSessionBackfillAuditWriter + isOwnedCopy(source: string, target: string, targetStat: Stats): boolean recordExisting( summary: CodexSessionBackfillSummary, source: string, @@ -54,6 +57,12 @@ export async function createCodexSessionBackfillAuditPass( return { appendRecord, + isOwnedCopy(source, target, targetStat): boolean { + const eventIds = coverage.ownedCopyEventIds.get( + createCodexSessionBackfillCopyOwnershipKey(source, target) + ) + return eventIds?.has(createCodexSessionBackfillFileEventId(target, targetStat)) === true + }, async recordExisting(summary, source, target, targetStat): Promise { const fileEventId = targetStat ? createCodexSessionBackfillFileEventId(target, targetStat) @@ -85,6 +94,9 @@ export async function createCodexSessionBackfillAuditPass( ) { if (fileEventId) { coverage.fileEventIds.add(fileEventId) + if (action === 'copy') { + addOwnedCopyEventId(coverage, source, target, fileEventId) + } } } }, diff --git a/src/main/codex/codex-session-backfill-audit.ts b/src/main/codex/codex-session-backfill-audit.ts index 3bc096a4d27..79c3f91b030 100644 --- a/src/main/codex/codex-session-backfill-audit.ts +++ b/src/main/codex/codex-session-backfill-audit.ts @@ -11,6 +11,7 @@ export type CodexSessionBackfillAuditWriter = (record: Record) export type CodexSessionBackfillAuditCoverage = { fileEventIds: Set diagnosticEventIds: Set + ownedCopyEventIds: Map> hasRunSummary: boolean } @@ -66,6 +67,7 @@ export async function readCodexSessionBackfillAuditCoverage( const coverage: CodexSessionBackfillAuditCoverage = { fileEventIds: new Set(), diagnosticEventIds: new Set(), + ownedCopyEventIds: new Map>(), hasRunSummary: false } const input = createReadStream(auditLogPath, { encoding: 'utf-8' }) @@ -85,6 +87,13 @@ export async function readCodexSessionBackfillAuditCoverage( typeof record.fileEventId === 'string' ) { coverage.fileEventIds.add(record.fileEventId) + if ( + record.action === 'copy' && + typeof record.source === 'string' && + typeof record.target === 'string' + ) { + addOwnedCopyEventId(coverage, record.source, record.target, record.fileEventId) + } } if ( typeof record.action === 'string' && @@ -106,17 +115,37 @@ export async function readCodexSessionBackfillAuditCoverage( } export function createCodexSessionBackfillFileEventId(targetPath: string, stat: Stats): string { - const stableFileIdentity = - stat.ino !== 0 || stat.birthtimeMs !== 0 - ? `${stat.dev}\0${stat.ino}\0${stat.birthtimeMs}` - : `${stat.size}\0${stat.mtimeMs}\0${stat.ctimeMs}` + const fileIdentity = [ + stat.dev, + stat.ino, + stat.birthtimeMs, + stat.size, + stat.mtimeMs, + stat.ctimeMs + ].join('\0') return createHash('sha256') .update(normalizeRuntimePathForComparison(targetPath)) .update('\0') - .update(stableFileIdentity) + .update(fileIdentity) .digest('hex') } +export function createCodexSessionBackfillCopyOwnershipKey(source: string, target: string): string { + return `${normalizeRuntimePathForComparison(source)}\0${normalizeRuntimePathForComparison(target)}` +} + +export function addOwnedCopyEventId( + coverage: CodexSessionBackfillAuditCoverage, + source: string, + target: string, + fileEventId: string +): void { + const key = createCodexSessionBackfillCopyOwnershipKey(source, target) + const eventIds = coverage.ownedCopyEventIds.get(key) ?? new Set() + eventIds.add(fileEventId) + coverage.ownedCopyEventIds.set(key, eventIds) +} + export function createCodexSessionBackfillDiagnosticEventId(args: { action: 'copy-unsupported' | 'failed' source: string diff --git a/src/main/codex/codex-session-backfill-copy.ts b/src/main/codex/codex-session-backfill-copy.ts index 8c033abf97d..ee7e5cc124b 100644 --- a/src/main/codex/codex-session-backfill-copy.ts +++ b/src/main/codex/codex-session-backfill-copy.ts @@ -1,5 +1,6 @@ import { randomUUID } from 'node:crypto' -import { copyFile, link, rm, writeFile } from 'node:fs/promises' +import type { Stats } from 'node:fs' +import { copyFile, link, lstat, rename, rm, writeFile } from 'node:fs/promises' import { dirname, join } from 'node:path' const ATOMIC_NO_REPLACE_UNSUPPORTED_CODE = 'ORCA_ATOMIC_NO_REPLACE_UNSUPPORTED' @@ -13,7 +14,7 @@ export async function copySessionFileWithoutOverwrite( // copy cannot strand a truncated session that a later retry would skip. await writeFile(temporaryPath, '', { encoding: 'utf-8', flag: 'wx', mode: 0o600 }) try { - await copyFile(sourcePath, temporaryPath) + await copyStableSessionFile(sourcePath, temporaryPath) try { // Why: this same-volume hardlink atomically installs the staged copy // without risking a collision overwrite after an EXDEV fallback. @@ -40,6 +41,49 @@ export async function copySessionFileWithoutOverwrite( } } +export async function replaceOwnedSessionCopy( + sourcePath: string, + targetPath: string, + expectedTargetStat: Stats +): Promise { + const temporaryPath = join(dirname(targetPath), `.orca-backfill-${randomUUID()}.tmp`) + await writeFile(temporaryPath, '', { encoding: 'utf-8', flag: 'wx', mode: 0o600 }) + try { + await copyStableSessionFile(sourcePath, temporaryPath) + const currentTargetStat = await lstat(targetPath) + if (!fileStatsMatch(currentTargetStat, expectedTargetStat)) { + const error = new Error(`Owned backfill target changed before refresh: ${targetPath}`) + ;(error as NodeJS.ErrnoException).code = 'EEXIST' + throw error + } + await rename(temporaryPath, targetPath) + } finally { + await rm(temporaryPath, { force: true }).catch(() => undefined) + } +} + +async function copyStableSessionFile(sourcePath: string, targetPath: string): Promise { + const sourceStatBefore = await lstat(sourcePath) + await copyFile(sourcePath, targetPath) + const sourceStatAfter = await lstat(sourcePath) + if (!fileStatsMatch(sourceStatAfter, sourceStatBefore)) { + const error = new Error(`Session changed while it was copied: ${sourcePath}`) + ;(error as NodeJS.ErrnoException).code = 'EBUSY' + throw error + } +} + +function fileStatsMatch(left: Stats, right: Stats): boolean { + return ( + left.dev === right.dev && + left.ino === right.ino && + left.birthtimeMs === right.birthtimeMs && + left.size === right.size && + left.mtimeMs === right.mtimeMs && + left.ctimeMs === right.ctimeMs + ) +} + function isExistsError(error: unknown): boolean { return (error as NodeJS.ErrnoException | null)?.code === 'EEXIST' } diff --git a/src/main/codex/codex-session-backfill-date.ts b/src/main/codex/codex-session-backfill-date.ts index 2b66118a8f2..25deaba91f6 100644 --- a/src/main/codex/codex-session-backfill-date.ts +++ b/src/main/codex/codex-session-backfill-date.ts @@ -1,4 +1,4 @@ -import { join } from 'node:path' +import { join, relative, sep } from 'node:path' import { listCodexSessionJsonlFilesIncrementally } from './codex-session-file-listing' import type { CodexSessionBackfillDate, @@ -13,6 +13,20 @@ export function getCodexSessionBackfillDate(date = new Date()): CodexSessionBack ] } +export function isCodexSessionRolloutPath(sessionsRoot: string, filePath: string): boolean { + const pathParts = relative(sessionsRoot, filePath).split(sep) + if (pathParts.length !== 4) { + return false + } + const [year, month, day, fileName] = pathParts + return ( + /^\d{4}$/.test(year) && + /^\d{2}$/.test(month) && + /^\d{2}$/.test(day) && + /^rollout-.+\.jsonl$/.test(fileName) + ) +} + export async function* listCodexSessionBackfillFilesForDates( sessionsRoot: string, options: CodexSessionBackfillOptions, diff --git a/src/main/codex/codex-session-backfill-owned-copy.ts b/src/main/codex/codex-session-backfill-owned-copy.ts new file mode 100644 index 00000000000..4b1f2ff33cd --- /dev/null +++ b/src/main/codex/codex-session-backfill-owned-copy.ts @@ -0,0 +1,46 @@ +import type { Stats } from 'node:fs' +import { describeCodexSessionBackfillErrorCode } from './codex-session-backfill-audit' +import { + readCodexSessionTargetStat, + type CodexSessionBackfillAuditPass +} from './codex-session-backfill-audit-pass' +import { replaceOwnedSessionCopy } from './codex-session-backfill-copy' +import type { CodexSessionBackfillSummary } from './codex-session-backfill-types' + +export async function refreshOwnedCodexSessionCopy(args: { + source: string + target: string + targetStat: Stats + summary: CodexSessionBackfillSummary + auditPass: CodexSessionBackfillAuditPass +}): Promise { + if (!args.auditPass.isOwnedCopy(args.source, args.target, args.targetStat)) { + return false + } + const sourceStat = await readCodexSessionTargetStat(args.source) + if (!sourceStat || sourceStat.size <= args.targetStat.size) { + return false + } + try { + await replaceOwnedSessionCopy(args.source, args.target, args.targetStat) + args.summary.copiedFiles += 1 + await args.auditPass.recordPublished(args.summary, 'copy', args.source, args.target) + } catch (error) { + args.summary.failedFiles += 1 + await args.auditPass.recordDiagnostic( + { + action: 'failed', + source: args.source, + target: args.target, + error: describeError(error), + errorCode: describeCodexSessionBackfillErrorCode(error) + }, + sourceStat + ) + } + return true +} + +function describeError(error: unknown): string { + return error instanceof Error ? error.message : String(error) +} diff --git a/src/main/codex/codex-session-backfill.test.ts b/src/main/codex/codex-session-backfill.test.ts index 8885881b906..bb30cf6e696 100644 --- a/src/main/codex/codex-session-backfill.test.ts +++ b/src/main/codex/codex-session-backfill.test.ts @@ -1,5 +1,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { + appendFileSync, existsSync, lstatSync, mkdirSync, @@ -25,6 +26,7 @@ const { fsMockState } = vi.hoisted(() => ({ failLink: false, failInstallLink: false, failInstallLinkTransiently: false, + growSourceDuringCopy: false, raceTargetIntoExistence: false, failCopy: false, failAuditMkdirOnce: false, @@ -116,7 +118,11 @@ vi.mock('node:fs/promises', async () => { error.code = 'EACCES' throw error } - return actual.copyFile(...args) + await actual.copyFile(...args) + if (fsMockState.growSourceDuringCopy) { + fsMockState.growSourceDuringCopy = false + await actual.appendFile(args[0], '{"event":"during-copy"}\n', 'utf-8') + } }, opendir: (...args: Parameters) => { if (args[0] === fsMockState.failDirectoryPath) { @@ -199,6 +205,7 @@ beforeEach(() => { fsMockState.failLink = false fsMockState.failInstallLink = false fsMockState.failInstallLinkTransiently = false + fsMockState.growSourceDuringCopy = false fsMockState.raceTargetIntoExistence = false fsMockState.failCopy = false fsMockState.failAuditMkdirOnce = false @@ -399,6 +406,52 @@ describe('backfillManagedCodexSessionsIntoSystemHome', () => { expect(readAuditActions()).toEqual(['copy', 'run-summary']) }) + it('refreshes an unchanged owned copy after its source grows', async () => { + fsMockState.failLink = true + const relativePath = join('2026', '05', '26', 'rollout-growing.jsonl') + const managedPath = writeManagedSession(relativePath, '{"id":"a"}\n') + const paths = resolveCodexSessionBackfillPaths() + + await backfillManagedCodexSessionsIntoSystemHome(paths) + appendFileSync(managedPath, '{"event":"later"}\n', 'utf-8') + const refreshed = await backfillManagedCodexSessionsIntoSystemHome(paths) + + expect(refreshed).toMatchObject({ copiedFiles: 1, failedFiles: 0 }) + expect(readFileSync(join(getSystemSessionsRoot(), relativePath), 'utf-8')).toBe( + '{"id":"a"}\n{"event":"later"}\n' + ) + }) + + it('does not publish a cross-volume snapshot while its source grows', async () => { + fsMockState.failLink = true + fsMockState.growSourceDuringCopy = true + const relativePath = join('2026', '05', '26', 'rollout-active.jsonl') + writeManagedSession(relativePath, '{"id":"a"}\n') + + const summary = await backfillManagedCodexSessionsIntoSystemHome( + resolveCodexSessionBackfillPaths() + ) + + expect(summary).toMatchObject({ copiedFiles: 0, failedFiles: 1 }) + expect(existsSync(join(getSystemSessionsRoot(), relativePath))).toBe(false) + }) + + it('does not refresh an owned copy after the target changes', async () => { + fsMockState.failLink = true + const relativePath = join('2026', '05', '26', 'rollout-diverged.jsonl') + const managedPath = writeManagedSession(relativePath, '{"id":"a"}\n') + const targetPath = join(getSystemSessionsRoot(), relativePath) + const paths = resolveCodexSessionBackfillPaths() + + await backfillManagedCodexSessionsIntoSystemHome(paths) + writeFileSync(targetPath, 'user-owned continuation\n', 'utf-8') + appendFileSync(managedPath, '{"event":"later"}\n', 'utf-8') + const repeated = await backfillManagedCodexSessionsIntoSystemHome(paths) + + expect(repeated).toMatchObject({ copiedFiles: 0, skippedExistingFiles: 1 }) + expect(readFileSync(targetPath, 'utf-8')).toBe('user-owned continuation\n') + }) + it('fails closed when the target filesystem cannot install without overwrite', async () => { fsMockState.failLink = true fsMockState.failInstallLink = true @@ -568,6 +621,46 @@ describe('startCodexSessionBackfillInBackground', () => { expect(scheduled).toMatchObject({ scannedFiles: 1, linkedFiles: 1 }) }) + it('does not let a bounded pass certify older unscanned history', async () => { + writeManagedSession(join('2026', '05', '26', 'rollout-baseline.jsonl'), 'baseline\n') + await startCodexSessionBackfillInBackground() + + invalidateCodexSessionBackfillMarker(getMarkerPath()) + const missedRelativePath = join('2026', '06', '01', 'rollout-missed.jsonl') + writeManagedSession(missedRelativePath, 'missed\n') + writeManagedSession(join('2026', '08', '05', 'rollout-launch.jsonl'), 'launch\n') + + const bounded = await startCodexSessionBackfillInBackground({ + ignoreCompletionMarker: true, + scanDates: [['2026', '08', '05']] + }) + expect(bounded).toMatchObject({ scannedFiles: 1, linkedFiles: 1 }) + expect(existsSync(getMarkerPath())).toBe(false) + + const recovered = await startCodexSessionBackfillInBackground() + expect(recovered).toMatchObject({ scannedFiles: 3, linkedFiles: 1 }) + expect(existsSync(join(getSystemSessionsRoot(), missedRelativePath))).toBe(true) + expect(existsSync(getMarkerPath())).toBe(true) + }) + + it('records a new heal event when a linked rollout grows in place', async () => { + const relativePath = join('2026', '05', '26', 'rollout-growing.jsonl') + const managedPath = writeManagedSession(relativePath, '{"id":"a"}\n') + await startCodexSessionBackfillInBackground() + + const firstRecord = readBackfillAuditRecords().find((record) => record.action === 'hardlink') + invalidateCodexSessionBackfillMarker(getMarkerPath()) + appendFileSync(managedPath, '{"event":"later"}\n', 'utf-8') + await startCodexSessionBackfillInBackground() + + const fileRecords = readBackfillAuditRecords().filter((record) => + ['hardlink', 'existing'].includes(record.action) + ) + expect(fileRecords).toHaveLength(2) + expect(fileRecords[1]).toMatchObject({ action: 'existing' }) + expect(fileRecords[1]?.fileEventId).not.toBe(firstRecord?.fileEventId) + }) + it('keeps repeated launch invalidations audit-stable', async () => { writeManagedSession(join('2026', '05', '26', 'rollout-a.jsonl'), '{"id":"a"}\n') diff --git a/src/main/codex/codex-session-backfill.ts b/src/main/codex/codex-session-backfill.ts index 1c79a35d44d..a547b140796 100644 --- a/src/main/codex/codex-session-backfill.ts +++ b/src/main/codex/codex-session-backfill.ts @@ -1,5 +1,5 @@ import { link, lstat, mkdir } from 'node:fs/promises' -import { dirname, join, relative, sep } from 'node:path' +import { dirname, join, relative } from 'node:path' import { getCodexSessionBackfillStateDirPath, getOrcaManagedCodexHomePath, @@ -15,12 +15,16 @@ import { copySessionFileWithoutOverwrite, isAtomicNoReplaceUnsupportedError } from './codex-session-backfill-copy' -import { listCodexSessionBackfillFilesForDates } from './codex-session-backfill-date' +import { + isCodexSessionRolloutPath, + listCodexSessionBackfillFilesForDates +} from './codex-session-backfill-date' import { captureCodexSessionBackfillMarkerGeneration, hasCompletedCodexSessionBackfillMarker, writeCodexSessionBackfillMarker as writeBackfillMarker } from './codex-session-backfill-marker' +import { refreshOwnedCodexSessionCopy } from './codex-session-backfill-owned-copy' import type { CodexSessionBackfillOptions, CodexSessionBackfillPaths, @@ -101,6 +105,7 @@ async function runCodexSessionBackfillOncePerHost( if ( !summary.stopped && options.shouldStop?.() !== true && + options.scanDates === undefined && summary.failedFiles === 0 && summary.failedDirectories === 0 && summary.failedHealAuditRecords === 0 @@ -159,7 +164,7 @@ export async function backfillManagedCodexSessionsIntoSystemHome( break } summary.scannedFiles += 1 - if (!isCodexRolloutPath(paths.managedSessionsRoot, managedSessionFilePath)) { + if (!isCodexSessionRolloutPath(paths.managedSessionsRoot, managedSessionFilePath)) { summary.skippedUnexpectedFiles += 1 continue } @@ -206,20 +211,6 @@ async function checkManagedSessionsRoot( } } -function isCodexRolloutPath(sessionsRoot: string, filePath: string): boolean { - const pathParts = relative(sessionsRoot, filePath).split(sep) - if (pathParts.length !== 4) { - return false - } - const [year, month, day, fileName] = pathParts - return ( - /^\d{4}$/.test(year) && - /^\d{2}$/.test(month) && - /^\d{2}$/.test(day) && - /^rollout-.+\.jsonl$/.test(fileName) - ) -} - async function backfillOneManagedSessionFile( paths: CodexSessionBackfillPaths, managedSessionFilePath: string, @@ -237,6 +228,17 @@ async function backfillOneManagedSessionFile( const systemSessionFilePath = join(paths.systemSessionsRoot, relativePath) const existingTargetStat = await readCodexSessionTargetStat(systemSessionFilePath) if (existingTargetStat) { + if ( + await refreshOwnedCodexSessionCopy({ + source: managedSessionFilePath, + target: systemSessionFilePath, + targetStat: existingTargetStat, + summary, + auditPass + }) + ) { + return + } await auditPass.recordExisting( summary, managedSessionFilePath, diff --git a/src/main/codex/codex-session-migration-scheduler.test.ts b/src/main/codex/codex-session-migration-scheduler.test.ts index da9d8c3872b..9d18f08190d 100644 --- a/src/main/codex/codex-session-migration-scheduler.test.ts +++ b/src/main/codex/codex-session-migration-scheduler.test.ts @@ -308,4 +308,50 @@ describe('createCodexSessionMigrationScheduler', () => { await vi.waitFor(() => expect(startBackfill).toHaveBeenCalledTimes(2)) await vi.waitFor(() => expect(startIndexHeal).toHaveBeenCalledOnce()) }) + + it('preserves scheduled identity and target recovery after a stopped pass', async () => { + let eligible = true + let target = '/old-history' + let releaseFirstBackfill: ((result: { stopped: boolean }) => void) | undefined + const prepareScheduledRun = vi.fn().mockReturnValueOnce(false).mockReturnValueOnce(true) + const finishScheduledRun = vi.fn() + const startBackfill = vi + .fn() + .mockImplementationOnce( + () => + new Promise<{ stopped: boolean }>((resolve) => { + releaseFirstBackfill = resolve + }) + ) + .mockResolvedValueOnce({ stopped: false }) + const scheduler = createCodexSessionMigrationScheduler({ + isEligible: () => eligible, + isQuitting: () => false, + resolveSystemCodexHomePathOverride: () => target, + prepareScheduledRun, + finishScheduledRun, + startBackfill, + startIndexHeal: vi.fn().mockResolvedValue(null), + initialDelayMs: 1_000 + }) + + scheduler.scheduleRun() + await vi.advanceTimersByTimeAsync(1_000) + const firstOptions = startBackfill.mock.calls[0]?.[0] + eligible = false + expect(firstOptions?.shouldStop()).toBe(true) + target = '/new-history' + releaseFirstBackfill?.({ stopped: true }) + await vi.waitFor(() => expect(prepareScheduledRun).toHaveBeenCalledOnce()) + expect(finishScheduledRun).not.toHaveBeenCalled() + + eligible = true + scheduler.requestRun() + await vi.waitFor(() => expect(startBackfill).toHaveBeenCalledTimes(2)) + expect(startBackfill).toHaveBeenLastCalledWith( + expect.objectContaining({ ignoreCompletionMarker: true, scanDates: undefined }), + '/new-history' + ) + await vi.waitFor(() => expect(finishScheduledRun).toHaveBeenCalledOnce()) + }) }) diff --git a/src/main/codex/codex-session-migration-scheduler.ts b/src/main/codex/codex-session-migration-scheduler.ts index 6e96a2c7017..fea33cfa17e 100644 --- a/src/main/codex/codex-session-migration-scheduler.ts +++ b/src/main/codex/codex-session-migration-scheduler.ts @@ -61,6 +61,7 @@ export function createCodexSessionMigrationScheduler(args: { return } const isScheduledRun = pendingScheduledRunGeneration !== null + const activeScheduledRunGeneration = pendingScheduledRunGeneration let preparationNeedsFullScan = false if (isScheduledRun) { pendingScheduledRunGeneration = null @@ -101,10 +102,15 @@ export function createCodexSessionMigrationScheduler(args: { void task.finally(() => { if (migrationTask === task) { migrationTask = null - const shouldRerun = rerunRequested || stoppedBackfill + const scheduledRunIncomplete = stoppedBackfill || activeRunStopObserved + const shouldRerun = rerunRequested || scheduledRunIncomplete rerunRequested = false activeRunStopObserved = false if (shouldRerun && isScheduledRun) { + pendingScheduledRunGeneration = Math.max( + activeScheduledRunGeneration!, + pendingScheduledRunGeneration ?? activeScheduledRunGeneration! + ) pendingFullScan ||= fullScanRequired for (const scanDate of scanDates ?? []) { pendingScanDates.set(scanDate.join('-'), scanDate) diff --git a/src/types/psl.ts b/src/types/psl.ts new file mode 100644 index 00000000000..b34a5329583 --- /dev/null +++ b/src/types/psl.ts @@ -0,0 +1,17 @@ +declare module 'psl' { + export type ParsedDomain = { + input: string + tld: string | null + sld: string | null + domain: string | null + subdomain: string | null + listed: boolean + } + + export type ParseError = { + input: string + error: { code: string; message: string } + } + + export function parse(input: string): ParsedDomain | ParseError +}