diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 458bb67e0a..261aece1d1 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -67,6 +67,14 @@ jobs: bump-dev-version: needs: publish if: ${{ inputs.dry-run != true }} + # A reusable-workflow CALL cannot grant the callee more than the calling job holds, + # and GitHub refuses the whole run at startup when the called workflow's own job + # declares permissions the caller did not pass down ("startup_failure", runs + # 33615174183 / 33615177849 — the first dispatches since #3129 wired this call). + # The callee's job declares exactly these two; nothing else in this file gains them. + permissions: + contents: write + pull-requests: write uses: ./.github/workflows/dev-version-bump.yml with: released-version: v${{ inputs.version }} diff --git a/src/codex/reset-credit-auto-redeem.ts b/src/codex/reset-credit-auto-redeem.ts index 19b6d3ae0e..d3d365a23b 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 { withConfigMutationLockSync } from "../config"; import { atomicWriteFile } from "../config/atomic-write"; import { getConfigDir } from "../config/paths"; import { registerOptionalShutdownHook } from "../lib/optional-shutdown-hooks"; @@ -118,6 +119,8 @@ export interface AutoRedeemDeps { maxSleepMs?: number; /** Interval to re-inspect when no credit is due yet (default 30 min). */ idleRecheckMs?: number; + /** @internal Test seam used to synchronize peer processes immediately before journal reservation. */ + beforeDispatchForTest?: () => Promise; } export type AutoRedeemOutcome = @@ -156,15 +159,25 @@ export function createResetCreditAutoRedeemer(deps: AutoRedeemDeps): ResetCredit }; 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); + let entry: JournalEntry; + try { + entry = withConfigMutationLockSync(() => { + const journal = readJournal(path); + let reserved = journal.entries.find(e => e.accountKey === accountKey && e.grantedAt === plan.grantedAt && e.expiresAt === plan.expiresAt); + if (!reserved) { + reserved = { accountKey, grantedAt: plan.grantedAt, expiresAt: plan.expiresAt, redeemRequestId: randomUUID(), state: "dispatched", updatedAt: now() }; + journal.entries.push(reserved); + // Reserve BEFORE the network call. The shared transaction makes minting the + // idempotency key atomic across sibling server processes. + writeJournal(path, journal); + } + return reserved; + }); + } catch (error) { + schedule(1_000); + return { kind: "error", message: error instanceof Error ? error.message : "journal reservation failed" }; } + if (entry.state === "settled") return { kind: "skipped", reason: "credit-gone" }; log(`[opencodex] reset-credit auto-redeem: dispatching for account ${accountKey} (credit expires ${plan.expiresAt})`); let result: { code: string }; try { @@ -174,9 +187,22 @@ export function createResetCreditAutoRedeemer(deps: AutoRedeemDeps): ResetCredit schedule(60_000); return { kind: "ambiguous", redeemRequestId: entry.redeemRequestId }; } - entry.state = "settled"; - entry.updatedAt = now(); - writeJournal(path, journal); + 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.state = "settled"; + current.updatedAt = now(); + writeJournal(path, journal); + } + }); + } catch (error) { + // The consume succeeded, so retain the same operation id and retry settlement rather + // than allowing a peer to mint a replacement reservation. + schedule(1_000); + 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 }; @@ -206,6 +232,7 @@ export function createResetCreditAutoRedeemer(deps: AutoRedeemDeps): ResetCredit return { kind: "error", message: error instanceof Error ? error.message : "inspect failed" } as AutoRedeemOutcome; } if (!creditStillPresent(fresh, plan)) { schedule(idleRecheckMs); return { kind: "skipped", reason: "credit-gone" } as AutoRedeemOutcome; } + await deps.beforeDispatchForTest?.(); return dispatch(plan); })().finally(() => { inFlight = null; }); return inFlight; diff --git a/tests/codex-reset-credit-auto-redeem.test.ts b/tests/codex-reset-credit-auto-redeem.test.ts index 21f651eef9..024e39335e 100644 --- a/tests/codex-reset-credit-auto-redeem.test.ts +++ b/tests/codex-reset-credit-auto-redeem.test.ts @@ -1,7 +1,8 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; -import { mkdtempSync, readFileSync} from "node:fs"; +import { mkdtempSync, readFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { pathToFileURL } from "node:url"; import { createResetCreditAutoRedeemer, planAutoRedeem, @@ -141,6 +142,49 @@ describe("reset-credit auto-redeemer runtime (#822)", () => { expect(resumed.consumed).toEqual([id]); }); + test("sibling processes share one idempotency key for the same credit", async () => { + const workers = 8; + const journalFile = join(dir, "j.json"); + const consumeLog = join(dir, "consume.log"); + const moduleUrl = pathToFileURL(join(import.meta.dir, "../src/codex/reset-credit-auto-redeem.ts")).href; + const script = ` + import { appendFileSync, readdirSync, writeFileSync } from "node:fs"; + import { join } from "node:path"; + import { createResetCreditAutoRedeemer } from ${JSON.stringify(moduleUrl)}; + const [dir, journalFile, consumeLog, worker, count] = process.argv.slice(1); + const now = ${T0 + 20 * MIN}; + const redeemer = createResetCreditAutoRedeemer({ + accountId: "acct-main", + settings: () => ({ enabled: true, leadTimeMinutes: 10 }), + inspect: async () => ({ credits: [{ granted_at: "2026-09-01T00:00:00Z", expires_at: "2026-09-02T10:30:00.000Z" }] }), + consume: async id => { appendFileSync(consumeLog, id + "\\n"); await Bun.sleep(25); return { code: "reset" }; }, + now: () => now, + setTimer: () => 1, + clearTimer: () => {}, + journalFile, + log: () => {}, + beforeDispatchForTest: async () => { + writeFileSync(join(dir, "ready-" + worker), ""); + while (readdirSync(dir).filter(name => name.startsWith("ready-")).length < Number(count)) await Bun.sleep(1); + }, + }); + for (;;) { + const outcome = await redeemer.tick(); + if (outcome.kind !== "error") break; + await Bun.sleep(2); + } + `; + const children = Array.from({ length: workers }, (_, worker) => Bun.spawn( + [process.execPath, "--eval", script, dir, journalFile, consumeLog, String(worker), String(workers)], + { env: { ...process.env, OPENCODEX_HOME: dir }, stdout: "pipe", stderr: "pipe" }, + )); + const exits = await Promise.all(children.map(child => child.exited)); + expect(exits).toEqual(Array(workers).fill(0)); + const ids = readFileSync(consumeLog, "utf8").trim().split("\n"); + expect(ids.length).toBeGreaterThan(0); + expect(new Set(ids).size).toBe(1); + }); + test("a manual redeem racing between the planning read and the pre-dispatch read is caught", async () => { const journalFile = join(dir, "j.json"); let reads = 0;