From d5dc52596b05e0d0d3d1a67aee2310fcdf5e7500 Mon Sep 17 00:00:00 2001 From: Bruno Date: Tue, 10 Nov 2026 16:32:00 -0300 Subject: [PATCH 1/3] fix(anthropic-pool): lift 429 quota cooldowns from a background usage read A fresh usage read that shows every rejected window open lifts an Anthropic 429 quota cooldown (#6989), but nothing issued that read on its own: it ran only when the dashboard, `ocx account refresh` or a reset admitted through the proxy asked for quota. A user whose window reopened elsewhere (a reset grant spent in the Claude app, a plan upgrade) kept getting "All Anthropic OAuth accounts are temporarily rate-limited" from a pool whose accounts upstream had already reopened, until the old deadline or a manual clear-cooldown. Add an Anthropic counterpart to the Codex cooldown recovery probe worker. On the shared state sweep it forces a fresh-dispatch usage read for every instance account whose cooldown is claimable, through fetchAnthropicAccountQuota so the claim, generation fence and settlement are the existing ones. A read that keeps the cooldown backs off per account (one sweep interval, doubling to ten minutes); a newer refusal resets the backoff; a deadline the sweeper reaches first is never read for; 403 entitlement refusals are never claimed. Co-Authored-By: Claude Fable 5.1 --- .../docs/reference/cli/providers-accounts.md | 5 + scripts/test-layout/layout.json | 2 +- .../anthropic-cooldown-recovery-worker.ts | 118 ++++++++++++++++ src/server/index.ts | 2 + structure/INDEX.md | 1 + structure/manifest.json | 1 + structure/providers/anthropic-account-pool.md | 14 +- ...anthropic-cooldown-recovery-worker.test.ts | 131 ++++++++++++++++++ tests/fixtures/test-layout-expected.json | 2 +- 9 files changed, 267 insertions(+), 9 deletions(-) create mode 100644 src/providers/quota/anthropic-cooldown-recovery-worker.ts create mode 100644 tests/adapters/anthropic/anthropic-cooldown-recovery-worker.test.ts diff --git a/docs-site/src/content/docs/reference/cli/providers-accounts.md b/docs-site/src/content/docs/reference/cli/providers-accounts.md index d0bfd8db0d7..95b0766b5d9 100644 --- a/docs-site/src/content/docs/reference/cli/providers-accounts.md +++ b/docs-site/src/content/docs/reference/cli/providers-accounts.md @@ -962,6 +962,11 @@ The command reports success even when no cooldown is active, with `cleared: fals an Anthropic cooldown also advances the account generation so an older in-flight quota probe cannot restore the cleared state or publish stale quota-derived eligibility afterward. +An Anthropic 429 quota cooldown also lifts on its own. The proxy re-reads the account's usage on its +background sweep (about once a minute, backing off while the window stays spent) and returns the +account to selection as soon as every rejected window shows headroom, for example after a reset spent +in the Claude app or a plan upgrade. `clear-cooldown` remains the manual override. + ### `ocx account refresh [--json]` For the Codex pool, use `ocx account refresh openai [--json]`. It force-refreshes account quotas and diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index de0c4fa181a..b33b442bce6 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -18,7 +18,7 @@ "anthropic-instance-isolation.test.ts": "adapters/anthropic", "anthropic-instance-pool-parity.test.ts": "adapters/anthropic", "anthropic-instance-quota.test.ts": "adapters/anthropic", "anthropic-instance-recovery.test.ts": "adapters/anthropic", "anthropic2-fast-parity.test.ts": "adapters/anthropic", "anthropic2-native-routing.test.ts": "claude-integration", - "oauth-anthropic-instance-health.test.ts": "oauth", "anthropic-quota-cooldown-recovery.test.ts": "adapters/anthropic", "anthropic-subscription-chain.test.ts": "adapters/anthropic", "combo-metered-consent.test.ts": "server", "combo-loopback-metered-consent.test.ts": "server", "anthropic-pool-usage-limit.test.ts": "adapters/anthropic", "oauth-account-plans.test.ts": "oauth", "anthropic2-responses-parity.test.ts": "responses", + "oauth-anthropic-instance-health.test.ts": "oauth", "anthropic-quota-cooldown-recovery.test.ts": "adapters/anthropic", "anthropic-cooldown-recovery-worker.test.ts": "adapters/anthropic", "anthropic-subscription-chain.test.ts": "adapters/anthropic", "combo-metered-consent.test.ts": "server", "combo-loopback-metered-consent.test.ts": "server", "anthropic-pool-usage-limit.test.ts": "adapters/anthropic", "oauth-account-plans.test.ts": "oauth", "anthropic2-responses-parity.test.ts": "responses", "oauth-anthropic-instance-multiprocess.test.ts": "oauth", "anthropic2-discovery-ownership.test.ts": "oauth", "codex-shim-runtime-preflight.test.ts": "codex-integration", diff --git a/src/providers/quota/anthropic-cooldown-recovery-worker.ts b/src/providers/quota/anthropic-cooldown-recovery-worker.ts new file mode 100644 index 00000000000..8caa8f773c5 --- /dev/null +++ b/src/providers/quota/anthropic-cooldown-recovery-worker.ts @@ -0,0 +1,118 @@ +import type { OcxConfig } from "../../types"; +import { ANTHROPIC_INSTANCE_IDS, type AnthropicInstanceId } from "../anthropic-instance-id"; +import { getAccountSet } from "../../oauth/store"; +import { anthropicRoutingFor } from "../../oauth/anthropic-routing"; +import { registerStateSweepAfterTick, STATE_SWEEP_INTERVAL_MS } from "../../lib/state-store-sweeper"; +import { fetchAnthropicAccountQuota } from "./anthropic-account-quota"; + +/** + * Background recovery for Anthropic quota cooldowns. + * + * A 429 cools the account until the deadline upstream named, and a fresh usage read that shows + * every rejected window open lifts it early (#6989). Until now that read only happened when + * something else asked for quota: the dashboard, `ocx account refresh`, or a reset admitted + * through this proxy. A Claude Code user whose window was reset elsewhere (a reset grant spent + * in the Claude app, a plan upgrade) kept receiving "temporarily rate-limited" from a pool + * whose accounts upstream had already reopened, until the old deadline or a manual + * clear-cooldown. + * + * This worker issues that read itself, on the shared state sweep, for every account whose + * cooldown is claimable. The read goes through `fetchAnthropicAccountQuota`, so the claim, + * generation fence and settlement are the same ones a dashboard read uses; the worker adds only + * the trigger and its pacing. + */ + +/** First retry after a read that kept the cooldown. Doubles per retained read up to the cap. */ +export const ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MIN_MS = STATE_SWEEP_INTERVAL_MS; +export const ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MAX_MS = 10 * 60_000; + +type RecoveryAttempt = { generation: number; nextAt: number; delayMs: number }; +type DueProbe = { instance: AnthropicInstanceId; accountId: string; key: string; generation: number }; + +/** Keyed per instance and account; bound to the cooldown generation the backoff was earned by. */ +const attempts = new Map(); +let inFlight: Promise | null = null; + +function attemptKey(instance: AnthropicInstanceId, accountId: string): string { + return `${instance}\0${accountId}`; +} + +function anthropicInstanceProbesUsage(config: OcxConfig, instance: AnthropicInstanceId): boolean { + const target = config.providers[instance]; + return !!target && target.disabled !== true && target.authMode === "oauth"; +} + +/** Cooled accounts whose refusal a usage read may lift and whose backoff has elapsed. */ +export function dueAnthropicCooldownRecoveryProbes(config: OcxConfig, now = Date.now()): DueProbe[] { + const due: DueProbe[] = []; + const live = new Set(); + for (const instance of ANTHROPIC_INSTANCE_IDS) { + if (!anthropicInstanceProbesUsage(config, instance)) continue; + const set = getAccountSet(instance); + if (!set) continue; + const routing = anthropicRoutingFor(instance); + for (const row of set.accounts) { + const key = attemptKey(instance, row.id); + const claim = routing.captureAnthropicCooldownRecovery(row.id, now); + if (!claim) continue; + live.add(key); + // A deadline the sweeper reaches first needs no upstream read. + const until = routing.getAnthropicAccountHealthSnapshot(row.id, now)?.cooldownUntil; + if (until === undefined || until - now <= STATE_SWEEP_INTERVAL_MS) continue; + const attempt = attempts.get(key); + if (attempt && attempt.generation === claim.cooldownGeneration && attempt.nextAt > now) continue; + due.push({ instance, accountId: row.id, key, generation: claim.cooldownGeneration }); + } + } + // A lifted, expired or removed cooldown drops its backoff with it. + for (const key of attempts.keys()) if (!live.has(key)) attempts.delete(key); + return due; +} + +function settleAttempt(probe: DueProbe, now: number): void { + const claim = anthropicRoutingFor(probe.instance).captureAnthropicCooldownRecovery(probe.accountId, now); + // Lifted, or replaced by a newer refusal that earns its own first read on the next sweep. + if (!claim || claim.cooldownGeneration !== probe.generation) { attempts.delete(probe.key); return; } + const previous = attempts.get(probe.key); + const delayMs = previous?.generation === probe.generation + ? Math.min(previous.delayMs * 2, ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MAX_MS) + : ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MIN_MS; + attempts.set(probe.key, { generation: probe.generation, nextAt: now + delayMs, delayMs }); +} + +/** + * One pass over every due cooled account. Single-flight: a sweep that fires while a pass is + * still reading joins it. Reads are sequential; a pool rarely has more than a few cooled accounts + * and this must never stampede the usage endpoint. + */ +export async function runAnthropicCooldownRecoveryProbes(config: OcxConfig, now = Date.now()): Promise { + if (inFlight) return inFlight; + inFlight = (async () => { + for (const probe of dueAnthropicCooldownRecoveryProbes(config, now)) { + try { + // Force bypasses the TTL; fresh dispatch refuses to join an older flight whose evidence + // predates the claim. Settlement happens inside the probe under the generation fence. + await fetchAnthropicAccountQuota(probe.instance, probe.accountId, true, config, true); + } catch { + // A stale or disowned probe settles nothing; the cooldown is kept and retried on backoff. + } + settleAttempt(probe, Date.now()); + } + })().catch(() => { + // Background recovery is best-effort; routing keeps the cooldown on failure. + }).finally(() => { inFlight = null; }); + return inFlight; +} + +export function registerAnthropicCooldownRecoveryProbeWorker(config: OcxConfig): () => void { + return registerStateSweepAfterTick({ + name: "anthropic-cooldown-recovery", + afterTick: () => { void runAnthropicCooldownRecoveryProbes(config); }, + }); +} + +/** Test-only: drop the backoff table and the single-flight promise. */ +export function clearAnthropicCooldownRecoveryProbeStateForTests(): void { + attempts.clear(); + inFlight = null; +} diff --git a/src/server/index.ts b/src/server/index.ts index a93927e265b..fcd20fcb33e 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -50,6 +50,7 @@ import { } from "../codex/auth-api"; import { activateResetCreditAutoRedeem } from "../codex/reset-credit-auto-redeem"; import { registerCodexQuotaAutoRefreshWorker } from "../codex/quota-auto-refresh"; +import { registerAnthropicCooldownRecoveryProbeWorker } from "../providers/quota/anthropic-cooldown-recovery-worker"; import { reconcileLiveStateStores, setLiveStateStoreConfig, @@ -310,6 +311,7 @@ function startServerWithSpendLedgerOwner(port: number | undefined, deps: StartSe // this line is the only one appending to usage.jsonl and the only one that may compact it. setUsageLedgerRetention(config.usageLedgerMaxBytes); registerCodexCooldownRecoveryProbeWorker(config); + registerAnthropicCooldownRecoveryProbeWorker(config); // Issue #42 Phase 3: opt-in archived auto-cleanup (default OFF). Unref'd hourly // tick for daily/weekly; startup evaluation is fire-and-forget after listen. // Heavy work runs in a Worker via the single-flight job controller. diff --git a/structure/INDEX.md b/structure/INDEX.md index 898fa09a543..0a27624f0f9 100644 --- a/structure/INDEX.md +++ b/structure/INDEX.md @@ -412,6 +412,7 @@ A source area can be described by more than one doc, because these docs are orga | `src/providers/quota-decision-publication.ts` | [`providers/jev-decision.md`](providers/jev-decision.md) | | `src/providers/quota-decision-snapshot.ts` | [`providers/jev-decision.md`](providers/jev-decision.md) | | `src/providers/quota/anthropic-account-quota.ts` | [`providers/anthropic-account-pool.md`](providers/anthropic-account-pool.md) | +| `src/providers/quota/anthropic-cooldown-recovery-worker.ts` | [`providers/anthropic-account-pool.md`](providers/anthropic-account-pool.md) | | `src/providers/quota/dsh-account.ts` | [`providers/dsh-account.md`](providers/dsh-account.md) | | `src/providers/quota/workbuddy.ts` | [`providers/workbuddy.md`](providers/workbuddy.md) | | `src/providers/request-pacing.ts` | [`providers/anthropic-key-billing.md`](providers/anthropic-key-billing.md) | diff --git a/structure/manifest.json b/structure/manifest.json index 51e51a7bce5..453f7d0d622 100644 --- a/structure/manifest.json +++ b/structure/manifest.json @@ -738,6 +738,7 @@ "src/oauth/", "src/providers/", "src/providers/quota/anthropic-account-quota.ts", + "src/providers/quota/anthropic-cooldown-recovery-worker.ts", "src/server/management/anthropic-pool-settings.ts", "src/sidecar/anthropic-binding.ts" ] diff --git a/structure/providers/anthropic-account-pool.md b/structure/providers/anthropic-account-pool.md index 648a675fd58..830b83fd2af 100644 --- a/structure/providers/anthropic-account-pool.md +++ b/structure/providers/anthropic-account-pool.md @@ -104,13 +104,13 @@ instance. Legacy named exports mean the primary pool. Model routes widen only within that instance's roster. Pure protocol/model transformations remain shared. Pause, policy and selection notifications affect only the matching existing state. -`src/oauth/anthropic-model-quota.ts`, `src/oauth/anthropic-rate-limit-policy.ts` -and `src/providers/quota/anthropic-cooldown-recovery.ts` expose corresponding -instance-bound owners. Family leases, admission pauses and probe generations -include the instance and account. Clear/removal fences prevent an older claim from -becoming current after the same account ID is re-added. -`src/lib/state-store-registrations.ts` sweeps and reconciles existing instance -buckets without starting dormant pools. +`src/oauth/anthropic-model-quota.ts`, `src/oauth/anthropic-rate-limit-policy.ts` and `src/providers/quota/anthropic-cooldown-recovery.ts` +expose corresponding instance-bound owners. Family leases, admission pauses and probe generations include the instance and +account. Clear/removal fences prevent an older claim from becoming current after the same account ID is re-added. +`src/lib/state-store-registrations.ts` sweeps and reconciles existing instance buckets without starting dormant pools. On that +sweep `src/providers/quota/anthropic-cooldown-recovery-worker.ts` forces the fresh usage read that lifts a claimable 429 quota +cooldown (#6989) for every instance and account, so recovery no longer waits for the dashboard, `ocx account refresh` or a reset +admitted here; a read that keeps it backs off per account (one sweep interval, doubling to ten minutes), a newer refusal resets that backoff, and a deadline the sweeper reaches first is never read for. Responses and native Messages retain the configured instance and authorized target through preparation, retries and continuations. Named Pool 2 OAuth routes diff --git a/tests/adapters/anthropic/anthropic-cooldown-recovery-worker.test.ts b/tests/adapters/anthropic/anthropic-cooldown-recovery-worker.test.ts new file mode 100644 index 00000000000..62eeba3d047 --- /dev/null +++ b/tests/adapters/anthropic/anthropic-cooldown-recovery-worker.test.ts @@ -0,0 +1,131 @@ +/** + * The background recovery worker issues the fresh usage read that lifts a 429 quota cooldown + * (#6989) without waiting for the dashboard, `ocx account refresh` or a reset admitted through + * this proxy. A read that keeps the cooldown backs off; a lifted or replaced refusal forgets it. + */ +import { afterEach, beforeEach, expect, test } from "bun:test"; +import { createAnthropicInstanceFixture, INSTANCE_FIXTURE_INSTANCES, type AnthropicInstanceFixture } from "../../helpers/anthropic-instance-fixture"; +import { + ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MAX_MS, + ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MIN_MS, + clearAnthropicCooldownRecoveryProbeStateForTests, + dueAnthropicCooldownRecoveryProbes, + runAnthropicCooldownRecoveryProbes, +} from "../../../src/providers/quota/anthropic-cooldown-recovery-worker"; + +let f: AnthropicInstanceFixture; +beforeEach(async () => { f = await createAnthropicInstanceFixture(); await f.seed(); clearAnthropicCooldownRecoveryProbeStateForTests(); }); +afterEach(async () => { clearAnthropicCooldownRecoveryProbeStateForTests(); await f?.dispose(); }); + +const HOUR = 3_600_000; +const DAY = 24 * HOUR; + +function rejected(windows: readonly ("5h" | "7d")[], resetAt: number): Headers { + const headers = new Headers(); + for (const window of windows) { + headers.set(`anthropic-ratelimit-unified-${window}-status`, "rejected"); + headers.set(`anthropic-ratelimit-unified-${window}-reset`, String(Math.floor(resetAt / 1000))); + } + return headers; +} + +/** Fake usage endpoint; counts dispatched reads. */ +function usage(fiveHour: number, weekly: number): { reads: number } { + const counter = { reads: 0 }; + globalThis.fetch = (async () => { + counter.reads += 1; + const resetsAt = new Date(Date.now() + HOUR).toISOString(); + return Response.json({ + five_hour: { utilization: fiveHour, resets_at: resetsAt }, + seven_day: { utilization: weekly, resets_at: resetsAt }, + limits: [], + }); + }) as typeof fetch; + return counter; +} + +function coolForDays(instance: "anthropic" | "anthropic2", id: string, days: number, now = Date.now()): void { + const routing = f.routing.anthropicRoutingFor(instance); + // The reported shape: a spent window whose Retry-After equals the window reset. + routing.recordAnthropicAccount429(f.config, id, String(days * 24 * 3600), now, rejected(["5h", "7d"], now + days * DAY)); + expect(routing.getAnthropicAccountHealthSnapshot(id, now)).toMatchObject({ cooldownSource: "retry-after" }); +} + +for (const instance of INSTANCE_FIXTURE_INSTANCES) { + test(`${instance}: the sweep lifts a Retry-After quota cooldown once upstream shows headroom`, async () => { + const routing = f.routing.anthropicRoutingFor(instance); + const id = f.ids[0]; + coolForDays(instance, id, 4); + const counter = usage(0, 0); + + await runAnthropicCooldownRecoveryProbes(f.config); + + expect(counter.reads).toBe(1); + expect(routing.getAnthropicAccountHealthSnapshot(id)).toBeNull(); + expect(routing.getEligibleAnthropicAccounts()).toContain(id); + expect(dueAnthropicCooldownRecoveryProbes(f.config)).toEqual([]); + }); +} + +test("a read that still shows a rejected window full keeps the cooldown and backs off", async () => { + const routing = f.routing.anthropicRoutingFor("anthropic"); + const id = f.ids[0]; + const start = Date.now(); + coolForDays("anthropic", id, 4, start); + const counter = usage(0, 100); + + await runAnthropicCooldownRecoveryProbes(f.config, start); + expect(counter.reads).toBe(1); + expect(routing.getAnthropicAccountHealthSnapshot(id)).toMatchObject({ cooldownSource: "retry-after" }); + + // The next sweep inside the backoff sends nothing; the one after it reads again. + await runAnthropicCooldownRecoveryProbes(f.config, Date.now()); + expect(counter.reads).toBe(1); + await runAnthropicCooldownRecoveryProbes(f.config, Date.now() + ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MIN_MS + 1); + expect(counter.reads).toBe(2); + // Each retained read doubles the wait, up to the cap. + await runAnthropicCooldownRecoveryProbes(f.config, Date.now() + ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MIN_MS + 1); + expect(counter.reads).toBe(2); + await runAnthropicCooldownRecoveryProbes(f.config, Date.now() + ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MAX_MS + 1); + expect(counter.reads).toBe(3); +}); + +test("a newer refusal replaces the earned backoff and is read on the next sweep", async () => { + const routing = f.routing.anthropicRoutingFor("anthropic"); + const id = f.ids[0]; + const start = Date.now(); + coolForDays("anthropic", id, 4, start); + const counter = usage(0, 100); + await runAnthropicCooldownRecoveryProbes(f.config, start); + expect(counter.reads).toBe(1); + + routing.recordAnthropicAccount429(f.config, id, String(4 * 24 * 3600), Date.now(), rejected(["5h", "7d"], Date.now() + 4 * DAY)); + await runAnthropicCooldownRecoveryProbes(f.config, Date.now()); + expect(counter.reads).toBe(2); +}); + +test("no cooled account sends nothing", async () => { + const counter = usage(0, 0); + await runAnthropicCooldownRecoveryProbes(f.config); + expect(counter.reads).toBe(0); +}); + +test("a cooldown the sweeper reaches first is left to expire", async () => { + const routing = f.routing.anthropicRoutingFor("anthropic"); + const id = f.ids[0]; + const now = Date.now(); + routing.recordAnthropicAccount429(f.config, id, "30", now, rejected(["5h"], now + 30_000)); + const counter = usage(0, 0); + await runAnthropicCooldownRecoveryProbes(f.config, now); + expect(counter.reads).toBe(0); + expect(routing.getAnthropicAccountHealthSnapshot(id, now)).not.toBeNull(); +}); + +test("an entitlement 403 is never read for", async () => { + const routing = f.routing.anthropicRoutingFor("anthropic"); + const now = Date.now(); + routing.recordAnthropicAccountRefusal(f.config, f.ids[0], 403, String(4 * 24 * 3600), now); + const counter = usage(0, 0); + await runAnthropicCooldownRecoveryProbes(f.config, now); + expect(counter.reads).toBe(0); +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index d03860e91ff..015c23dfec8 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -166,7 +166,7 @@ "anthropic-instance-isolation.test.ts": "adapters/anthropic", "anthropic-instance-pool-parity.test.ts": "adapters/anthropic", "anthropic-instance-quota.test.ts": "adapters/anthropic", "anthropic-instance-recovery.test.ts": "adapters/anthropic", "anthropic2-fast-parity.test.ts": "adapters/anthropic", "anthropic2-native-routing.test.ts": "claude-integration", - "oauth-anthropic-instance-health.test.ts": "oauth", "anthropic-quota-cooldown-recovery.test.ts": "adapters/anthropic", "anthropic-subscription-chain.test.ts": "adapters/anthropic", "combo-metered-consent.test.ts": "server", "combo-loopback-metered-consent.test.ts": "server", "anthropic-pool-usage-limit.test.ts": "adapters/anthropic", "oauth-account-plans.test.ts": "oauth", "anthropic2-responses-parity.test.ts": "responses", + "oauth-anthropic-instance-health.test.ts": "oauth", "anthropic-quota-cooldown-recovery.test.ts": "adapters/anthropic", "anthropic-cooldown-recovery-worker.test.ts": "adapters/anthropic", "anthropic-subscription-chain.test.ts": "adapters/anthropic", "combo-metered-consent.test.ts": "server", "combo-loopback-metered-consent.test.ts": "server", "anthropic-pool-usage-limit.test.ts": "adapters/anthropic", "oauth-account-plans.test.ts": "oauth", "anthropic2-responses-parity.test.ts": "responses", "oauth-anthropic-instance-multiprocess.test.ts": "oauth", "anthropic2-discovery-ownership.test.ts": "oauth", "codex-shim-runtime-preflight.test.ts": "codex-integration", From edbe41c6b006df595f63c9d5df1f2997fd69953a Mon Sep 17 00:00:00 2001 From: JUN Date: Wed, 11 Nov 2026 18:17:42 +0900 Subject: [PATCH 2/3] fix(anthropic-pool): detach the recovery worker and qualify which cooldowns it lifts Carry of #8504 review findings: register the worker inside the startup transaction and detach it on a failed start and on shutdown; state that only window-backed 429 cooldowns on enabled OAuth instances recover early; test the doubled and capped backoff on one controlled clock; add the windowless Retry-After case. Co-authored-by: Bruno --- .../docs/reference/cli/providers-accounts.md | 10 ++-- src/server/index.ts | 5 +- structure/providers/anthropic-account-pool.md | 6 +- ...anthropic-cooldown-recovery-worker.test.ts | 58 +++++++++++++------ tests/lab/core-lab-boundary.test.ts | 1 + 5 files changed, 53 insertions(+), 27 deletions(-) diff --git a/docs-site/src/content/docs/reference/cli/providers-accounts.md b/docs-site/src/content/docs/reference/cli/providers-accounts.md index 95b0766b5d9..d3d3a29f3ca 100644 --- a/docs-site/src/content/docs/reference/cli/providers-accounts.md +++ b/docs-site/src/content/docs/reference/cli/providers-accounts.md @@ -962,10 +962,12 @@ The command reports success even when no cooldown is active, with `cleared: fals an Anthropic cooldown also advances the account generation so an older in-flight quota probe cannot restore the cleared state or publish stale quota-derived eligibility afterward. -An Anthropic 429 quota cooldown also lifts on its own. The proxy re-reads the account's usage on its -background sweep (about once a minute, backing off while the window stays spent) and returns the -account to selection as soon as every rejected window shows headroom, for example after a reset spent -in the Claude app or a plan upgrade. `clear-cooldown` remains the manual override. +An Anthropic 429 quota cooldown that names its rejected usage windows also lifts on its own. For +enabled OAuth instances, the proxy re-reads the account's usage on its background sweep (about once a +minute, backing off to every ten minutes while the window stays spent) and returns the account to +selection as soon as every rejected window shows headroom, for example after a reset spent in the +Claude app or a plan upgrade. A 429 that carries only `Retry-After`, with no window to re-read, keeps +its cooldown until that deadline. `clear-cooldown` remains the manual override for both. ### `ocx account refresh [--json]` diff --git a/src/server/index.ts b/src/server/index.ts index fcd20fcb33e..60ef2025b59 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -311,7 +311,6 @@ function startServerWithSpendLedgerOwner(port: number | undefined, deps: StartSe // this line is the only one appending to usage.jsonl and the only one that may compact it. setUsageLedgerRetention(config.usageLedgerMaxBytes); registerCodexCooldownRecoveryProbeWorker(config); - registerAnthropicCooldownRecoveryProbeWorker(config); // Issue #42 Phase 3: opt-in archived auto-cleanup (default OFF). Unref'd hourly // tick for daily/weekly; startup evaluation is fire-and-forget after listen. // Heavy work runs in a Worker via the single-flight job controller. @@ -612,6 +611,7 @@ function startServerWithSpendLedgerOwner(port: number | undefined, deps: StartSe const linkPolicy = (): RequestPolicyView => requestPolicyView(config, LINK_INGRESS_HOSTNAME, { allowedKeyIds: optionalListeners.linkAdmissionKeyIds() }); let backgroundLifecycle: ReturnType | null = null; let unregisterQuotaAutoRefresh: (() => void) | null = null; + let unregisterAnthropicCooldownRecovery: (() => void) | null = null; let remoteWorkspaceStopping = false; let remoteWorkspaceShutdown: (() => Promise) | undefined; let desktopCompatibilityStartup: ReturnType | undefined; @@ -638,6 +638,7 @@ function startServerWithSpendLedgerOwner(port: number | undefined, deps: StartSe backgroundLifecycle = acquireServerBackgroundLifecycle(applyPolicy, config); unregisterQuotaAutoRefresh = (deps.registerCodexQuotaAutoRefreshWorker ?? registerCodexQuotaAutoRefreshWorker)(config); + unregisterAnthropicCooldownRecovery = registerAnthropicCooldownRecoveryProbeWorker(config); // Poll external pricing edits; the startup catch releases this owner-scoped lease. userCostOverlayReconciler = startUserCostOverlayReconciler({ liveConfig: config }); const serveOptions = createServeOptions({ @@ -715,6 +716,7 @@ function startServerWithSpendLedgerOwner(port: number | undefined, deps: StartSe maxRequestBodySize: inboundBodyLimitBytes, dispatch: (req, requestServer) => serveOptions.fetch(req, requestServer) }); } catch (error) { unregisterQuotaAutoRefresh?.(); + unregisterAnthropicCooldownRecovery?.(); userCostOverlayReconciler?.stop(); backgroundLifecycle?.releaseAfterFailedStart(); void nativeMainLifecycle.release(); @@ -760,6 +762,7 @@ function startServerWithSpendLedgerOwner(port: number | undefined, deps: StartSe userCostOverlayReconciler?.stop(); } finally { unregisterQuotaAutoRefresh?.(); + unregisterAnthropicCooldownRecovery?.(); } }, ], diff --git a/structure/providers/anthropic-account-pool.md b/structure/providers/anthropic-account-pool.md index 830b83fd2af..82a9d7de865 100644 --- a/structure/providers/anthropic-account-pool.md +++ b/structure/providers/anthropic-account-pool.md @@ -108,9 +108,9 @@ Pause, policy and selection notifications affect only the matching existing stat expose corresponding instance-bound owners. Family leases, admission pauses and probe generations include the instance and account. Clear/removal fences prevent an older claim from becoming current after the same account ID is re-added. `src/lib/state-store-registrations.ts` sweeps and reconciles existing instance buckets without starting dormant pools. On that -sweep `src/providers/quota/anthropic-cooldown-recovery-worker.ts` forces the fresh usage read that lifts a claimable 429 quota -cooldown (#6989) for every instance and account, so recovery no longer waits for the dashboard, `ocx account refresh` or a reset -admitted here; a read that keeps it backs off per account (one sweep interval, doubling to ten minutes), a newer refusal resets that backoff, and a deadline the sweeper reaches first is never read for. +sweep `src/providers/quota/anthropic-cooldown-recovery-worker.ts` forces the usage read that lifts a claimable 429 quota cooldown +(#6989) on enabled OAuth instances only; a windowless `Retry-After` cooldown is unclaimable and waits for its deadline. Retained +reads back off per account (one sweep, doubling to ten minutes); a newer refusal resets it; a deadline due first is never read for. Responses and native Messages retain the configured instance and authorized target through preparation, retries and continuations. Named Pool 2 OAuth routes diff --git a/tests/adapters/anthropic/anthropic-cooldown-recovery-worker.test.ts b/tests/adapters/anthropic/anthropic-cooldown-recovery-worker.test.ts index 62eeba3d047..dbf1482693e 100644 --- a/tests/adapters/anthropic/anthropic-cooldown-recovery-worker.test.ts +++ b/tests/adapters/anthropic/anthropic-cooldown-recovery-worker.test.ts @@ -3,7 +3,7 @@ * (#6989) without waiting for the dashboard, `ocx account refresh` or a reset admitted through * this proxy. A read that keeps the cooldown backs off; a lifted or replaced refusal forgets it. */ -import { afterEach, beforeEach, expect, test } from "bun:test"; +import { afterEach, beforeEach, expect, spyOn, test } from "bun:test"; import { createAnthropicInstanceFixture, INSTANCE_FIXTURE_INSTANCES, type AnthropicInstanceFixture } from "../../helpers/anthropic-instance-fixture"; import { ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MAX_MS, @@ -67,27 +67,47 @@ for (const instance of INSTANCE_FIXTURE_INSTANCES) { }); } -test("a read that still shows a rejected window full keeps the cooldown and backs off", async () => { +test("a read that still shows a rejected window full keeps the cooldown and backs off to the cap", async () => { const routing = f.routing.anthropicRoutingFor("anthropic"); const id = f.ids[0]; - const start = Date.now(); - coolForDays("anthropic", id, 4, start); - const counter = usage(0, 100); - - await runAnthropicCooldownRecoveryProbes(f.config, start); - expect(counter.reads).toBe(1); - expect(routing.getAnthropicAccountHealthSnapshot(id)).toMatchObject({ cooldownSource: "retry-after" }); + // One clock for the sweep and for settlement, so each deadline is checked on both sides. + let clock = Date.now(); + const now = spyOn(Date, "now").mockImplementation(() => clock); + try { + coolForDays("anthropic", id, 4, clock); + const counter = usage(0, 100); + await runAnthropicCooldownRecoveryProbes(f.config); + expect(counter.reads).toBe(1); + expect(routing.getAnthropicAccountHealthSnapshot(id)).toMatchObject({ cooldownSource: "retry-after" }); + + // Retained reads wait MIN, 2*MIN, 4*MIN, ... and never longer than MAX. + let delay = ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MIN_MS; + for (let read = 2; read <= 7; read += 1) { + const settledAt = clock; + clock = settledAt + delay - 1; + await runAnthropicCooldownRecoveryProbes(f.config); + expect(counter.reads).toBe(read - 1); + clock = settledAt + delay; + await runAnthropicCooldownRecoveryProbes(f.config); + expect(counter.reads).toBe(read); + delay = Math.min(delay * 2, ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MAX_MS); + } + // Five doublings from one minute pass the ten-minute cap, so the last waits were capped. + expect(delay).toBe(ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MAX_MS); + } finally { + now.mockRestore(); + } +}); - // The next sweep inside the backoff sends nothing; the one after it reads again. - await runAnthropicCooldownRecoveryProbes(f.config, Date.now()); - expect(counter.reads).toBe(1); - await runAnthropicCooldownRecoveryProbes(f.config, Date.now() + ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MIN_MS + 1); - expect(counter.reads).toBe(2); - // Each retained read doubles the wait, up to the cap. - await runAnthropicCooldownRecoveryProbes(f.config, Date.now() + ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MIN_MS + 1); - expect(counter.reads).toBe(2); - await runAnthropicCooldownRecoveryProbes(f.config, Date.now() + ANTHROPIC_COOLDOWN_RECOVERY_PROBE_MAX_MS + 1); - expect(counter.reads).toBe(3); +test("a unified-rejected 429 that names no window keeps its Retry-After cooldown unread", async () => { + const routing = f.routing.anthropicRoutingFor("anthropic"); + const now = Date.now(); + routing.recordAnthropicAccount429(f.config, f.ids[0], String(4 * 24 * 3600), now, + new Headers({ "anthropic-ratelimit-unified-status": "rejected" })); + expect(routing.getAnthropicAccountHealthSnapshot(f.ids[0], now)).not.toBeNull(); + const counter = usage(0, 0); + await runAnthropicCooldownRecoveryProbes(f.config, now); + expect(counter.reads).toBe(0); }); test("a newer refusal replaces the earned backoff and is read on the next sweep", async () => { diff --git a/tests/lab/core-lab-boundary.test.ts b/tests/lab/core-lab-boundary.test.ts index 5b04d204d08..c307b35e45b 100644 --- a/tests/lab/core-lab-boundary.test.ts +++ b/tests/lab/core-lab-boundary.test.ts @@ -1059,6 +1059,7 @@ describe("activation window stays synchronous", () => { */ const UNRESOLVED_CALLEES: Record = { unregisterQuotaAutoRefresh: "let-binding holding the return of registerCodexQuotaAutoRefreshWorker, optional-called on the bind-failure path. There is no `function unregisterQuotaAutoRefresh` to inspect; following the assignment would be depth 2.", + unregisterAnthropicCooldownRecovery: "let-binding holding the return of registerAnthropicCooldownRecoveryProbeWorker (a synchronous sweep registration), optional-called on the bind-failure path. Same shape as unregisterQuotaAutoRefresh.", }; From 278656f4f56126bb7618a20d4c28537c2be20b93 Mon Sep 17 00:00:00 2001 From: JUN Date: Wed, 11 Nov 2026 18:34:00 +0900 Subject: [PATCH 3/3] fix(anthropic-pool): never read usage for a cooldown a read cannot lift A reset-derived 429 that names no rejected window is claimable, but settlement always retains it, so the worker re-read usage every one to ten minutes until a deadline days away. The claim now reports whether it names a window and the worker skips one that does not, along with paused and signed-out accounts. Adds the reset-derived, disabled/API-key instance and backoff-pruning cases. Co-authored-by: Bruno --- src/oauth/anthropic-routing.ts | 7 ++- .../anthropic-cooldown-recovery-worker.ts | 5 ++- ...anthropic-cooldown-recovery-worker.test.ts | 44 +++++++++++++++++++ 3 files changed, 53 insertions(+), 3 deletions(-) diff --git a/src/oauth/anthropic-routing.ts b/src/oauth/anthropic-routing.ts index 36ffe9b0587..305a3d46417 100644 --- a/src/oauth/anthropic-routing.ts +++ b/src/oauth/anthropic-routing.ts @@ -274,6 +274,8 @@ export type AnthropicCooldownRecoveryClaim = Readonly<{ accountId: string; cooldownGeneration: number; claimedAt: number; + /** False for a reset-derived refusal that named no window: a usage read can only retain it. */ + namesRejectedWindow: boolean; }>; export type AnthropicCooldownRecoverySettlement = "cleared" | "retained" | "superseded"; @@ -434,8 +436,9 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting now = Date.now(), ): AnthropicCooldownRecoveryClaim | null { const entry = upstreamHealth.get(accountId); - if (!entry || entry.cooldownUntil <= now || recoverableQuotaWindows(entry) === null) return null; - return { instance, accountId, cooldownGeneration: entry.cooldownGeneration, claimedAt: now }; + const windows = entry && entry.cooldownUntil > now ? recoverableQuotaWindows(entry) : null; + if (!entry || windows === null) return null; + return { instance, accountId, cooldownGeneration: entry.cooldownGeneration, claimedAt: now, namesRejectedWindow: windows.length > 0 }; } /** diff --git a/src/providers/quota/anthropic-cooldown-recovery-worker.ts b/src/providers/quota/anthropic-cooldown-recovery-worker.ts index 8caa8f773c5..bfbf835af0f 100644 --- a/src/providers/quota/anthropic-cooldown-recovery-worker.ts +++ b/src/providers/quota/anthropic-cooldown-recovery-worker.ts @@ -52,9 +52,12 @@ export function dueAnthropicCooldownRecoveryProbes(config: OcxConfig, now = Date if (!set) continue; const routing = anthropicRoutingFor(instance); for (const row of set.accounts) { + // A paused or signed-out account is refused by the probe before any request; skip it here. + if (row.paused === true || row.needsReauth === true) continue; const key = attemptKey(instance, row.id); const claim = routing.captureAnthropicCooldownRecovery(row.id, now); - if (!claim) continue; + // A refusal that named no window settles as retained whatever usage shows: never read for it. + if (!claim || !claim.namesRejectedWindow) continue; live.add(key); // A deadline the sweeper reaches first needs no upstream read. const until = routing.getAnthropicAccountHealthSnapshot(row.id, now)?.cooldownUntil; diff --git a/tests/adapters/anthropic/anthropic-cooldown-recovery-worker.test.ts b/tests/adapters/anthropic/anthropic-cooldown-recovery-worker.test.ts index dbf1482693e..884c7b9a01b 100644 --- a/tests/adapters/anthropic/anthropic-cooldown-recovery-worker.test.ts +++ b/tests/adapters/anthropic/anthropic-cooldown-recovery-worker.test.ts @@ -149,3 +149,47 @@ test("an entitlement 403 is never read for", async () => { await runAnthropicCooldownRecoveryProbes(f.config, now); expect(counter.reads).toBe(0); }); +test("a reset-derived 429 that names no window is never read for", async () => { + const routing = f.routing.anthropicRoutingFor("anthropic"); + const now = Date.now(); + routing.recordAnthropicAccount429(f.config, f.ids[0], null, now, new Headers({ + "anthropic-ratelimit-unified-status": "rejected", + "anthropic-ratelimit-unified-reset": String(Math.floor((now + 4 * DAY) / 1000)), + })); + expect(routing.getAnthropicAccountHealthSnapshot(f.ids[0], now)).toMatchObject({ cooldownSource: "reset-derived" }); + const counter = usage(0, 0); + await runAnthropicCooldownRecoveryProbes(f.config, now); + expect(counter.reads).toBe(0); + expect(dueAnthropicCooldownRecoveryProbes(f.config, now)).toEqual([]); +}); + +test("a disabled or API-key instance is never read for", async () => { + const id = f.ids[0]; + coolForDays("anthropic", id, 4); + const counter = usage(0, 0); + const provider = f.config.providers.anthropic!; + provider.disabled = true; + await runAnthropicCooldownRecoveryProbes(f.config); + provider.disabled = false; + const authMode = provider.authMode; + provider.authMode = "key" as typeof authMode; + await runAnthropicCooldownRecoveryProbes(f.config); + provider.authMode = authMode; + expect(counter.reads).toBe(0); +}); + +test("disabling an instance forgets its earned backoff", async () => { + const id = f.ids[0]; + const start = Date.now(); + coolForDays("anthropic", id, 4, start); + const counter = usage(0, 100); + await runAnthropicCooldownRecoveryProbes(f.config, start); + expect(counter.reads).toBe(1); + expect(dueAnthropicCooldownRecoveryProbes(f.config, Date.now())).toEqual([]); + const provider = f.config.providers.anthropic!; + provider.disabled = true; + expect(dueAnthropicCooldownRecoveryProbes(f.config, Date.now())).toEqual([]); + provider.disabled = false; + // Same cooldown generation, but the backoff row was pruned while the instance was off. + expect(dueAnthropicCooldownRecoveryProbes(f.config, Date.now()).map(probe => probe.accountId)).toEqual([id]); +});