diff --git a/mutations.toml b/mutations.toml index f6278336..14a13074 100644 --- a/mutations.toml +++ b/mutations.toml @@ -1233,6 +1233,22 @@ broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core broad_report = "tmp/mutations/core.xml" broad_id = "{name}" +[[control]] +id = "relay-direct-fallback-sends-once" +guards = "A direct fallback that fails after it started must not be sent a second time by sendViaRelay's outer catch: its request may already have reached Anthropic." +file = "packages/core/src/relay.ts" +old = " directFallbackStarted = true" +new = " directFallbackStarted = false // NON-VACUITY BREAK: forget that the direct fallback started, so a failure is sent again" +test_file = "packages/core/src/tests/relay-authorization.test.ts" +runner = "command" +command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/relay-authorization.test.ts", "{test}"] +test_count_pattern = "CKDEV executed {count} tests" +expect_red = ["a direct fallback after a relay 5xx that fails is not sent again", "a direct fallback after a relay-only 401 that fails is not sent again", "a failed WebSocket relay whose HTTP relay fallback falls back direct and fails is not sent again"] +expect_message = "error: expect(received).toBe(expected)" +broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] +broad_report = "tmp/mutations/core.xml" +broad_id = "{name}" + [[control]] id = "quota-call-pause-default-waits" guards = "Without an injected pause, QuotaManager waits on a real timer so physical quota GETs start at least one second apart." diff --git a/packages/core/src/relay.ts b/packages/core/src/relay.ts index af7ecce5..2663d1aa 100644 --- a/packages/core/src/relay.ts +++ b/packages/core/src/relay.ts @@ -1867,6 +1867,15 @@ export async function sendViaRelay(options: { responseMode ??= requestResponseMode(bodyText) return responseMode } + // The direct fallback runs at most once per request. Once it has started, + // its request may already have reached Anthropic (a connection can reset + // after the body was sent), so its failure goes to the caller instead of + // sending the same request again. + let directFallbackStarted = false + const directFallback = () => { + directFallbackStarted = true + return fallback() + } try { let result: RelaySendResult const sendStart = perfNowMs() @@ -1902,7 +1911,7 @@ export async function sendViaRelay(options: { payload, bodyText, requestMode, - fallback, + fallback: directFallback, signal: init?.signal, authorizeAttempt, }) @@ -1913,7 +1922,7 @@ export async function sendViaRelay(options: { payload, bodyText, requestMode, - fallback, + fallback: directFallback, signal: init?.signal, authorizeAttempt, }) @@ -1983,6 +1992,12 @@ export async function sendViaRelay(options: { // The upstream already answered this request; a direct fallback would // send it a second time. if (error instanceof RelayResponseMetadataError) throw error + if (directFallbackStarted) { + relayLog( + `direct fallback failed session=${shortAffinity(affinity)}; not sending it again code=relay_transport_error`, + ) + throw error + } if (!config.fallbackToDirect) { relayLog( `relay failed session=${shortAffinity(affinity)} and fallbackToDirect=false code=relay_transport_error`, diff --git a/packages/core/src/tests/relay-authorization.test.ts b/packages/core/src/tests/relay-authorization.test.ts index 14021c86..8be0b38f 100644 --- a/packages/core/src/tests/relay-authorization.test.ts +++ b/packages/core/src/tests/relay-authorization.test.ts @@ -173,6 +173,108 @@ test('a relay transport 401 without upstream provenance falls back direct withou } }) +// A direct fallback can fail after its request already reached Anthropic, +// for example when the connection resets after the body was sent. Sending it +// again would bill the same request twice, so its error goes to the caller. +for (const [label, status] of [ + ['a relay 5xx', 503], + ['a relay-only 401', 401], +] as const) { + test(`a direct fallback after ${label} that fails is not sent again`, async () => { + const server = Bun.serve({ + hostname: '127.0.0.1', + port: 0, + fetch: () => new Response('relay unavailable', { status }), + }) + let direct = 0 + try { + await expect( + sendViaRelay({ + config: config(server.url.href), + input: 'https://api.anthropic.com/v1/messages', + init: { method: 'POST' }, + headers: new Headers(), + body: body(`failed-fallback-${status}`), + affinity: randomUUID(), + fallback: async () => { + direct++ + throw new Error('connection reset after the request was sent') + }, + authorizeAttempt: async () => ({ + headers: new Headers({ authorization: 'Bearer test-access' }), + }), + }), + ).rejects.toThrow('connection reset after the request was sent') + expect(direct).toBe(1) + } finally { + await server.stop(true) + } + }) +} + +test('a failed WebSocket relay whose HTTP relay fallback falls back direct and fails is not sent again', async () => { + // Refusing the upgrade fails the WebSocket; the HTTP relay then answers 503. + const server = Bun.serve({ + hostname: '127.0.0.1', + port: 0, + fetch: () => new Response('relay unavailable', { status: 503 }), + }) + let direct = 0 + try { + await expect( + sendViaRelay({ + config: config(server.url.href, 'websocket'), + input: 'https://api.anthropic.com/v1/messages', + init: { method: 'POST' }, + headers: new Headers(), + body: body('failed-websocket-fallback'), + affinity: randomUUID(), + fallback: async () => { + direct++ + throw new Error('connection reset after the request was sent') + }, + authorizeAttempt: async () => ({ + headers: new Headers({ authorization: 'Bearer test-access' }), + }), + }), + ).rejects.toThrow('connection reset after the request was sent') + expect(direct).toBe(1) + } finally { + server.stop(true) + } +}) + +test('a relay that fails before any direct fallback still falls back direct once', async () => { + // A stopped server leaves a port that refuses connections, so the relay + // POST itself rejects before the relay could forward anything. + const server = Bun.serve({ + hostname: '127.0.0.1', + port: 0, + fetch: () => new Response('unused'), + }) + const url = server.url.href + await server.stop(true) + let direct = 0 + const response = await sendViaRelay({ + config: config(url), + input: 'https://api.anthropic.com/v1/messages', + init: { method: 'POST' }, + headers: new Headers(), + body: body('relay-unreachable'), + affinity: randomUUID(), + fallback: async () => { + direct++ + return new Response('direct-ok', { status: 200 }) + }, + authorizeAttempt: async () => ({ + headers: new Headers({ authorization: 'Bearer test-access' }), + }), + }) + expect(response.status).toBe(200) + expect(await response.text()).toBe('direct-ok') + expect(direct).toBe(1) +}) + test('optimistic WebSocket reconnect uses a fresh receipt without reporting its synthetic 200', async () => { const sent: Array<{ id: string diff --git a/packages/opencode/src/tests/index.test.ts b/packages/opencode/src/tests/index.test.ts index 30d1f5c3..fc49f2d0 100644 --- a/packages/opencode/src/tests/index.test.ts +++ b/packages/opencode/src/tests/index.test.ts @@ -19265,31 +19265,56 @@ describe('auth.loader', () => { }), ) const records: LogTestRecord[] = [] + const affinity = 'quota-relay-direct-fallback' + // A marker in this request's body tells its own sends apart from any + // other request that reaches the mock. When an earlier test overruns + // Bun's timeouts, its body and its teardown keep running into this test: + // its own model request can arrive here, and its teardown can re-enable + // profile hydration, which sends GET /api/oauth/profile. Counting those + // would report a duplicate send that this request never made. + const probe = 'relay-direct-fallback-probe' let relay503Calls = 0 let directCalls = 0 __setLogTestSink((record) => records.push(record)) globalThis.fetch = mock( - withNativeAdmission((input: string | URL | Request) => { - const url = extractUrl(input) - if (url === 'https://relay.example.test/') { - relay503Calls += 1 - return Promise.resolve( - new Response('relay unavailable', { status: 503 }), - ) - } - directCalls += 1 - return Promise.resolve( - new Response('direct', { headers: quotaHeaders }), - ) - }), + withNativeAdmission( + (input: string | URL | Request, init?: RequestInit) => { + const url = extractUrl(input) + const body = typeof init?.body === 'string' ? init.body : '' + if (url === 'https://relay.example.test/') { + let relayAffinity: unknown + try { + relayAffinity = JSON.parse(body).affinity + } catch {} + if (relayAffinity === affinity) relay503Calls += 1 + return Promise.resolve( + new Response('relay unavailable', { status: 503 }), + ) + } + if ( + url.startsWith(MESSAGES_URL) && + init?.method === 'POST' && + body.includes(probe) + ) { + directCalls += 1 + return Promise.resolve( + new Response('direct', { headers: quotaHeaders }), + ) + } + return Promise.resolve(new Response('unrelated', { status: 404 })) + }, + ), ) as unknown as typeof fetch const result = await loadFetch() setLogLevel('debug') try { const response = await result.fetch(MESSAGES_URL, { - ...EMPTY_POST, - headers: { 'x-session-affinity': 'quota-relay-direct-fallback' }, + method: 'POST', + body: JSON.stringify({ + messages: [{ role: 'user', content: probe }], + }), + headers: { 'x-session-affinity': affinity }, }) expect(await response.text()).toBe('direct') expect(relay503Calls).toBe(1)