fix(codex): bound state DB recovery retries (#13109)
Co-authored-by: Jinwoo-H <Jinwoo-H@users.noreply.github.com>
This commit is contained in:
Jinwoo Hong
GitHub
parent
c3939ebf0e
commit
4b5157b147
|
|
@ -10083,6 +10083,84 @@
|
|||
"tests/ is outside every tsconfig include, so the harness is linted and executed but not typechecked."
|
||||
],
|
||||
"demotionRule": "Demote if the baseline checkout cannot be materialized in CI, if a pairing has to be skipped to keep the lane green, or if the journey stops asserting the full step list and frame sequence."
|
||||
},
|
||||
{
|
||||
"id": "codex.state-db-backfill-recovery-bounds",
|
||||
"title": "Codex state DB backfill recovery has bounded process and timer use",
|
||||
"maturity": "experimental",
|
||||
"protection": "partial",
|
||||
"owner": "codex-runtime",
|
||||
"layer": "main-process-supervisor",
|
||||
"surfaces": [
|
||||
"Codex state DB backfill recovery",
|
||||
"managed Codex home startup",
|
||||
"Codex PTY and rate-limit recovery triggers"
|
||||
],
|
||||
"platforms": ["macos", "linux", "windows"],
|
||||
"providers": ["local", "wsl", "paired-runtime-host"],
|
||||
"coveredPlatforms": ["macos"],
|
||||
"coveredProviders": ["local", "wsl"],
|
||||
"coverageNotes": "A deterministic injected claimant and clock cover exits before and after the former ten-second classification boundary, exact spawn/poll/backoff counts, transient exit and spawn-error recovery, owner-lock reacquisition, WSL routing, and cleanup. Live Windows, WSL, and paired-runtime-host process evidence remains uncollected.",
|
||||
"motivatingLinks": ["https://github.com/stablyai/orca/releases/tag/v1.4.177-rc.0"],
|
||||
"invariant": "One authoritative Codex state DB backfill supervisor spawns at most five claimants regardless of process lifetime, retains at most one child and one timer, rejects later automatic restarts after claimant exhaustion, and bounds pre-claim coordinator failures to five triggers while transient claimant and owner-lock failures can recover within those budgets.",
|
||||
"oracle": "Use an injected process and clock to exit every claimant at 10,001ms without waiting. Permanent failure must give up after five spawns and exits with exactly 15 poll timers and four backoffs, no live child or pending timer, and no safety-fuse abort; repeated triggers must reuse the exhausted result with zero new lock or runner calls. Separately fail twice at 10,001ms and complete on the third claimant, recover from a first-spawn error on attempt two, reacquire after a transient owner-lock failure, and cap permanent lock-root failures at five attempts across six triggers.",
|
||||
"commands": [
|
||||
"pnpm exec vitest run --config config/vitest.config.ts src/main/codex/codex-state-db-backfill-recovery.test.ts"
|
||||
],
|
||||
"testFiles": ["src/main/codex/codex-state-db-backfill-recovery.test.ts"],
|
||||
"assertionRefs": [
|
||||
{
|
||||
"file": "src/main/codex/codex-state-db-backfill-recovery.test.ts",
|
||||
"assertions": [
|
||||
"does not restart an exhausted supervisor when later triggers arrive",
|
||||
"releases a completed supervisor entry",
|
||||
"releases a failed owner-lock attempt for later arbitration",
|
||||
"bounds permanent coordinator failures across later triggers",
|
||||
"bounds permanent claimant exits just beyond the fast-failure window",
|
||||
"recovers after claimant exits just beyond the old fast-failure window",
|
||||
"recovers after a transient process spawn error",
|
||||
"routes a WSL managed home through its distro and Linux CODEX_HOME",
|
||||
"does not interfere with a live supervisor from another Orca instance"
|
||||
]
|
||||
}
|
||||
],
|
||||
"evidenceRuns": [
|
||||
{
|
||||
"date": "2026-08-07",
|
||||
"runner": "local",
|
||||
"platform": "macos",
|
||||
"result": "passed",
|
||||
"command": "pnpm exec vitest run --config config/vitest.config.ts src/main/codex/codex-state-db-backfill-recovery.test.ts",
|
||||
"durationSeconds": 0.4,
|
||||
"summary": "Twelve deterministic recovery and owner-lock tests passed, including exact permanent and transient 10,001ms exit accounting, spawn-error recovery, trigger-level exhaustion memoization, bounded coordinator failure, successful owner-lock reacquisition, and zero retained children or timers."
|
||||
}
|
||||
],
|
||||
"runtimeBudget": {
|
||||
"p95Seconds": 2,
|
||||
"scope": "deterministic main-process supervisor and owner-lock contract"
|
||||
},
|
||||
"flakeHistory": {
|
||||
"status": "not-started",
|
||||
"evidence": "New fake-clock gate; no arbitrary sleeps or elapsed wall-time oracle."
|
||||
},
|
||||
"redGreenEvidence": {
|
||||
"status": "complete",
|
||||
"evidence": "The byte-identical production source at release commit 9e948fbdf462ede3c0160c719474100fc5cbefb7 and origin/main spawned six claimants and reached the harness fuse with 18 polls and six backoffs. The candidate gives up after five claimants with 15 polls and four backoffs; removing the total-spawn cap restores the exact red result."
|
||||
},
|
||||
"performanceBudget": {
|
||||
"required": true,
|
||||
"evidence": "The supervisor retains at most one claimant and one awaited timer. A permanent process lifetime is bounded to five subprocesses, 15 status polls, and four backoffs; later triggers reuse one settled promise and spawn nothing. Pre-claim coordinator failures perform at most five status/lock attempts. The fix adds no polling, listener, provider call, scan, wire field, or hot-path work."
|
||||
},
|
||||
"promotionCriteria": [
|
||||
"Collect 100 consecutive focused CI passes or 14 days of soak history.",
|
||||
"Collect live Windows/WSL and paired-runtime-host recovery evidence.",
|
||||
"Keep exact spawn, exit, timer, backoff, cleanup, and transient-recovery assertions green."
|
||||
],
|
||||
"knownGaps": [
|
||||
"Live child-process behavior is represented by an injected EventEmitter claimant.",
|
||||
"Physical Windows, WSL, SSH, and paired-runtime-host journeys were not run."
|
||||
],
|
||||
"demotionRule": "Demote if any exit duration can evade the spawn budget, concurrent children or timers appear, cleanup retains resources, or a transient failure cannot recover within the budget."
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,8 +5,10 @@ import { join } from 'node:path'
|
|||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import * as ownerIdentity from '../agent-hooks/managed-hook-owner-identity'
|
||||
import {
|
||||
_internals,
|
||||
resolveCodexBackfillSupervisorLockRoot,
|
||||
runCodexStateDbBackfillRecovery,
|
||||
startCodexStateDbBackfillRecoveryInBackground,
|
||||
withCodexBackfillSupervisorLock
|
||||
} from './codex-state-db-backfill-recovery'
|
||||
import type { CodexStateDbBackfillStatus } from './codex-state-db'
|
||||
|
|
@ -35,6 +37,7 @@ async function createTemporaryRoot(): Promise<string> {
|
|||
}
|
||||
|
||||
afterEach(async () => {
|
||||
_internals.resetForTests()
|
||||
Object.defineProperty(process, 'platform', { configurable: true, value: originalPlatform })
|
||||
vi.restoreAllMocks()
|
||||
vi.unstubAllEnvs()
|
||||
|
|
@ -44,6 +47,306 @@ afterEach(async () => {
|
|||
})
|
||||
|
||||
describe('Codex state DB backfill recovery', () => {
|
||||
it('does not restart an exhausted supervisor when later triggers arrive', async () => {
|
||||
vi.spyOn(console, 'info').mockImplementation(() => {})
|
||||
const run = vi.fn(async () => ({ outcome: 'gave-up' as const, spawnCount: 5 }))
|
||||
const withLock = vi.fn(
|
||||
async (_home: string, _signal: AbortSignal | undefined, claim: () => Promise<unknown>) =>
|
||||
await claim()
|
||||
)
|
||||
const dependencies = {
|
||||
isPending: vi.fn(() => true),
|
||||
run,
|
||||
withLock: withLock as never
|
||||
}
|
||||
|
||||
const first = startCodexStateDbBackfillRecoveryInBackground('/managed-home', dependencies)
|
||||
await expect(first).resolves.toEqual({ outcome: 'gave-up', spawnCount: 5 })
|
||||
const repeated = startCodexStateDbBackfillRecoveryInBackground('/managed-home', dependencies)
|
||||
|
||||
expect(repeated).toBe(first)
|
||||
await expect(repeated).resolves.toEqual({ outcome: 'gave-up', spawnCount: 5 })
|
||||
expect(dependencies.isPending).toHaveBeenCalledTimes(1)
|
||||
expect(withLock).toHaveBeenCalledTimes(1)
|
||||
expect(run).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('releases a completed supervisor entry', async () => {
|
||||
vi.spyOn(console, 'info').mockImplementation(() => {})
|
||||
const run = vi.fn(async () => ({ outcome: 'completed' as const, spawnCount: 1 }))
|
||||
const dependencies = {
|
||||
isPending: vi.fn().mockReturnValueOnce(true).mockReturnValue(false),
|
||||
run,
|
||||
withLock: vi.fn(
|
||||
async (_home: string, _signal: AbortSignal | undefined, claim: () => Promise<unknown>) =>
|
||||
await claim()
|
||||
) as never
|
||||
}
|
||||
|
||||
await expect(
|
||||
startCodexStateDbBackfillRecoveryInBackground('/managed-home', dependencies)
|
||||
).resolves.toEqual({ outcome: 'completed', spawnCount: 1 })
|
||||
await expect(
|
||||
startCodexStateDbBackfillRecoveryInBackground('/managed-home', dependencies)
|
||||
).resolves.toBeNull()
|
||||
|
||||
expect(dependencies.isPending).toHaveBeenCalledTimes(2)
|
||||
expect(run).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('releases a failed owner-lock attempt for later arbitration', async () => {
|
||||
vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
vi.spyOn(console, 'info').mockImplementation(() => {})
|
||||
const dependencies = {
|
||||
isPending: vi.fn(() => true),
|
||||
run: vi.fn(async () => ({ outcome: 'completed' as const, spawnCount: 1 })),
|
||||
withLock: vi
|
||||
.fn()
|
||||
.mockRejectedValueOnce(new Error('owner still live'))
|
||||
.mockImplementation(
|
||||
async (_home: string, _signal: AbortSignal | undefined, claim: () => Promise<unknown>) =>
|
||||
await claim()
|
||||
)
|
||||
}
|
||||
|
||||
await expect(
|
||||
startCodexStateDbBackfillRecoveryInBackground('/managed-home', dependencies as never)
|
||||
).resolves.toBeNull()
|
||||
await expect(
|
||||
startCodexStateDbBackfillRecoveryInBackground('/managed-home', dependencies as never)
|
||||
).resolves.toEqual({ outcome: 'completed', spawnCount: 1 })
|
||||
|
||||
expect(dependencies.isPending).toHaveBeenCalledTimes(2)
|
||||
expect(dependencies.withLock).toHaveBeenCalledTimes(2)
|
||||
expect(dependencies.run).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('bounds permanent coordinator failures across later triggers', async () => {
|
||||
vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
const dependencies = {
|
||||
isPending: vi.fn(() => true),
|
||||
run: vi.fn(),
|
||||
withLock: vi.fn(async () => {
|
||||
throw new Error('permanent lock-root failure')
|
||||
})
|
||||
}
|
||||
const tasks: Promise<unknown>[] = []
|
||||
|
||||
for (let attempt = 0; attempt < 6; attempt += 1) {
|
||||
const task = startCodexStateDbBackfillRecoveryInBackground(
|
||||
'/managed-home',
|
||||
dependencies as never
|
||||
)
|
||||
tasks.push(task)
|
||||
await expect(task).resolves.toBeNull()
|
||||
}
|
||||
|
||||
expect(tasks[5]).toBe(tasks[4])
|
||||
expect(dependencies.isPending).toHaveBeenCalledTimes(5)
|
||||
expect(dependencies.withLock).toHaveBeenCalledTimes(5)
|
||||
expect(dependencies.run).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('bounds permanent claimant exits just beyond the fast-failure window', async () => {
|
||||
const controller = new AbortController()
|
||||
const children = new Map<ReturnType<typeof createFakeChild>, number>()
|
||||
const timerDurations: number[] = []
|
||||
let now = 0
|
||||
let exitCount = 0
|
||||
let pendingTimers = 0
|
||||
let maxPendingTimers = 0
|
||||
let maxLiveChildren = 0
|
||||
let safetyFuseTripped = false
|
||||
const terminate = vi.fn(async () => {})
|
||||
const spawnProcess = vi.fn(() => {
|
||||
const child = createFakeChild()
|
||||
children.set(child, now + 10_001)
|
||||
maxLiveChildren = Math.max(maxLiveChildren, children.size)
|
||||
return child
|
||||
})
|
||||
const sleep = vi.fn(async (ms: number) => {
|
||||
timerDurations.push(ms)
|
||||
pendingTimers += 1
|
||||
maxPendingTimers = Math.max(maxPendingTimers, pendingTimers)
|
||||
const wakeAt = now + ms
|
||||
for (const [child, exitAt] of children) {
|
||||
if (exitAt <= wakeAt) {
|
||||
now = exitAt
|
||||
children.delete(child)
|
||||
exitCount += 1
|
||||
child.emit('exit', 1, null)
|
||||
}
|
||||
}
|
||||
now = wakeAt
|
||||
pendingTimers -= 1
|
||||
if (ms === 2_000 && spawnProcess.mock.calls.length === 6) {
|
||||
safetyFuseTripped = true
|
||||
controller.abort()
|
||||
throw Object.assign(new Error('reproduction safety fuse'), { name: 'AbortError' })
|
||||
}
|
||||
await Promise.resolve()
|
||||
})
|
||||
|
||||
const summary = await runCodexStateDbBackfillRecovery('/managed-home', controller.signal, {
|
||||
spawnProcess: spawnProcess as never,
|
||||
readStatus: vi.fn(
|
||||
(): CodexStateDbBackfillStatus => ({
|
||||
kind: 'incomplete',
|
||||
stateDbPath: '/state.sqlite',
|
||||
status: 'running'
|
||||
})
|
||||
),
|
||||
terminate,
|
||||
sleep,
|
||||
now: () => now
|
||||
})
|
||||
|
||||
expect({
|
||||
summary,
|
||||
spawnCount: spawnProcess.mock.calls.length,
|
||||
exitCount,
|
||||
timerCount: timerDurations.length,
|
||||
pollCount: timerDurations.filter((ms) => ms === 5_000).length,
|
||||
backoffCount: timerDurations.filter((ms) => ms === 2_000).length,
|
||||
liveChildren: children.size,
|
||||
pendingTimers,
|
||||
maxLiveChildren,
|
||||
maxPendingTimers,
|
||||
terminateCount: terminate.mock.calls.length,
|
||||
safetyFuseTripped
|
||||
}).toEqual({
|
||||
summary: { outcome: 'gave-up', spawnCount: 5 },
|
||||
spawnCount: 5,
|
||||
exitCount: 5,
|
||||
timerCount: 19,
|
||||
pollCount: 15,
|
||||
backoffCount: 4,
|
||||
liveChildren: 0,
|
||||
pendingTimers: 0,
|
||||
maxLiveChildren: 1,
|
||||
maxPendingTimers: 1,
|
||||
terminateCount: 0,
|
||||
safetyFuseTripped: false
|
||||
})
|
||||
})
|
||||
|
||||
it('recovers after claimant exits just beyond the old fast-failure window', async () => {
|
||||
const children = new Map<ReturnType<typeof createFakeChild>, number>()
|
||||
const timerDurations: number[] = []
|
||||
let now = 0
|
||||
let exitCount = 0
|
||||
let pendingTimers = 0
|
||||
let maxLiveChildren = 0
|
||||
const spawnProcess = vi.fn(() => {
|
||||
const child = createFakeChild()
|
||||
children.set(child, now + 10_001)
|
||||
maxLiveChildren = Math.max(maxLiveChildren, children.size)
|
||||
return child
|
||||
})
|
||||
const terminate = vi.fn(async (child: ReturnType<typeof createFakeChild>) => {
|
||||
children.delete(child)
|
||||
})
|
||||
const sleep = vi.fn(async (ms: number) => {
|
||||
timerDurations.push(ms)
|
||||
pendingTimers += 1
|
||||
const wakeAt = now + ms
|
||||
for (const [child, exitAt] of children) {
|
||||
if (exitAt <= wakeAt) {
|
||||
now = exitAt
|
||||
children.delete(child)
|
||||
exitCount += 1
|
||||
child.emit('exit', 1, null)
|
||||
}
|
||||
}
|
||||
now = wakeAt
|
||||
pendingTimers -= 1
|
||||
await Promise.resolve()
|
||||
})
|
||||
|
||||
const summary = await runCodexStateDbBackfillRecovery(
|
||||
'/managed-home',
|
||||
new AbortController().signal,
|
||||
{
|
||||
spawnProcess: spawnProcess as never,
|
||||
readStatus: vi.fn(
|
||||
(): CodexStateDbBackfillStatus =>
|
||||
spawnProcess.mock.calls.length >= 3
|
||||
? { kind: 'complete', stateDbPath: '/state.sqlite' }
|
||||
: {
|
||||
kind: 'incomplete',
|
||||
stateDbPath: '/state.sqlite',
|
||||
status: 'running'
|
||||
}
|
||||
),
|
||||
terminate: terminate as never,
|
||||
sleep,
|
||||
now: () => now
|
||||
}
|
||||
)
|
||||
|
||||
expect({
|
||||
summary,
|
||||
spawnCount: spawnProcess.mock.calls.length,
|
||||
exitCount,
|
||||
timerCount: timerDurations.length,
|
||||
pollCount: timerDurations.filter((ms) => ms === 5_000).length,
|
||||
backoffCount: timerDurations.filter((ms) => ms === 2_000).length,
|
||||
liveChildren: children.size,
|
||||
pendingTimers,
|
||||
maxLiveChildren,
|
||||
terminateCount: terminate.mock.calls.length
|
||||
}).toEqual({
|
||||
summary: { outcome: 'completed', spawnCount: 3 },
|
||||
spawnCount: 3,
|
||||
exitCount: 2,
|
||||
timerCount: 9,
|
||||
pollCount: 7,
|
||||
backoffCount: 2,
|
||||
liveChildren: 0,
|
||||
pendingTimers: 0,
|
||||
maxLiveChildren: 1,
|
||||
terminateCount: 1
|
||||
})
|
||||
})
|
||||
|
||||
it('recovers after a transient process spawn error', async () => {
|
||||
const children = [createFakeChild(), createFakeChild()]
|
||||
let spawnCount = 0
|
||||
let now = 0
|
||||
const timerDurations: number[] = []
|
||||
const spawnProcess = vi.fn(() => {
|
||||
const child = children[spawnCount++]
|
||||
if (spawnCount === 1) {
|
||||
queueMicrotask(() => child.emit('error', new Error('temporary launch failure')))
|
||||
}
|
||||
return child
|
||||
})
|
||||
const terminate = vi.fn(async () => {})
|
||||
|
||||
await expect(
|
||||
runCodexStateDbBackfillRecovery('/managed-home', new AbortController().signal, {
|
||||
spawnProcess: spawnProcess as never,
|
||||
readStatus: vi.fn(
|
||||
(): CodexStateDbBackfillStatus =>
|
||||
spawnCount >= 2
|
||||
? { kind: 'complete', stateDbPath: '/state.sqlite' }
|
||||
: { kind: 'incomplete', stateDbPath: '/state.sqlite', status: 'running' }
|
||||
),
|
||||
terminate,
|
||||
sleep: vi.fn(async (ms: number) => {
|
||||
timerDurations.push(ms)
|
||||
now += ms
|
||||
await Promise.resolve()
|
||||
}),
|
||||
now: () => now
|
||||
})
|
||||
).resolves.toEqual({ outcome: 'completed', spawnCount: 2 })
|
||||
expect(spawnProcess).toHaveBeenCalledTimes(2)
|
||||
expect(timerDurations).toEqual([5_000, 2_000, 5_000])
|
||||
expect(terminate).toHaveBeenCalledTimes(1)
|
||||
expect(terminate).toHaveBeenCalledWith(children[1])
|
||||
})
|
||||
|
||||
it('keeps the successful app-server claimant alive until Codex marks its DB complete', async () => {
|
||||
const child = createFakeChild()
|
||||
const terminate = vi.fn(async () => {})
|
||||
|
|
|
|||
|
|
@ -21,21 +21,14 @@ import {
|
|||
|
||||
const RECOVERY_POLL_INTERVAL_MS = 5_000
|
||||
const RECOVERY_RETRY_DELAY_MS = 2_000
|
||||
const RECOVERY_FAST_EXIT_MS = 10_000
|
||||
const RECOVERY_MAX_FAST_FAILURES = 5
|
||||
const RECOVERY_MAX_COORDINATOR_FAILURES = 5
|
||||
const RECOVERY_MAX_SPAWNS = 5
|
||||
const RECOVERY_MAX_TOTAL_MS = 60 * 60_000
|
||||
const RECOVERY_OWNER_CHECK_TIMEOUT_MS = 1_000
|
||||
const RECOVERY_CODEX_ARGS = ['-s', 'read-only', '-a', 'untrusted', 'app-server'] as const
|
||||
|
||||
export type CodexStateDbBackfillRecoverySummary = {
|
||||
outcome:
|
||||
| 'completed'
|
||||
| 'already-complete'
|
||||
| 'not-needed'
|
||||
| 'unreadable'
|
||||
| 'stopped'
|
||||
| 'gave-up'
|
||||
| 'codex-unavailable'
|
||||
outcome: 'completed' | 'already-complete' | 'not-needed' | 'unreadable' | 'stopped' | 'gave-up'
|
||||
spawnCount: number
|
||||
}
|
||||
|
||||
|
|
@ -129,22 +122,15 @@ export async function runCodexStateDbBackfillRecovery(
|
|||
|
||||
const deadline = dependencies.now() + RECOVERY_MAX_TOTAL_MS
|
||||
let spawnCount = 0
|
||||
let fastFailures = 0
|
||||
while (!signal.aborted && dependencies.now() < deadline) {
|
||||
const spawnedAt = dependencies.now()
|
||||
while (!signal.aborted && dependencies.now() < deadline && spawnCount < RECOVERY_MAX_SPAWNS) {
|
||||
const child = spawnRecoveryProcess(codexHomePath, dependencies)
|
||||
spawnCount += 1
|
||||
let childDown = false
|
||||
let spawnFailed = false
|
||||
let exitedAt = spawnedAt
|
||||
child.once('error', () => {
|
||||
childDown = true
|
||||
spawnFailed = true
|
||||
exitedAt = dependencies.now()
|
||||
})
|
||||
child.once('exit', () => {
|
||||
childDown = true
|
||||
exitedAt = dependencies.now()
|
||||
})
|
||||
|
||||
try {
|
||||
|
|
@ -175,14 +161,8 @@ export async function runCodexStateDbBackfillRecovery(
|
|||
await dependencies.terminate(child)
|
||||
return finish('gave-up', spawnCount)
|
||||
}
|
||||
if (spawnFailed && spawnCount === 1) {
|
||||
return finish('codex-unavailable', spawnCount)
|
||||
}
|
||||
if (exitedAt - spawnedAt < RECOVERY_FAST_EXIT_MS) {
|
||||
fastFailures += 1
|
||||
if (fastFailures >= RECOVERY_MAX_FAST_FAILURES) {
|
||||
return finish('gave-up', spawnCount)
|
||||
}
|
||||
if (spawnCount >= RECOVERY_MAX_SPAWNS) {
|
||||
return finish('gave-up', spawnCount)
|
||||
}
|
||||
try {
|
||||
await dependencies.sleep(RECOVERY_RETRY_DELAY_MS, signal)
|
||||
|
|
@ -227,38 +207,82 @@ type ActiveRecovery = {
|
|||
}
|
||||
|
||||
const activeRecoveries = new Map<string, ActiveRecovery>()
|
||||
const coordinatorFailureCounts = new Map<string, number>()
|
||||
let stopping = false
|
||||
|
||||
type RecoveryCoordinatorDependencies = {
|
||||
isPending: typeof isCodexStateDbBackfillPending
|
||||
run: typeof runCodexStateDbBackfillRecovery
|
||||
withLock: typeof withCodexBackfillSupervisorLock
|
||||
}
|
||||
|
||||
const defaultCoordinatorDependencies: RecoveryCoordinatorDependencies = {
|
||||
isPending: isCodexStateDbBackfillPending,
|
||||
run: runCodexStateDbBackfillRecovery,
|
||||
withLock: withCodexBackfillSupervisorLock
|
||||
}
|
||||
|
||||
function settleRecoveryEntry(
|
||||
key: string,
|
||||
task: ActiveRecovery['task'],
|
||||
summary: CodexStateDbBackfillRecoverySummary | null
|
||||
): void {
|
||||
if (activeRecoveries.get(key)?.task !== task) {
|
||||
return
|
||||
}
|
||||
if (summary?.outcome === 'gave-up') {
|
||||
coordinatorFailureCounts.delete(key)
|
||||
return
|
||||
}
|
||||
if (summary === null) {
|
||||
const failureCount = (coordinatorFailureCounts.get(key) ?? 0) + 1
|
||||
if (failureCount >= RECOVERY_MAX_COORDINATOR_FAILURES) {
|
||||
coordinatorFailureCounts.delete(key)
|
||||
return
|
||||
}
|
||||
coordinatorFailureCounts.set(key, failureCount)
|
||||
} else {
|
||||
coordinatorFailureCounts.delete(key)
|
||||
}
|
||||
activeRecoveries.delete(key)
|
||||
}
|
||||
|
||||
export function startCodexStateDbBackfillRecoveryInBackground(
|
||||
codexHomePath: string
|
||||
codexHomePath: string,
|
||||
dependenciesOverride: Partial<RecoveryCoordinatorDependencies> = {}
|
||||
): Promise<CodexStateDbBackfillRecoverySummary | null> {
|
||||
const dependencies = { ...defaultCoordinatorDependencies, ...dependenciesOverride }
|
||||
const key = normalizeRuntimePathForComparison(codexHomePath)
|
||||
const existing = activeRecoveries.get(key)
|
||||
if (existing) {
|
||||
return existing.task
|
||||
}
|
||||
if (stopping || !isCodexStateDbBackfillPending(codexHomePath)) {
|
||||
if (stopping) {
|
||||
return Promise.resolve(null)
|
||||
}
|
||||
if (!dependencies.isPending(codexHomePath)) {
|
||||
coordinatorFailureCounts.delete(key)
|
||||
return Promise.resolve(null)
|
||||
}
|
||||
const controller = new AbortController()
|
||||
let markReady!: () => void
|
||||
const ready = new Promise<void>((resolve) => (markReady = resolve))
|
||||
const task = withCodexBackfillSupervisorLock(codexHomePath, controller.signal, async () => {
|
||||
console.info(`[codex-state-db-backfill] supervising Codex index at ${codexHomePath}`)
|
||||
markReady()
|
||||
return await runCodexStateDbBackfillRecovery(codexHomePath, controller.signal)
|
||||
}).catch((error: unknown) => {
|
||||
if (!controller.signal.aborted) {
|
||||
console.warn('[codex-state-db-backfill] recovery supervisor stopped:', error)
|
||||
}
|
||||
return null
|
||||
})
|
||||
const task = dependencies
|
||||
.withLock(codexHomePath, controller.signal, async () => {
|
||||
console.info(`[codex-state-db-backfill] supervising Codex index at ${codexHomePath}`)
|
||||
markReady()
|
||||
return await dependencies.run(codexHomePath, controller.signal)
|
||||
})
|
||||
.catch((error: unknown) => {
|
||||
if (!controller.signal.aborted) {
|
||||
console.warn('[codex-state-db-backfill] recovery supervisor stopped:', error)
|
||||
}
|
||||
return null
|
||||
})
|
||||
void task.finally(markReady)
|
||||
activeRecoveries.set(key, { controller, ready, task })
|
||||
void task.finally(() => {
|
||||
if (activeRecoveries.get(key)?.task === task) {
|
||||
activeRecoveries.delete(key)
|
||||
}
|
||||
void task.then((summary) => {
|
||||
settleRecoveryEntry(key, task, summary)
|
||||
})
|
||||
return task
|
||||
}
|
||||
|
|
@ -284,5 +308,6 @@ export const _internals = {
|
|||
resetForTests(): void {
|
||||
stopping = false
|
||||
activeRecoveries.clear()
|
||||
coordinatorFailureCounts.clear()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue