From b2b2854c5906ea9a49ac6e158d6baaa3b51bb9a9 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Mon, 7 Sep 2026 23:11:27 +0900 Subject: [PATCH 1/3] fix(codex): serialize reset-credit journal mutations (cherry picked from commit c817db47b28b44466036975577d4666083060d53) --- .../docs/reference/configuration/server.md | 2 +- src/codex/reset-credit-auto-redeem.ts | 67 +++++-- .../codex-reset-credit-auto-redeem.test.ts | 167 +++++++++++++++++- 3 files changed, 217 insertions(+), 19 deletions(-) diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index d80d6991af..cdc8d6af52 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -27,7 +27,7 @@ runs helper features around provider requests. | `codexAutoStart?` | `boolean` | `true` | Let the Codex shim run `ocx ensure` before launching Codex. False makes ensure a no-op. | | `codexShimAutoRestore?` | `boolean` | `true` | Restore an installed shim after a completed external Codex update replaces it. Environment opt-out: `OPENCODEX_CODEX_SHIM_AUTO_RESTORE=0`. | | `codexDesktopAuthless?` | `boolean` | `false` | Opt-in authless Codex Desktop routing on a loopback bind: inject the dedicated `opencodex` provider with `requires_openai_auth = false` so Desktop opens without a ChatGPT login. Ignored on non-loopback binds. `ocx system settings --desktop-authless on`. See [Codex integration](/guides/codex-integration/#authless-codex-desktop-opt-in). | -| `resetCreditAutoRedeem?` | `{ enabled?: boolean; leadTimeMinutes?: number }` | off | Opt-in: redeem the main Codex account's soonest-expiring reset credit `leadTimeMinutes` (1–60, default 10) before it expires. Every attempt re-reads the upstream credit list first and skips when the credit is gone (for example, redeemed by hand); the `redeem_request_id` is journaled in `$OPENCODEX_HOME/reset-credit-auto-redeem.json` before the call so a crash replays the same idempotent request instead of spending a second credit. Logs carry a hashed account key only. | +| `resetCreditAutoRedeem?` | `{ enabled?: boolean; leadTimeMinutes?: number }` | off | Opt-in: redeem the main Codex account's soonest-expiring reset credit `leadTimeMinutes` (1–60, default 10) before it expires. Every attempt re-reads the upstream credit list first and skips when the credit is gone (for example, redeemed by hand); the `redeem_request_id` is journaled in `$OPENCODEX_HOME/reset-credit-auto-redeem.json` before the call so a crash replays the same idempotent request instead of spending a second credit. Servers sharing this configuration directory coordinate reservations and settlements so one process does not replace another's request record. Logs carry a hashed account key only. | | `syncResumeHistory?` | `boolean` | `true` | Reversible Codex App history compatibility. Original metadata is backed up and restored by `ocx stop` / `ocx restore`. | | `shadowCallIntercept?` | `{ enabled?: boolean; model?: string; sourceModels?: string[] }` | off | Redirect recognized Codex helper/shadow calls to a chosen model while preserving the request's configured reasoning effort. The default source prefix is `gpt-5.6-luna`; older clients through 0.144.x used `gpt-5.4-mini`, which `sourceModels` can restore. | | `webSearchSidecar?` | `OcxWebSearchSidecarConfig` | on when usable | Web-search sidecar options. | diff --git a/src/codex/reset-credit-auto-redeem.ts b/src/codex/reset-credit-auto-redeem.ts index 19b6d3ae0e..0d445ae106 100644 --- a/src/codex/reset-credit-auto-redeem.ts +++ b/src/codex/reset-credit-auto-redeem.ts @@ -1,6 +1,7 @@ import { createHash, randomUUID } from "node:crypto"; import { existsSync, readFileSync } from "node:fs"; import { join } from "node:path"; +import { ConfigMutationLockError, withConfigMutationLockSync } from "../config"; import { atomicWriteFile } from "../config/atomic-write"; import { getConfigDir } from "../config/paths"; import { registerOptionalShutdownHook } from "../lib/optional-shutdown-hooks"; @@ -91,9 +92,9 @@ function readJournal(path: string): Journal { } } -function writeJournal(path: string, journal: Journal): void { +function writeJournal(path: string, journal: Journal, now: number): void { // Keep only entries whose credit could still matter: settled ones older than a week are noise. - const cutoff = Date.now() - 7 * 24 * 60 * 60_000; + const cutoff = now - 7 * 24 * 60 * 60_000; journal.entries = journal.entries.filter(e => e.state !== "settled" || e.updatedAt > cutoff); atomicWriteFile(path, JSON.stringify(journal, null, 2)); } @@ -112,6 +113,7 @@ export interface AutoRedeemDeps { now?: () => number; setTimer?: (fn: () => void, ms: number) => unknown; clearTimer?: (handle: unknown) => void; + /** Callers sharing an overridden journal must also share the OPENCODEX_HOME mutation coordinator. */ journalFile?: string; log?: (line: string) => void; /** Upper bound on one sleep so a laptop sleep or clock jump re-checks rather than trusting a stale plan. */ @@ -155,15 +157,38 @@ export function createResetCreditAutoRedeemer(deps: AutoRedeemDeps): ResetCredit handle = setTimer(() => { handle = null; void tick(); }, Math.max(0, Math.min(ms, maxSleepMs))); }; + const retryJournal = (error: unknown): void => { + const cause = error instanceof ConfigMutationLockError ? error.cause : undefined; + const code = cause && typeof cause === "object" && "code" in cause ? String(cause.code) : ""; + const busy = code === "SQLITE_BUSY" || code === "SQLITE_LOCKED" + || (cause instanceof Error && /database (?:is|table is) locked/i.test(cause.message)); + schedule(busy ? 1_000 : idleRecheckMs); + }; + const dispatch = async (plan: AutoRedeemPlan): Promise => { - const journal = readJournal(path); - let entry = journal.entries.find(e => e.accountKey === accountKey && e.grantedAt === plan.grantedAt && e.expiresAt === plan.expiresAt); - if (entry?.state === "settled") return { kind: "skipped", reason: "credit-gone" }; - if (!entry) { - entry = { accountKey, grantedAt: plan.grantedAt, expiresAt: plan.expiresAt, redeemRequestId: randomUUID(), state: "dispatched", updatedAt: now() }; - journal.entries.push(entry); - // Journal BEFORE the network call: a crash after this line replays the same request id. - writeJournal(path, journal); + // Reserve under the shared config-mutation lock. `inFlight` only serializes ticks inside + // ONE process; two servers on the same config dir would otherwise both read a journal with + // no entry, each mint a different `redeem_request_id`, and spend two credits for one plan. + let entry: JournalEntry; + try { + entry = withConfigMutationLockSync(() => { + const journal = readJournal(path); + const existing = journal.entries.find(e => e.accountKey === accountKey && e.grantedAt === plan.grantedAt && e.expiresAt === plan.expiresAt); + if (existing) return existing; + const created: JournalEntry = { accountKey, grantedAt: plan.grantedAt, expiresAt: plan.expiresAt, redeemRequestId: randomUUID(), state: "dispatched", updatedAt: now() }; + journal.entries.push(created); + // Journal BEFORE the network call: a crash after this line replays the same request id. + writeJournal(path, journal, created.updatedAt); + return created; + }); + } catch (error) { + // Only contention gets a short retry; persistent storage failures must not spin. + retryJournal(error); + return { kind: "error", message: error instanceof Error ? error.message : "journal reservation failed" }; + } + if (entry.state === "settled") { + schedule(idleRecheckMs); + return { kind: "skipped", reason: "credit-gone" }; } log(`[opencodex] reset-credit auto-redeem: dispatching for account ${accountKey} (credit expires ${plan.expiresAt})`); let result: { code: string }; @@ -174,9 +199,25 @@ export function createResetCreditAutoRedeemer(deps: AutoRedeemDeps): ResetCredit schedule(60_000); return { kind: "ambiguous", redeemRequestId: entry.redeemRequestId }; } - entry.state = "settled"; - entry.updatedAt = now(); - writeJournal(path, journal); + // Re-read under the lock: a peer may have appended its own entries since the reservation, + // and writing a stale in-memory journal would drop them. + try { + withConfigMutationLockSync(() => { + const journal = readJournal(path); + const current = journal.entries.find(e => e.accountKey === accountKey && e.grantedAt === plan.grantedAt && e.expiresAt === plan.expiresAt); + if (!current || current.redeemRequestId !== entry.redeemRequestId) { + throw new Error("auto-redeem journal reservation changed before settlement"); + } + current.state = "settled"; + current.updatedAt = now(); + writeJournal(path, journal, current.updatedAt); + }); + } catch (error) { + // Upstream answered, but settlement could not be committed. Preserve any reservation; + // a later dispatch must reuse its request id. A vanished credit may never dispatch again. + retryJournal(error); + return { kind: "error", message: error instanceof Error ? error.message : "journal settlement failed" }; + } log(`[opencodex] reset-credit auto-redeem: upstream answered ${result.code} for account ${accountKey}`); schedule(idleRecheckMs); return { kind: "dispatched", code: result.code, redeemRequestId: entry.redeemRequestId }; diff --git a/tests/codex-integration/codex-reset-credit-auto-redeem.test.ts b/tests/codex-integration/codex-reset-credit-auto-redeem.test.ts index f19f208cc1..d7656e2193 100644 --- a/tests/codex-integration/codex-reset-credit-auto-redeem.test.ts +++ b/tests/codex-integration/codex-reset-credit-auto-redeem.test.ts @@ -1,5 +1,6 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; -import { mkdtempSync, readFileSync} from "node:fs"; +import { Database } from "bun:sqlite"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { @@ -9,6 +10,7 @@ import { type ResetCredit, } from "../../src/codex/reset-credit-auto-redeem"; import { removeTreeWithRetry } from "../helpers/remove-tree"; +import { readConfigGeneration } from "../../src/config"; const T0 = Date.parse("2026-09-02T10:00:00Z"); const MIN = 60_000; @@ -18,19 +20,20 @@ const credit = (expiresInMin: number, grantedAt = "2026-09-01T00:00:00Z"): Reset }); /** Fake clock + manual timer: fire() runs the pending timer at its due time. */ -function harness(opts: { credits: () => ResetCredit[]; enabled?: () => boolean; lead?: number; journalFile: string; consumeCode?: string; consumeThrows?: boolean }) { +function harness(opts: { credits: () => ResetCredit[]; enabled?: () => boolean; lead?: number; journalFile: string; accountId?: string; consumeCode?: string; consumeThrows?: boolean; consume?: (id: string) => Promise<{ code: string }> }) { let now = T0; let pending: { fn: () => void; at: number } | null = null; const consumed: string[] = []; const logs: string[] = []; let inspects = 0; const redeemer = createResetCreditAutoRedeemer({ - accountId: "acct-main", + accountId: opts.accountId ?? "acct-main", settings: () => ({ enabled: opts.enabled ? opts.enabled() : true, leadTimeMinutes: opts.lead ?? 10 }), inspect: async () => { inspects += 1; return { credits: opts.credits() }; }, consume: async id => { if (opts.consumeThrows) throw new Error("socket hangup"); consumed.push(id); + if (opts.consume) return opts.consume(id); return { code: opts.consumeCode ?? "reset" }; }, now: () => now, @@ -49,8 +52,17 @@ function harness(opts: { credits: () => ResetCredit[]; enabled?: () => boolean; } let dir = ""; -beforeEach(() => { dir = mkdtempSync(join(tmpdir(), "ocx-auto-redeem-")); }); -afterEach(() => { removeTreeWithRetry(dir); }); +let oldHome: string | undefined; +beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "ocx-auto-redeem-")); + oldHome = process.env.OPENCODEX_HOME; + process.env.OPENCODEX_HOME = dir; +}); +afterEach(() => { + if (oldHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = oldHome; + removeTreeWithRetry(dir); +}); describe("reset-credit auto-redeem settings + plan (#822)", () => { test("default off; malformed reads as off; lead time clamped", () => { @@ -150,6 +162,151 @@ describe("reset-credit auto-redeemer runtime (#822)", () => { expect(h.consumed).toHaveLength(0); }); + test("settling a delayed consume preserves a peer's settled journal entry", async () => { + const journalFile = join(dir, "j.json"); + let entered!: () => void; + let release!: () => void; + const started = new Promise(resolve => { entered = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const a = harness({ credits: () => [credit(30)], journalFile, accountId: "acct-a", consume: async () => { + entered(); + await gate; + return { code: "reset" }; + } }); + const b = harness({ credits: () => [credit(30)], journalFile, accountId: "acct-b" }); + a.setNow(T0 + 20 * MIN); + b.setNow(T0 + 20 * MIN); + const first = a.redeemer.tick(); + try { + await Promise.race([started, first.then(() => { throw new Error("first consume was not entered"); })]); + expect((await b.redeemer.tick()).kind).toBe("dispatched"); + } finally { + release(); + await first; + } + expect((await first).kind).toBe("dispatched"); + const entries = JSON.parse(readFileSync(journalFile, "utf8")).entries as Array<{ redeemRequestId: string; state: string }>; + expect(entries).toHaveLength(2); + expect(entries.map(entry => entry.redeemRequestId).sort()).toEqual([...a.consumed, ...b.consumed].sort()); + expect(entries.every(entry => entry.state === "settled")).toBe(true); + expect((await b.redeemer.tick()).kind).toBe("skipped"); + expect(b.consumed).toHaveLength(1); + }); + + test("a separate SQLite writer blocks reservation before any consume", async () => { + const journalFile = join(dir, "j.json"); + const h = harness({ credits: () => [credit(30)], journalFile }); + h.setNow(T0 + 20 * MIN); + expect(readConfigGeneration().kind).toBe("ready"); + const holder = new Database(join(dir, "config-mutation.sqlite"), { readwrite: true, create: false }); + holder.exec("PRAGMA busy_timeout = 0; BEGIN IMMEDIATE"); + try { + expect((await h.redeemer.tick()).kind).toBe("error"); + expect(h.consumed).toHaveLength(0); + expect(existsSync(journalFile)).toBe(false); + expect(h.pendingAt()).toBe(T0 + 20 * MIN + 1_000); + } finally { + holder.exec("ROLLBACK"); + holder.close(); + } + h.setNow(T0 + 20 * MIN + 1_000); + expect((await h.redeemer.tick()).kind).toBe("dispatched"); + expect(h.consumed).toHaveLength(1); + const entries = JSON.parse(readFileSync(journalFile, "utf8")).entries; + expect(entries).toHaveLength(1); + expect(entries[0].redeemRequestId).toBe(h.consumed[0]); + expect(entries[0].state).toBe("settled"); + }); + + test("a peer that observes a settled credit keeps checking for future credits", async () => { + const journalFile = join(dir, "j.json"); + const first = harness({ credits: () => [credit(30)], journalFile }); + const peer = harness({ credits: () => [credit(30)], journalFile }); + first.setNow(T0 + 20 * MIN); + peer.setNow(T0 + 20 * MIN); + expect((await first.redeemer.tick()).kind).toBe("dispatched"); + expect((await peer.redeemer.tick()).kind).toBe("skipped"); + expect(peer.consumed).toHaveLength(0); + expect(peer.pendingAt()).toBe(T0 + 35 * MIN); + }); + + test("settlement contention keeps the reserved request id for a later retry", async () => { + const journalFile = join(dir, "j.json"); + let holder: Database | null = null; + let attempts = 0; + const h = harness({ credits: () => [credit(30)], journalFile, consume: async () => { + if (attempts++ === 0) { + holder = new Database(join(dir, "config-mutation.sqlite"), { readwrite: true, create: false }); + holder.exec("PRAGMA busy_timeout = 0; BEGIN IMMEDIATE"); + } + return { code: "reset" }; + } }); + h.setNow(T0 + 20 * MIN); + try { + expect((await h.redeemer.tick()).kind).toBe("error"); + const entries = JSON.parse(readFileSync(journalFile, "utf8")).entries; + expect(entries).toHaveLength(1); + expect(entries[0].state).toBe("dispatched"); + expect(entries[0].redeemRequestId).toBe(h.consumed[0]); + expect(h.pendingAt()).toBe(T0 + 20 * MIN + 1_000); + } finally { + if (holder) { + (holder as Database).exec("ROLLBACK"); + (holder as Database).close(); + } + } + h.setNow(T0 + 20 * MIN + 1_000); + expect((await h.redeemer.tick()).kind).toBe("dispatched"); + expect(h.consumed).toHaveLength(2); + expect(h.consumed[0]).toBe(h.consumed[1]); + expect(JSON.parse(readFileSync(journalFile, "utf8")).entries[0].state).toBe("settled"); + }); + + test("journal retention uses the redeemer's injected clock", async () => { + const start = Date.parse("2000-01-01T00:00:00Z"); + const journalFile = join(dir, "j.json"); + const h = harness({ journalFile, credits: () => [{ + granted_at: "1999-12-31T00:00:00Z", + expires_at: new Date(start + 30 * MIN).toISOString(), + }] }); + h.setNow(start + 20 * MIN); + expect((await h.redeemer.tick()).kind).toBe("dispatched"); + const entries = JSON.parse(readFileSync(journalFile, "utf8")).entries; + expect(entries).toHaveLength(1); + expect(entries[0].updatedAt).toBe(start + 20 * MIN); + }); + + test("a persistent reservation write failure uses the idle retry interval", async () => { + const journalFile = join(dir, "journal-directory"); + mkdirSync(journalFile); + const h = harness({ credits: () => [credit(30)], journalFile }); + h.setNow(T0 + 20 * MIN); + expect((await h.redeemer.tick()).kind).toBe("error"); + expect(h.consumed).toHaveLength(0); + expect(h.pendingAt()).toBe(T0 + 35 * MIN); + }); + + for (const changedReservation of ["missing", "replaced"]) { + test(`settlement rejects a ${changedReservation} reservation without overwriting it`, async () => { + const journalFile = join(dir, "j.json"); + let replacement = ""; + const h = harness({ credits: () => [credit(30)], journalFile, consume: async () => { + const journal = JSON.parse(readFileSync(journalFile, "utf8")); + if (changedReservation === "missing") journal.entries = []; + else journal.entries[0].redeemRequestId = "replacement-request"; + replacement = JSON.stringify(journal); + writeFileSync(journalFile, replacement); + return { code: "reset" }; + } }); + h.setNow(T0 + 20 * MIN); + const outcome = await h.redeemer.tick(); + expect(outcome).toEqual({ kind: "error", message: "auto-redeem journal reservation changed before settlement" }); + expect(h.consumed).toHaveLength(1); + expect(readFileSync(journalFile, "utf8")).toBe(replacement); + expect(h.pendingAt()).toBe(T0 + 35 * MIN); + }); + } + test("stop clears the timer", async () => { const h = harness({ credits: () => [credit(30)], journalFile: join(dir, "j.json") }); await h.redeemer.tick(); From ccafde814d084526a24e10b7470c2722e230b938 Mon Sep 17 00:00:00 2001 From: t Date: Tue, 8 Sep 2026 08:37:31 +0900 Subject: [PATCH 2/3] fix(codex): retain short retry for raw journal commit contention Co-authored-by: luvs01 <27862058+luvs01@users.noreply.github.com> --- src/codex/reset-credit-auto-redeem.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/codex/reset-credit-auto-redeem.ts b/src/codex/reset-credit-auto-redeem.ts index 0d445ae106..2c6292368a 100644 --- a/src/codex/reset-credit-auto-redeem.ts +++ b/src/codex/reset-credit-auto-redeem.ts @@ -158,7 +158,7 @@ export function createResetCreditAutoRedeemer(deps: AutoRedeemDeps): ResetCredit }; const retryJournal = (error: unknown): void => { - const cause = error instanceof ConfigMutationLockError ? error.cause : undefined; + const cause = error instanceof ConfigMutationLockError ? error.cause : error; const code = cause && typeof cause === "object" && "code" in cause ? String(cause.code) : ""; const busy = code === "SQLITE_BUSY" || code === "SQLITE_LOCKED" || (cause instanceof Error && /database (?:is|table is) locked/i.test(cause.message)); From a742211d0841989e67bdd5cc31ee7a35c08c9369 Mon Sep 17 00:00:00 2001 From: t Date: Tue, 8 Sep 2026 08:41:26 +0900 Subject: [PATCH 3/3] test(codex): verify journal commit contention and cross-process identity Co-authored-by: luvs01 <27862058+luvs01@users.noreply.github.com> --- .../codex-reset-credit-auto-redeem.test.ts | 260 +++++++++++++++++- 1 file changed, 257 insertions(+), 3 deletions(-) diff --git a/tests/codex-integration/codex-reset-credit-auto-redeem.test.ts b/tests/codex-integration/codex-reset-credit-auto-redeem.test.ts index d7656e2193..eb7e213420 100644 --- a/tests/codex-integration/codex-reset-credit-auto-redeem.test.ts +++ b/tests/codex-integration/codex-reset-credit-auto-redeem.test.ts @@ -1,8 +1,9 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; import { Database } from "bun:sqlite"; -import { existsSync, mkdirSync, mkdtempSync, readFileSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, renameSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { pathToFileURL } from "node:url"; import { createResetCreditAutoRedeemer, planAutoRedeem, @@ -10,6 +11,7 @@ import { type ResetCredit, } from "../../src/codex/reset-credit-auto-redeem"; import { removeTreeWithRetry } from "../helpers/remove-tree"; +import { repoPath } from "../helpers/repo-root"; import { readConfigGeneration } from "../../src/config"; const T0 = Date.parse("2026-09-02T10:00:00Z"); @@ -46,7 +48,8 @@ function harness(opts: { credits: () => ResetCredit[]; enabled?: () => boolean; redeemer, consumed, logs, inspects: () => inspects, pendingAt: () => pending?.at ?? null, - advanceAndFire: async () => { if (!pending) throw new Error("no timer"); now = pending.at; const fn = pending.fn; pending = null; fn(); await new Promise(r => setTimeout(r, 5)); }, + // The timer synchronously installs inFlight; join that tick instead of sleeping. + advanceAndFire: async () => { if (!pending) throw new Error("no timer"); now = pending.at; const fn = pending.fn; pending = null; fn(); return await redeemer.tick(); }, setNow: (t: number) => { now = t; }, }; } @@ -84,6 +87,20 @@ describe("reset-credit auto-redeem settings + plan (#822)", () => { }); describe("reset-credit auto-redeemer runtime (#822)", () => { + test("a disabled tick creates neither a journal nor a mutation coordinator", async () => { + const journalFile = join(dir, "reset-credit-auto-redeem.json"); + expect(readdirSync(dir)).toEqual([]); + const h = harness({ credits: () => [credit(30)], enabled: () => false, journalFile }); + h.setNow(T0 + 20 * MIN); + expect(await h.redeemer.tick()).toEqual({ kind: "disabled" }); + expect(h.inspects()).toBe(0); + expect(h.consumed).toHaveLength(0); + expect(h.pendingAt()).toBeNull(); + expect(existsSync(journalFile)).toBe(false); + expect(existsSync(join(dir, "config-mutation.sqlite"))).toBe(false); + expect(readdirSync(dir)).toEqual([]); + }); + test("schedules at expiry minus lead, re-reads before dispatch, journals the request id first", async () => { const journalFile = join(dir, "j.json"); const h = harness({ credits: () => [credit(30)], journalFile }); @@ -218,16 +235,253 @@ describe("reset-credit auto-redeemer runtime (#822)", () => { expect(entries[0].state).toBe("settled"); }); + test("a SQLite reader blocks COMMIT after reservation publication and retries the same id", async () => { + const journalFile = join(dir, "j.json"); + const databaseFile = join(dir, "config-mutation.sqlite"); + expect(existsSync(databaseFile)).toBe(false); + const reader = new Database(databaseFile, { create: true }); + const h = harness({ credits: () => [credit(30)], journalFile }); + h.setNow(T0 + 20 * MIN); + let reservationId = ""; + try { + // No readConfigGeneration pre-initialization: the coordinator's first acquisition + // must write its schema, so COMMIT needs an exclusive rollback-journal lock. + reader.exec("PRAGMA journal_mode = DELETE; PRAGMA busy_timeout = 0"); + reader.exec("BEGIN; CREATE TABLE reader_fixture (value INTEGER); INSERT INTO reader_fixture VALUES (1); COMMIT"); + expect(reader.query("PRAGMA journal_mode").get()).toEqual({ journal_mode: "delete" }); + expect(reader.query("SELECT name FROM sqlite_master WHERE name = 'config_generation'").all()).toEqual([]); + reader.exec("BEGIN"); + // BEGIN alone holds no read lock. This SELECT materializes the read transaction. + expect(reader.query("SELECT value FROM reader_fixture").all()).toEqual([{ value: 1 }]); + expect(reader.inTransaction).toBe(true); + const outcome = await h.redeemer.tick(); + expect(outcome).toEqual({ kind: "error", message: expect.stringMatching(/database (?:is|table is) locked/i) }); + expect(h.consumed).toHaveLength(0); + // An acquisition failure cannot publish this row: the callback ran before COMMIT failed. + const entries = JSON.parse(readFileSync(journalFile, "utf8")).entries; + expect(entries).toHaveLength(1); + expect(entries[0].state).toBe("dispatched"); + reservationId = entries[0].redeemRequestId; + expect(reservationId).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i); + expect(h.pendingAt()).toBe(T0 + 20 * MIN + 1_000); + } finally { + try { if (reader.inTransaction) reader.exec("ROLLBACK"); } finally { reader.close(); } + } + expect(await h.advanceAndFire()).toEqual({ kind: "dispatched", code: "reset", redeemRequestId: reservationId }); + expect(h.consumed).toEqual([reservationId]); + const entries = JSON.parse(readFileSync(journalFile, "utf8")).entries; + expect(entries).toHaveLength(1); + expect(entries[0].redeemRequestId).toBe(reservationId); + expect(entries[0].state).toBe("settled"); + }); + + test("two processes reserve one durable id before either consume settles", async () => { + const journalFile = join(dir, "j.json"); + const moduleUrl = pathToFileURL(repoPath("src/codex/reset-credit-auto-redeem.ts")).href; + const deadline = performance.now() + 25_000; + const markerPath = (name: string) => join(dir, name + ".json"); + const publish = (name: string) => { + const path = markerPath(name); + const temporary = path + "." + process.pid + ".tmp"; + writeFileSync(temporary, JSON.stringify({ pid: process.pid })); + renameSync(temporary, path); + }; + const launch = (worker: string) => { + const source = ` + import { existsSync, writeFileSync, renameSync } from "node:fs"; + import { join } from "node:path"; + import { createResetCreditAutoRedeemer } from ${JSON.stringify(moduleUrl)}; + const home = ${JSON.stringify(dir)}; + const worker = ${JSON.stringify(worker)}; + const deadline = performance.now() + 20_000; + const marker = name => join(home, name + ".json"); + const publish = (name, value) => { + const path = marker(name); + const temporary = path + "." + process.pid + ".tmp"; + writeFileSync(temporary, JSON.stringify({ ...value, pid: process.pid })); + renameSync(temporary, path); + }; + const waitFor = async name => { + while (!existsSync(marker(name))) { + if (performance.now() >= deadline) throw new Error("timed out waiting for " + name); + await Bun.sleep(10); + } + }; + const consumes = []; + const retries = []; + let scheduledMs = null; + const redeemer = createResetCreditAutoRedeemer({ + accountId: "acct-process-fixture", + journalFile: ${JSON.stringify(journalFile)}, + settings: () => ({ enabled: true, leadTimeMinutes: 10 }), + inspect: async () => ({ credits: [${JSON.stringify(credit(30))}] }), + now: () => ${T0 + 20 * MIN}, + // Only the loop below owns ticks; recorded timers cannot launch overlapping work. + setTimer: (_fn, ms) => { scheduledMs = ms; return 1; }, + clearTimer: () => { scheduledMs = null; }, + log: () => {}, + consume: async redeemRequestId => { + consumes.push(redeemRequestId); + if (consumes.length !== 1) throw new Error("unexpected repeated consume"); + publish(worker + "-consume", { redeemRequestId }); + await waitFor(worker + "-release"); + return { code: "reset" }; + }, + }); + try { + publish(worker + "-ready", {}); + await waitFor("start"); + let outcome; + while (true) { + if (performance.now() >= deadline) throw new Error("reservation contention deadline exceeded"); + scheduledMs = null; + outcome = await redeemer.tick(); + if (outcome.kind === "dispatched") break; + const contention = outcome.kind === "error" && ( + outcome.message === "Config mutation already in progress" + || /database (?:is|table is) locked/i.test(outcome.message) + ); + if (!contention || scheduledMs !== 1000 || consumes.length !== 0) { + throw new Error("unexpected tick: " + JSON.stringify({ outcome, scheduledMs, consumes })); + } + retries.push({ message: outcome.message, scheduledMs }); + // Honor the recorded contention delay; never retry arbitrary errors or settlement. + await Bun.sleep(scheduledMs); + } + publish(worker + "-result", { outcome, consumes, retries }); + } catch (error) { + publish(worker + "-result", { error: String(error), consumes, retries }); + console.error(error); + process.exitCode = 1; + } finally { + redeemer.stop(); + } + `; + const child = Bun.spawn([process.execPath, "-e", source], { + cwd: repoPath(), + env: { ...process.env, OPENCODEX_HOME: dir }, + stdin: "ignore", stdout: "pipe", stderr: "pipe", + }); + const output = { stdout: "", stderr: "" }; + const drain = async (stream: ReadableStream, key: "stdout" | "stderr") => { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + try { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + output[key] += decoder.decode(value, { stream: true }); + } + output[key] += decoder.decode(); + } catch (error) { + output[key] += "\npipe read failed: " + String(error); + } finally { reader.releaseLock(); } + }; + // Start draining both pipes immediately, including while waiting at the barriers. + const drained = Promise.all([drain(child.stdout, "stdout"), drain(child.stderr, "stderr")]); + return { worker, child, output, drained }; + }; + const children: ReturnType[] = []; + const released = new Set(); + const diagnostics = () => children.map(({ worker, child, output }) => + `${worker} pid=${child.pid} exit=${child.exitCode}\nstdout: ${output.stdout}\nstderr: ${output.stderr}`).join("\n"); + const waitUntil = async (label: string, ready: () => boolean) => { + while (true) { + for (const { worker, child } of children) { + if (child.exitCode !== null && (!released.has(worker) || child.exitCode !== 0)) { + throw new Error(`premature child exit waiting for ${label}\n${diagnostics()}`); + } + } + if (ready()) return; + if (performance.now() >= deadline) throw new Error(`timed out waiting for ${label}\n${diagnostics()}`); + await Bun.sleep(10); + } + }; + const readMarker = (name: string) => JSON.parse(readFileSync(markerPath(name), "utf8")); + try { + children.push(launch("a")); + children.push(launch("b")); + await waitUntil("both ready", () => children.every(({ worker }) => existsSync(markerPath(worker + "-ready")))); + for (const { worker, child } of children) expect(readMarker(worker + "-ready").pid).toBe(child.pid); + expect(children[0]!.child.pid).not.toBe(children[1]!.child.pid); + expect(existsSync(journalFile)).toBe(false); + publish("start"); + await waitUntil("both consumes", () => children.every(({ worker }) => existsSync(markerPath(worker + "-consume")))); + const ids = children.map(({ worker, child }) => { + const marker = readMarker(worker + "-consume"); + expect(marker.pid).toBe(child.pid); + expect(marker.redeemRequestId).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i); + expect(existsSync(markerPath(worker + "-result"))).toBe(false); + return marker.redeemRequestId as string; + }); + expect(new Set(ids).size).toBe(1); + const reserved = JSON.parse(readFileSync(journalFile, "utf8")).entries; + expect(reserved).toHaveLength(1); + expect(reserved[0].redeemRequestId).toBe(ids[0]); + expect(reserved[0].state).toBe("dispatched"); + // Release one child at a time so settlement needs no timing-dependent retries. + for (const { worker, child } of children) { + released.add(worker); + publish(worker + "-release"); + await waitUntil(worker + " result", () => existsSync(markerPath(worker + "-result"))); + const result = readMarker(worker + "-result"); + expect(result.pid).toBe(child.pid); + expect(result.error).toBeUndefined(); + expect(result.outcome).toEqual({ kind: "dispatched", code: "reset", redeemRequestId: ids[0] }); + expect(result.consumes).toEqual([ids[0]]); + await waitUntil(worker + " exit", () => child.exitCode !== null); + expect(await child.exited).toBe(0); + } + const settled = JSON.parse(readFileSync(journalFile, "utf8")).entries; + expect(settled).toHaveLength(1); + expect(settled[0].redeemRequestId).toBe(ids[0]); + expect(settled[0].state).toBe("settled"); + } catch (error) { + throw new Error(`${String(error)}\n${diagnostics()}`); + } finally { + try { + for (const { worker } of children) { + if (!existsSync(markerPath(worker + "-release"))) publish(worker + "-release"); + } + } finally { + // Start every cleanup even if another child's kill races its natural exit. + const cleanup = await Promise.allSettled(children.map(async ({ child, drained }) => { + try { + if (child.exitCode === null) child.kill("SIGKILL"); + } finally { + await child.exited; + await drained; + } + })); + const failedCleanup = cleanup.filter(result => result.status === "rejected"); + if (failedCleanup.length > 0) throw new AggregateError(failedCleanup.map(result => result.reason), "journal fixture child cleanup failed"); + } + } + }, 35_000); + test("a peer that observes a settled credit keeps checking for future credits", async () => { const journalFile = join(dir, "j.json"); const first = harness({ credits: () => [credit(30)], journalFile }); - const peer = harness({ credits: () => [credit(30)], journalFile }); + let peerCredits = [credit(30)]; + const peer = harness({ credits: () => peerCredits, journalFile }); first.setNow(T0 + 20 * MIN); peer.setNow(T0 + 20 * MIN); expect((await first.redeemer.tick()).kind).toBe("dispatched"); expect((await peer.redeemer.tick()).kind).toBe("skipped"); expect(peer.consumed).toHaveLength(0); expect(peer.pendingAt()).toBe(T0 + 35 * MIN); + const futureCredit = credit(45, "2026-09-02T10:30:00Z"); + peerCredits = [futureCredit]; + const outcome = await peer.advanceAndFire(); + expect(outcome).toEqual({ kind: "dispatched", code: "reset", redeemRequestId: expect.any(String) }); + expect(peer.consumed).toHaveLength(1); + expect(peer.consumed[0]).not.toBe(first.consumed[0]); + const entries = JSON.parse(readFileSync(journalFile, "utf8")).entries; + expect(entries).toHaveLength(2); + expect(entries.map((entry: { redeemRequestId: string }) => entry.redeemRequestId).sort()).toEqual([...first.consumed, ...peer.consumed].sort()); + expect(entries.find((entry: { redeemRequestId: string }) => entry.redeemRequestId === peer.consumed[0])).toMatchObject({ + grantedAt: futureCredit.granted_at, expiresAt: futureCredit.expires_at, state: "settled", + }); }); test("settlement contention keeps the reserved request id for a later retry", async () => {