fix(server): pass through final upstream errors (#2333)

Signed-off-by: RainbowBird <git@luoling.moe>
This commit is contained in:
RainbowBird
2026-09-06 16:08:47 +00:00
committed by GitHub
parent f166736a76
commit 96b629165b
13 changed files with 417 additions and 136 deletions
+1
View File
@@ -15,6 +15,7 @@ Create a reviewable PR from the exact commits intended for publication.
4. Publish the intended commits through the available GitHub/`gh` workflow. 4. Publish the intended commits through the available GitHub/`gh` workflow.
5. Compose the PR body with a concise summary, exact verification commands, and the required visual table. 5. Compose the PR body with a concise summary, exact verification commands, and the required visual table.
6. Create the PR, then open it and verify its title, base/head branches, body, and embedded images. 6. Create the PR, then open it and verify its title, base/head branches, body, and embedded images.
7. Get the PR review threads, comments, and check status. If a comment identifies a confirmed error, fix it, run focused checks, push the update, reply with evidence, and resolve the thread.
## Visual Evidence Workflow ## Visual Evidence Workflow
+1
View File
@@ -256,6 +256,7 @@ as a first language.
## PR / Workflow Tips ## PR / Workflow Tips
- When asked to create, open, publish, or prepare a pull request, always use the repo-local `create-pr` skill. For user-visible changes it orchestrates `use-vishot` and the matching runtime variant, then uploads before/after screenshots as GitHub user assets in the PR body. - When asked to create, open, publish, or prepare a pull request, always use the repo-local `create-pr` skill. For user-visible changes it orchestrates `use-vishot` and the matching runtime variant, then uploads before/after screenshots as GitHub user assets in the PR body.
- After you create a pull request, get its review threads, comments, and check status. If a review identifies a confirmed error, fix it, run focused checks, push the update, reply with evidence, and resolve the thread.
- Rebase pulls; branch naming `username/feat/short-name`; clear commit messages (gitmoji is prohibited). - Rebase pulls; branch naming `username/feat/short-name`; clear commit messages (gitmoji is prohibited).
- Summarize changes, how tested (commands), and follow-ups. - Summarize changes, how tested (commands), and follow-ups.
- Improve legacy you touch; avoid one-off patterns. - Improve legacy you touch; avoid one-off patterns.
+10
View File
@@ -9,3 +9,13 @@
- Use `safeParse` if the caller branches on valid and invalid data. - Use `safeParse` if the caller branches on valid and invalid data.
- Do not use `typeof`, `Record<string, unknown>`, or type casts as runtime input validation. - Do not use `typeof`, `Record<string, unknown>`, or type casts as runtime input validation.
- Infer TypeScript types from Valibot schemas. Do not duplicate the contract in an interface. - Infer TypeScript types from Valibot schemas. Do not duplicate the contract in an interface.
## Architecture decisions
- Put server ADRs in `../airi-docs/adr/`.
- Create or update the ADR before you change a server boundary.
- A server boundary includes an HTTP contract, provider contract, persistence model, or cross-module lifecycle.
- Add a module dependency graph, an affected-file tree, and a sequence diagram to every implementation ADR.
- State the decision, scope, non-goals, and test plan in the ADR.
- Keep the ADR status `accepted` only after the decision is confirmed.
- Update the ADR when implementation changes the accepted design.
@@ -0,0 +1,22 @@
import { describe, expect, it } from 'vitest'
import { buildSafeErrorResponseHeaders } from './response'
describe('buildSafeErrorResponseHeaders', () => {
it('keeps the provider error content type and retry delay only', () => {
const headers = buildSafeErrorResponseHeaders(new Response('provider error', {
status: 429,
headers: {
'content-type': 'application/problem+json',
'retry-after': '30',
'set-cookie': 'provider-session=secret',
'x-request-id': 'provider-request-id',
},
}))
expect(headers.get('content-type')).toBe('application/problem+json')
expect(headers.get('retry-after')).toBe('30')
expect(headers.get('set-cookie')).toBeNull()
expect(headers.get('x-request-id')).toBeNull()
})
})
@@ -5,6 +5,11 @@ const SAFE_RESPONSE_HEADERS = new Set([
'cache-control', 'cache-control',
]) ])
const SAFE_ERROR_RESPONSE_HEADERS = new Set([
'content-type',
'retry-after',
])
export function buildSafeResponseHeaders(response: Response): Headers { export function buildSafeResponseHeaders(response: Response): Headers {
const headers = new Headers() const headers = new Headers()
response.headers.forEach((value, key) => { response.headers.forEach((value, key) => {
@@ -13,3 +18,18 @@ export function buildSafeResponseHeaders(response: Response): Headers {
}) })
return headers return headers
} }
/**
* Builds the headers for a final upstream error response.
*
* Upstream errors can expose useful body types and retry delays. Other
* upstream headers can contain credentials, cookies, or provider internals.
*/
export function buildSafeErrorResponseHeaders(response: Response): Headers {
const headers = new Headers()
response.headers.forEach((value, key) => {
if (SAFE_ERROR_RESPONSE_HEADERS.has(key.toLowerCase()))
headers.set(key, value)
})
return headers
}
@@ -9,7 +9,7 @@ import { useLogger } from '@guiiai/logg'
import { extractUsageFromBody } from '../../../../../services/domain/billing/billing' import { extractUsageFromBody } from '../../../../../services/domain/billing/billing'
import { createBadRequestError } from '../../../../../utils/error' import { createBadRequestError } from '../../../../../utils/error'
import { nanoid } from '../../../../../utils/id' import { nanoid } from '../../../../../utils/id'
import { buildSafeResponseHeaders } from '../../http/response' import { buildSafeErrorResponseHeaders, buildSafeResponseHeaders } from '../../http/response'
import { createOpenAiRouteBilling } from '../../middlewares/billing' import { createOpenAiRouteBilling } from '../../middlewares/billing'
import { createRouteTelemetry, newRouteContext } from '../../middlewares/telemetry' import { createRouteTelemetry, newRouteContext } from '../../middlewares/telemetry'
@@ -143,7 +143,7 @@ export function chatCompletions(deps: V1RouteDeps): GatewayCallback<'chat.comple
return new Response(response.body, { return new Response(response.body, {
status: response.status, status: response.status,
headers: buildSafeResponseHeaders(response), headers: buildSafeErrorResponseHeaders(response),
}) })
} }
@@ -259,7 +259,8 @@ async function routeChatAliasCandidates(input: {
routeCtx: ReturnType<typeof newRouteContext> routeCtx: ReturnType<typeof newRouteContext>
}> { }> {
let lastError: unknown let lastError: unknown
for (const modelId of input.modelIds) { for (let index = 0; index < input.modelIds.length; index += 1) {
const modelId = input.modelIds[index]
const routeCtx = newRouteContext() const routeCtx = newRouteContext()
try { try {
const response = await input.deps.llmRouter.route({ const response = await input.deps.llmRouter.route({
@@ -268,7 +269,13 @@ async function routeChatAliasCandidates(input: {
headers: {}, headers: {},
abortSignal: input.abortSignal, abortSignal: input.abortSignal,
}, routeCtx) }, routeCtx)
return { modelId, response, routeCtx } if (response.ok || index === input.modelIds.length - 1)
return { modelId, response, routeCtx }
// The alias owns the next configured model candidate. Its non-2xx body
// cannot reach the client while a later candidate can still serve the
// request, so release it before the next route attempt.
await response.body?.cancel().catch(() => {})
} }
catch (err) { catch (err) {
if (input.abortSignal?.aborted) if (input.abortSignal?.aborted)
@@ -3,6 +3,7 @@ import { Buffer } from 'node:buffer'
import { describe, expect, it, vi } from 'vitest' import { describe, expect, it, vi } from 'vitest'
import { dashscopeCosyvoiceAdapter } from './dashscope-cosyvoice' import { dashscopeCosyvoiceAdapter } from './dashscope-cosyvoice'
import { TtsUpstreamResponseError } from './types'
const UNSPEECH = 'http://unspeech.local:5933' const UNSPEECH = 'http://unspeech.local:5933'
const SPEECH_URL = `${UNSPEECH}/v1/audio/speech` const SPEECH_URL = `${UNSPEECH}/v1/audio/speech`
@@ -52,11 +53,12 @@ describe('dashscopeCosyvoiceAdapter', () => {
expect(Array.from(out)).toEqual(Array.from(audioBytes)) expect(Array.from(out)).toEqual(Array.from(audioBytes))
}) })
it('throws Error with .status when unspeech returns non-2xx (router walks to next key)', async () => { it('throws TtsUpstreamResponseError when unspeech returns non-2xx', async () => {
const fetchImpl = vi.fn().mockResolvedValueOnce(new Response('bad key', { status: 401 })) const fetchImpl = vi.fn().mockResolvedValueOnce(new Response('bad key', { status: 401 }))
await expect( let caught: unknown
dashscopeCosyvoiceAdapter.send( try {
await dashscopeCosyvoiceAdapter.send(
{ text: 'hi', voice: 'longxiaochun_v2' }, { text: 'hi', voice: 'longxiaochun_v2' },
{ {
keyPlaintext: Buffer.from('sk-test', 'utf8'), keyPlaintext: Buffer.from('sk-test', 'utf8'),
@@ -65,8 +67,16 @@ describe('dashscopeCosyvoiceAdapter', () => {
adapterParams: {}, adapterParams: {},
fetchImpl: fetchImpl as unknown as typeof fetch, fetchImpl: fetchImpl as unknown as typeof fetch,
}, },
), )
).rejects.toMatchObject({ status: 401, message: expect.stringContaining('401') }) }
catch (error) {
caught = error
}
expect(caught).toBeInstanceOf(TtsUpstreamResponseError)
if (!(caught instanceof TtsUpstreamResponseError))
throw caught
expect(caught.response.status).toBe(401)
expect(fetchImpl).toHaveBeenCalledTimes(1) expect(fetchImpl).toHaveBeenCalledTimes(1)
}) })
@@ -4,6 +4,7 @@ import { describe, expect, it, vi } from 'vitest'
import { ApiError } from '../../../utils/error' import { ApiError } from '../../../utils/error'
import { getAdapter } from './index' import { getAdapter } from './index'
import { TtsUpstreamResponseError } from './types'
describe('getAdapter', () => { describe('getAdapter', () => {
it('returns the azure adapter by id', () => { it('returns the azure adapter by id', () => {
@@ -280,20 +281,31 @@ describe('azureAdapter.send', () => {
expect(fetchImpl).not.toHaveBeenCalled() expect(fetchImpl).not.toHaveBeenCalled()
}) })
it('throws Error with .status when unspeech non-2xx', async () => { it('throws TtsUpstreamResponseError when unspeech returns non-2xx', async () => {
const adapter = getAdapter('azure') const adapter = getAdapter('azure')
const fetchImpl = vi.fn(async () => new Response('upstream rejected', { status: 401 })) as unknown as typeof fetch const fetchImpl = vi.fn(async () => new Response('upstream rejected', { status: 401 })) as unknown as typeof fetch
await expect(adapter.send( let caught: unknown
{ text: 'hi', voice: 'en-US-AvaMultilingualNeural' }, try {
{ await adapter.send(
keyPlaintext: Buffer.from('k', 'utf8'), { text: 'hi', voice: 'en-US-AvaMultilingualNeural' },
baseURL: 'https://eastasia.tts.speech.microsoft.com/cognitiveservices/v1', {
unspeechBaseURL: 'http://unspeech.local:5933', keyPlaintext: Buffer.from('k', 'utf8'),
adapterParams: { region: 'eastasia' }, baseURL: 'https://eastasia.tts.speech.microsoft.com/cognitiveservices/v1',
fetchImpl, unspeechBaseURL: 'http://unspeech.local:5933',
}, adapterParams: { region: 'eastasia' },
)).rejects.toMatchObject({ status: 401 }) fetchImpl,
},
)
}
catch (error) {
caught = error
}
expect(caught).toBeInstanceOf(TtsUpstreamResponseError)
if (!(caught instanceof TtsUpstreamResponseError))
throw caught
expect(caught.response.status).toBe(401)
}) })
}) })
@@ -459,20 +471,31 @@ describe('stepfunAdapter', () => {
expect(body.extra_body.voice_label).toEqual({ emotion: '高兴' }) expect(body.extra_body.voice_label).toEqual({ emotion: '高兴' })
}) })
it('throws Error with .status when unspeech returns non-2xx', async () => { it('throws TtsUpstreamResponseError when unspeech returns non-2xx', async () => {
const adapter = getAdapter('stepfun') const adapter = getAdapter('stepfun')
const fetchImpl = vi.fn(async () => new Response('bad key', { status: 401 })) as unknown as typeof fetch const fetchImpl = vi.fn(async () => new Response('bad key', { status: 401 })) as unknown as typeof fetch
await expect(adapter.send( let caught: unknown
{ text: 'hi', voice: 'cixingnansheng' }, try {
{ await adapter.send(
keyPlaintext: Buffer.from('bad-key', 'utf8'), { text: 'hi', voice: 'cixingnansheng' },
baseURL: 'https://api.stepfun.com', {
unspeechBaseURL: 'http://unspeech.local', keyPlaintext: Buffer.from('bad-key', 'utf8'),
adapterParams: { model: 'stepaudio-2.5-tts' }, baseURL: 'https://api.stepfun.com',
fetchImpl, unspeechBaseURL: 'http://unspeech.local',
}, adapterParams: { model: 'stepaudio-2.5-tts' },
)).rejects.toMatchObject({ status: 401 }) fetchImpl,
},
)
}
catch (error) {
caught = error
}
expect(caught).toBeInstanceOf(TtsUpstreamResponseError)
if (!(caught instanceof TtsUpstreamResponseError))
throw caught
expect(caught.response.status).toBe(401)
}) })
it('preserves an unspeech request abort for router timeout classification', async () => { it('preserves an unspeech request abort for router timeout classification', async () => {
@@ -70,6 +70,19 @@ export interface TtsResult {
body: ArrayBuffer | ReadableStream<Uint8Array> body: ArrayBuffer | ReadableStream<Uint8Array>
} }
/**
* A non-2xx response from the immediate TTS upstream.
*
* The router uses this response for configured fallback. It returns the final
* response to the client when every permitted fallback has failed.
*/
export class TtsUpstreamResponseError extends Error {
constructor(readonly response: Response) {
super(`TTS upstream responded with ${response.status}`)
this.name = 'TtsUpstreamResponseError'
}
}
/** /**
* Stable provider identifier for the v1 adapter registry. * Stable provider identifier for the v1 adapter registry.
* *
@@ -120,9 +133,8 @@ export interface TtsVoiceCatalogContext {
* *
* Returns: * Returns:
* - A {@link TtsResult} on 2xx upstream responses. * - A {@link TtsResult} on 2xx upstream responses.
* - Throws (Error subclass) on upstream non-2xx the router maps the error to * - Throws {@link TtsUpstreamResponseError} on upstream non-2xx. The router
* the next fallback key/upstream or to a 5xx for the caller. Adapters MUST * can use the response for a fallback or return it to the caller.
* NOT swallow upstream failures.
*/ */
export interface TtsAdapter { export interface TtsAdapter {
/** Stable id used by the registry and config (`tts.upstreams[i].adapter`). */ /** Stable id used by the registry and config (`tts.upstreams[i].adapter`). */
@@ -6,6 +6,7 @@ import { errorMessageFrom } from '@moeru/std'
import { generateSpeechResponse, listVoices, UnSpeechAPIError } from 'unspeech' import { generateSpeechResponse, listVoices, UnSpeechAPIError } from 'unspeech'
import { createBadGatewayError, createInternalError } from '../../../utils/error' import { createBadGatewayError, createInternalError } from '../../../utils/error'
import { TtsUpstreamResponseError } from './types'
interface SendSpeechOptions { interface SendSpeechOptions {
ctx: TtsAdapterContext ctx: TtsAdapterContext
@@ -72,9 +73,10 @@ export async function sendSpeechViaUnSpeech(options: SendSpeechOptions): Promise
throw error throw error
if (error instanceof UnSpeechAPIError) { if (error instanceof UnSpeechAPIError) {
const err = new Error(`${providerLabel} tts upstream ${error.status}: ${error.responseBody.slice(0, 256)}`) as Error & { status?: number } throw new TtsUpstreamResponseError(new Response(error.responseBody, {
err.status = error.status status: error.status,
throw err headers: error.responseHeaders,
}))
} }
throw createInternalError(`${providerLabel} tts fetch failed: ${errorMessageFrom(error) ?? 'unknown'}`) throw createInternalError(`${providerLabel} tts fetch failed: ${errorMessageFrom(error) ?? 'unknown'}`)
@@ -8,6 +8,7 @@ import type { EnvelopeCrypto } from '../../../utils/envelope-crypto'
import type { ConfigKVService } from '../../adapters/config-kv' import type { ConfigKVService } from '../../adapters/config-kv'
import type { TtsAdapterId, TtsInput } from '../../adapters/tts/types' import type { TtsAdapterId, TtsInput } from '../../adapters/tts/types'
import type { ConcurrencyLedger } from './concurrency-ledger' import type { ConcurrencyLedger } from './concurrency-ledger'
import type { UpstreamAttempt } from './error-mapping'
import type { LlmRouteContext, LlmRouteRequest, LlmRoutingGroup, LlmUpstream, RouteFailureTriggers, TtsRoutingGroup, TtsUpstream } from './types' import type { LlmRouteContext, LlmRouteRequest, LlmRoutingGroup, LlmUpstream, RouteFailureTriggers, TtsRoutingGroup, TtsUpstream } from './types'
import { Buffer as NodeBuffer } from 'node:buffer' import { Buffer as NodeBuffer } from 'node:buffer'
@@ -24,12 +25,29 @@ import {
AIRI_ATTR_GEN_AI_GATEWAY_UPSTREAM_URL, AIRI_ATTR_GEN_AI_GATEWAY_UPSTREAM_URL,
} from '../../../utils/observability' } from '../../../utils/observability'
import { getAdapter } from '../../adapters/tts' import { getAdapter } from '../../adapters/tts'
import { TtsUpstreamResponseError } from '../../adapters/tts/types'
import { createConfigLoader } from './config-loader' import { createConfigLoader } from './config-loader'
import { mapUpstreamError } from './error-mapping' import { mapUpstreamError } from './error-mapping'
import { createKeyRotator } from './key-rotator' import { createKeyRotator } from './key-rotator'
const UPSTREAM_BODY_SNIPPET_MAX = 256 const UPSTREAM_BODY_SNIPPET_MAX = 256
interface HttpAttemptFailure {
keyId: string
status: number | 'timeout'
bodySnippet?: string
errorMessage?: string
response?: Response
}
function toDiagnosticAttempt({ provider, keyId, status, bodySnippet, errorMessage }: HttpAttemptFailure & { provider: string }): UpstreamAttempt {
return { provider, keyId, status, bodySnippet, errorMessage }
}
async function discardUpstreamResponse(response: Response | undefined): Promise<void> {
await response?.body?.cancel().catch(() => {})
}
/** /**
* Read at most `maxBytes` from an upstream non-2xx response body for * Read at most `maxBytes` from an upstream non-2xx response body for
* diagnostic logging, then cancel the rest so the socket can return to * diagnostic logging, then cancel the rest so the socket can return to
@@ -214,8 +232,8 @@ function ttsVoicesCacheKey(provider: string, modelName: string): string {
* Returns: * Returns:
* - `route(req)` picks an upstream + key, fetches the upstream, walks * - `route(req)` picks an upstream + key, fetches the upstream, walks
* fallback on non-2xx until one succeeds or every (upstream, key) has * fallback on non-2xx until one succeeds or every (upstream, key) has
* been tried. Returns a `Response` on the first 2xx; throws `ApiError` * been tried. Returns a `Response` on the first 2xx or terminal upstream
* per KTD-1 mapping on full exhaustion. * HTTP error. Throws `ApiError` when every attempt fails before a response.
* *
* The router does NOT open its own OTel span the route handler in U4 * The router does NOT open its own OTel span the route handler in U4
* owns the span. The router only enriches the *active* span with * owns the span. The router only enriches the *active* span with
@@ -243,15 +261,15 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
req: LlmRouteRequest, req: LlmRouteRequest,
perAttemptTimeoutMs: number, perAttemptTimeoutMs: number,
fallbackHttpCodes: number[], fallbackHttpCodes: number[],
onAttemptFailure: (failure: { keyId: string, status: number | 'timeout', bodySnippet?: string, errorMessage?: string }) => void, onAttemptFailure: (failure: HttpAttemptFailure) => void,
): Promise< ): Promise<
| { kind: 'ok', response: Response, attemptIndex: number, upstreamModel: string } | { kind: 'ok', response: Response, attemptIndex: number, upstreamModel: string }
| { kind: 'exhausted', failures: Array<{ keyId: string, status: number | 'timeout', bodySnippet?: string, errorMessage?: string }> } | { kind: 'exhausted', failures: HttpAttemptFailure[] }
> { > {
const provider = deriveProviderTag(upstream.baseURL) const provider = deriveProviderTag(upstream.baseURL)
const rotator = createKeyRotator(upstream, options.envelopeCrypto, req.modelName, options.gatewayMetrics, provider) const rotator = createKeyRotator(upstream, options.envelopeCrypto, req.modelName, options.gatewayMetrics, provider)
const failures: Array<{ keyId: string, status: number | 'timeout', bodySnippet?: string, errorMessage?: string }> = [] const failures: HttpAttemptFailure[] = []
let attemptIndex = 0 let attemptIndex = 0
for (const key of rotator) { for (const key of rotator) {
@@ -299,6 +317,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
} }
if (response.ok) { if (response.ok) {
await Promise.all(failures.map(failure => discardUpstreamResponse(failure.response)))
// First 2xx wins. Enrich the active span and return. // First 2xx wins. Enrich the active span and return.
trace.getActiveSpan()?.setAttributes({ trace.getActiveSpan()?.setAttributes({
[AIRI_ATTR_GEN_AI_GATEWAY_UPSTREAM_URL]: upstream.baseURL, [AIRI_ATTR_GEN_AI_GATEWAY_UPSTREAM_URL]: upstream.baseURL,
@@ -311,17 +330,15 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
const status = response.status const status = response.status
// NOTICE: // NOTICE:
// Drain at most UPSTREAM_BODY_SNIPPET_MAX bytes of the failed body // Read a cloned response for diagnostic logging. The original response
// for diagnostic logging (operators need to see the upstream's real // stays unread because it can become the final client response. When a
// error, not just the status code), then cancel the rest so the // later fallback wins, this router cancels the discarded response.
// socket can return to the pool. Without the cancel, a 401/429/5xx
// fallback storm leaves half-read bodies in flight and exhausts the
// connection pool exactly when the upstream is sick.
// Source: codex review 2026-05-15 HIGH #2 (cancel) + cause-propagation // Source: codex review 2026-05-15 HIGH #2 (cancel) + cause-propagation
// follow-up 2026-05-16 (snippet). // follow-up 2026-05-16 (snippet).
const bodySnippet = await readUpstreamBodySnippet(response) const bodySnippet = await readUpstreamBodySnippet(response.clone())
failures.push({ keyId: key.id, status, bodySnippet }) const failure = { keyId: key.id, status, bodySnippet, response }
onAttemptFailure({ keyId: key.id, status, bodySnippet }) failures.push(failure)
onAttemptFailure(failure)
options.gatewayMetrics?.fallbackCount.add(1, { options.gatewayMetrics?.fallbackCount.add(1, {
provider, provider,
from_key: key.id, from_key: key.id,
@@ -345,6 +362,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
// timeout. The router does NOT fall back on caller-abort: there is // timeout. The router does NOT fall back on caller-abort: there is
// no longer a client waiting for a response. // no longer a client waiting for a response.
if (req.abortSignal?.aborted) { if (req.abortSignal?.aborted) {
await Promise.all(failures.map(failure => discardUpstreamResponse(failure.response)))
logger.withError(err).withFields({ keyId: key.id }).debug('Caller aborted upstream fetch; propagating without fallback') logger.withError(err).withFields({ keyId: key.id }).debug('Caller aborted upstream fetch; propagating without fallback')
throw err throw err
} }
@@ -355,8 +373,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
// ApiError.cause so operators can tell apart "DNS failed" from // ApiError.cause so operators can tell apart "DNS failed" from
// "attempt-timeout" without re-running the request. // "attempt-timeout" without re-running the request.
const errorMessage = errorMessageFromUnknown(err) const errorMessage = errorMessageFromUnknown(err)
failures.push({ keyId: key.id, status: 'timeout', errorMessage }) const failure = { keyId: key.id, status: 'timeout' as const, errorMessage }
onAttemptFailure({ keyId: key.id, status: 'timeout', errorMessage }) failures.push(failure)
onAttemptFailure(failure)
options.gatewayMetrics?.fallbackCount.add(1, { options.gatewayMetrics?.fallbackCount.add(1, {
provider, provider,
from_key: key.id, from_key: key.id,
@@ -372,6 +391,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
attemptIndex += 1 attemptIndex += 1
} }
await Promise.all(failures.slice(0, -1).map(failure => discardUpstreamResponse(failure.response)))
return { kind: 'exhausted', failures } return { kind: 'exhausted', failures }
} }
@@ -392,8 +412,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
const defaults = slice.defaults ?? { perAttemptTimeoutMs: 30000, fullChainTimeoutMs: 60000, fallbackHttpCodes: [401, 402, 403, 429, 500, 502, 503, 504] } const defaults = slice.defaults ?? { perAttemptTimeoutMs: 30000, fullChainTimeoutMs: 60000, fallbackHttpCodes: [401, 402, 403, 429, 500, 502, 503, 504] }
const fallbackHttpCodes = llmModel.fallbackTriggers?.httpCodes ?? defaults.fallbackHttpCodes ?? [401, 402, 403, 429, 500, 502, 503, 504] const fallbackHttpCodes = llmModel.fallbackTriggers?.httpCodes ?? defaults.fallbackHttpCodes ?? [401, 402, 403, 429, 500, 502, 503, 504]
const allFailures: Array<{ provider: string, keyId: string, status: number | 'timeout', bodySnippet?: string, errorMessage?: string }> = [] const allFailures: Array<HttpAttemptFailure & { provider: string }> = []
let triedUpstreams = 0 let triedUpstreams = 0
let terminalResponse: Response | undefined
async function attemptUpstream(upstream: LlmUpstream, index: number) { async function attemptUpstream(upstream: LlmUpstream, index: number) {
const provider = deriveProviderTag(upstream.baseURL) const provider = deriveProviderTag(upstream.baseURL)
@@ -424,12 +445,13 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
return { return {
kind: 'exhausted' as const, kind: 'exhausted' as const,
statuses: result.failures.map(failure => failure.status), statuses: result.failures.map(failure => failure.status),
response: result.failures.at(-1)?.response,
} }
} }
async function routeGroup(group: LlmRoutingGroup): Promise< async function routeGroup(group: LlmRoutingGroup): Promise<
| { kind: 'ok', response: Response } | { kind: 'ok', response: Response }
| { kind: 'exhausted', statuses: Array<number | 'timeout'>, transitionBlocked: boolean } | { kind: 'exhausted', statuses: Array<number | 'timeout'>, transitionBlocked: boolean, response?: Response }
> { > {
const statuses: Array<number | 'timeout'> = [] const statuses: Array<number | 'timeout'> = []
for (let groupCandidateIndex = 0; groupCandidateIndex < group.upstreamIds.length; groupCandidateIndex += 1) { for (let groupCandidateIndex = 0; groupCandidateIndex < group.upstreamIds.length; groupCandidateIndex += 1) {
@@ -447,10 +469,12 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
statuses.push(...result.statuses) statuses.push(...result.statuses)
const hasNextCandidate = groupCandidateIndex < group.upstreamIds.length - 1 const hasNextCandidate = groupCandidateIndex < group.upstreamIds.length - 1
if (hasNextCandidate && !failuresMatch(result.statuses, group.retryOn)) if (hasNextCandidate && !failuresMatch(result.statuses, group.retryOn))
return { kind: 'exhausted', statuses, transitionBlocked: true } return { kind: 'exhausted', statuses, transitionBlocked: true, response: result.response }
if (hasNextCandidate)
await discardUpstreamResponse(result.response)
} }
return { kind: 'exhausted', statuses, transitionBlocked: false } return { kind: 'exhausted', statuses, transitionBlocked: false, response: allFailures.at(-1)?.response }
} }
if (llmModel.routing != null) { if (llmModel.routing != null) {
@@ -466,8 +490,11 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|| !hasNextGroup || !hasNextGroup
|| !failuresMatch(result.statuses, group.continueOn) || !failuresMatch(result.statuses, group.continueOn)
) { ) {
if (result.response != null)
terminalResponse = result.response
break break
} }
await discardUpstreamResponse(result.response)
} }
} }
else { else {
@@ -475,6 +502,10 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
const result = await attemptUpstream(llmModel.upstreams[index], index) const result = await attemptUpstream(llmModel.upstreams[index], index)
if (result.kind === 'ok') if (result.kind === 'ok')
return result.response return result.response
if (index < llmModel.upstreams.length - 1)
await discardUpstreamResponse(result.response)
else if (result.response != null)
terminalResponse = result.response
} }
} }
@@ -512,6 +543,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
} }
} }
if (terminalResponse != null)
return terminalResponse
throw mapUpstreamError( throw mapUpstreamError(
lastFailure.status, lastFailure.status,
{ {
@@ -519,15 +553,15 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
triedUpstreams, triedUpstreams,
lastStatusCode: lastFailure.status, lastStatusCode: lastFailure.status,
}, },
allFailures, allFailures.map(toDiagnosticAttempt),
) )
} }
/** /**
* Run one TTS upstream's key list in order, parallel to {@link dispatchOneUpstream} * Run one TTS upstream's key list in order, parallel to {@link dispatchOneUpstream}
* but delegating actual HTTP to the provider adapter. Adapters surface * but delegating actual HTTP to the provider adapter. Adapters surface
* upstream non-2xx as `Error & { status: number }`; network failures / * upstream non-2xx as {@link TtsUpstreamResponseError}; network failures
* timeouts arrive as plain `Error` with no status. * and timeouts arrive as errors without an upstream response.
*/ */
async function dispatchOneTtsUpstream( async function dispatchOneTtsUpstream(
upstream: TtsUpstream, upstream: TtsUpstream,
@@ -539,15 +573,15 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
perAttemptTimeoutMs: number, perAttemptTimeoutMs: number,
fallbackHttpCodes: number[], fallbackHttpCodes: number[],
unspeechBaseURL: string, unspeechBaseURL: string,
onAttemptFailure: (failure: { keyId: string, status: number | 'timeout', errorMessage?: string }) => void, onAttemptFailure: (failure: HttpAttemptFailure) => void,
): Promise< ): Promise<
| { kind: 'ok', contentType: string, body: ArrayBuffer | ReadableStream<Uint8Array>, attemptIndex: number } | { kind: 'ok', contentType: string, body: ArrayBuffer | ReadableStream<Uint8Array>, attemptIndex: number }
| { kind: 'exhausted', failures: Array<{ keyId: string, status: number | 'timeout', errorMessage?: string }> } | { kind: 'exhausted', failures: HttpAttemptFailure[] }
> { > {
const providerTag = deriveProviderTag(upstream.baseURL) const providerTag = deriveProviderTag(upstream.baseURL)
const rotator = createKeyRotator(upstream, options.envelopeCrypto, modelName, options.gatewayMetrics, providerTag) const rotator = createKeyRotator(upstream, options.envelopeCrypto, modelName, options.gatewayMetrics, providerTag)
const adapter = getAdapter(providerId) const adapter = getAdapter(providerId)
const failures: Array<{ keyId: string, status: number | 'timeout', errorMessage?: string }> = [] const failures: HttpAttemptFailure[] = []
let attemptIndex = 0 let attemptIndex = 0
for (const key of rotator) { for (const key of rotator) {
@@ -587,14 +621,38 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
[AIRI_ATTR_GEN_AI_GATEWAY_KEY_ID]: key.id, [AIRI_ATTR_GEN_AI_GATEWAY_KEY_ID]: key.id,
[AIRI_ATTR_GEN_AI_GATEWAY_FALLBACK_DEPTH]: attemptIndex, [AIRI_ATTR_GEN_AI_GATEWAY_FALLBACK_DEPTH]: attemptIndex,
}) })
await Promise.all(failures.map(failure => discardUpstreamResponse(failure.response)))
return { kind: 'ok', contentType: result.contentType, body: result.body, attemptIndex } return { kind: 'ok', contentType: result.contentType, body: result.body, attemptIndex }
} }
catch (err) { catch (err) {
if (abortSignal?.aborted) { if (abortSignal?.aborted) {
await Promise.all(failures.map(failure => discardUpstreamResponse(failure.response)))
logger.withError(err).withFields({ keyId: key.id }).debug('Caller aborted upstream tts fetch; propagating without fallback') logger.withError(err).withFields({ keyId: key.id }).debug('Caller aborted upstream tts fetch; propagating without fallback')
throw err throw err
} }
if (err instanceof TtsUpstreamResponseError) {
const failure = { keyId: key.id, status: err.response.status, response: err.response }
failures.push(failure)
onAttemptFailure(failure)
options.gatewayMetrics?.fallbackCount.add(1, {
provider: providerTag,
from_key: key.id,
reason: String(failure.status),
})
options.gatewayMetrics?.upstreamErrors.add(1, {
provider: providerTag,
status_code: failure.status,
})
if (!fallbackHttpCodes.includes(failure.status)) {
attemptIndex += 1
break
}
logger.withError(err).withFields({ keyId: key.id, upstream: upstream.baseURL }).warn('Upstream TTS attempt failed')
attemptIndex += 1
continue
}
// Adapter contract (see `server/apps/api/src/services/tts-adapters/types.ts` // Adapter contract (see `server/apps/api/src/services/tts-adapters/types.ts`
// and the three impls): // and the three impls):
// //
@@ -613,34 +671,20 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
if (err instanceof ApiError && err.statusCode < 500) if (err instanceof ApiError && err.statusCode < 500)
throw err throw err
const rawStatus const rawStatus = err instanceof ApiError ? err.statusCode : undefined
= (err as { status?: unknown }).status const failureStatus: number | 'timeout' = rawStatus ?? 'timeout'
?? (err instanceof ApiError ? err.statusCode : undefined)
const failureStatus: number | 'timeout' = typeof rawStatus === 'number' ? rawStatus : 'timeout'
// TTS adapters bake the upstream body snippet into err.message
// (azure: `azure tts upstream 403: <body>`, cosyvoice / volcengine
// analogous), so a single errorMessage carries both the status
// and the upstream payload diagnostics.
const errorMessage = errorMessageFromUnknown(err) const errorMessage = errorMessageFromUnknown(err)
failures.push({ keyId: key.id, status: failureStatus, errorMessage }) const failure = { keyId: key.id, status: failureStatus, errorMessage }
onAttemptFailure({ keyId: key.id, status: failureStatus, errorMessage }) failures.push(failure)
onAttemptFailure(failure)
options.gatewayMetrics?.fallbackCount.add(1, { options.gatewayMetrics?.fallbackCount.add(1, {
provider: providerTag, provider: providerTag,
from_key: key.id, from_key: key.id,
reason: String(failureStatus), reason: String(failureStatus),
}) })
if (typeof rawStatus === 'number') { if (typeof rawStatus === 'number' && !fallbackHttpCodes.includes(rawStatus)) {
options.gatewayMetrics?.upstreamErrors.add(1, { attemptIndex += 1
provider: providerTag, break
status_code: rawStatus,
})
if (!fallbackHttpCodes.includes(rawStatus)) {
// Same key-level policy as chat: stop rotating credentials in this
// candidate. The enclosing group separately decides whether another
// candidate may be tried.
attemptIndex += 1
break
}
} }
logger.withError(err).withFields({ keyId: key.id, upstream: upstream.baseURL }).warn('Upstream TTS attempt failed') logger.withError(err).withFields({ keyId: key.id, upstream: upstream.baseURL }).warn('Upstream TTS attempt failed')
} }
@@ -651,6 +695,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
attemptIndex += 1 attemptIndex += 1
} }
await Promise.all(failures.slice(0, -1).map(failure => discardUpstreamResponse(failure.response)))
return { kind: 'exhausted', failures } return { kind: 'exhausted', failures }
} }
@@ -672,13 +717,13 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
modelName: string, modelName: string,
attemptUpstream: (upstream: TtsUpstream, index: number) => Promise< attemptUpstream: (upstream: TtsUpstream, index: number) => Promise<
| { kind: 'ok', response: Response } | { kind: 'ok', response: Response }
| { kind: 'exhausted', sawTooManyRequests: boolean, statuses: Array<number | 'timeout'> } | { kind: 'exhausted', sawTooManyRequests: boolean, statuses: Array<number | 'timeout'>, response?: Response }
>, >,
retryOn?: RouteFailureTriggers, retryOn?: RouteFailureTriggers,
strategy: 'least-inflight' | 'ordered' = 'least-inflight', strategy: 'least-inflight' | 'ordered' = 'least-inflight',
): Promise< ): Promise<
| { kind: 'ok', response: Response } | { kind: 'ok', response: Response }
| { kind: 'exhausted', statuses: Array<number | 'timeout'>, transitionBlocked: boolean } | { kind: 'exhausted', statuses: Array<number | 'timeout'>, transitionBlocked: boolean, response?: Response }
> { > {
async function markSaturated(upstream: TtsUpstream, poolId: string): Promise<void> { async function markSaturated(upstream: TtsUpstream, poolId: string): Promise<void> {
await ledger.markSaturated(poolId, ttsPoolSaturationTtlSeconds) await ledger.markSaturated(poolId, ttsPoolSaturationTtlSeconds)
@@ -719,6 +764,13 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
let dispatchedAny = false let dispatchedAny = false
let attemptedPools = 0 let attemptedPools = 0
const statuses: Array<number | 'timeout'> = [] const statuses: Array<number | 'timeout'> = []
let lastResponse: Response | undefined
async function discardPreviousPoolResponse(): Promise<void> {
await discardUpstreamResponse(lastResponse)
lastResponse = undefined
}
for (let rankedIndex = 0; rankedIndex < ranked.length; rankedIndex += 1) { for (let rankedIndex = 0; rankedIndex < ranked.length; rankedIndex += 1) {
const { upstream, index, poolId, maxConcurrency } = ranked[rankedIndex] const { upstream, index, poolId, maxConcurrency } = ranked[rankedIndex]
const hasNextCandidate = rankedIndex < ranked.length - 1 const hasNextCandidate = rankedIndex < ranked.length - 1
@@ -726,14 +778,16 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
// Unlimited pool — dispatch without occupying a slot. // Unlimited pool — dispatch without occupying a slot.
dispatchedAny = true dispatchedAny = true
attemptedPools += 1 attemptedPools += 1
await discardPreviousPoolResponse()
const result = await attemptUpstream(upstream, index) const result = await attemptUpstream(upstream, index)
if (result.kind === 'ok') if (result.kind === 'ok')
return result return result
statuses.push(...result.statuses) statuses.push(...result.statuses)
lastResponse = result.response
if (result.sawTooManyRequests) if (result.sawTooManyRequests)
await markSaturated(upstream, poolId) await markSaturated(upstream, poolId)
if (hasNextCandidate && retryOn != null && !failuresMatch(result.statuses, retryOn)) if (hasNextCandidate && retryOn != null && !failuresMatch(result.statuses, retryOn))
return { kind: 'exhausted', statuses, transitionBlocked: true } return { kind: 'exhausted', statuses, transitionBlocked: true, response: result.response }
continue continue
} }
@@ -750,14 +804,16 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
dispatchedAny = true dispatchedAny = true
attemptedPools += 1 attemptedPools += 1
try { try {
await discardPreviousPoolResponse()
const result = await attemptUpstream(upstream, index) const result = await attemptUpstream(upstream, index)
if (result.kind === 'ok') if (result.kind === 'ok')
return result return result
statuses.push(...result.statuses) statuses.push(...result.statuses)
lastResponse = result.response
if (result.sawTooManyRequests) if (result.sawTooManyRequests)
await markSaturated(upstream, poolId) await markSaturated(upstream, poolId)
if (hasNextCandidate && retryOn != null && !failuresMatch(result.statuses, retryOn)) if (hasNextCandidate && retryOn != null && !failuresMatch(result.statuses, retryOn))
return { kind: 'exhausted', statuses, transitionBlocked: true } return { kind: 'exhausted', statuses, transitionBlocked: true, response: result.response }
} }
finally { finally {
await ledger.release(poolId) await ledger.release(poolId)
@@ -780,6 +836,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
kind: 'exhausted', kind: 'exhausted',
statuses, statuses,
transitionBlocked: attemptedPools !== upstreams.length, transitionBlocked: attemptedPools !== upstreams.length,
response: lastResponse,
} }
} }
@@ -804,8 +861,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
const unspeechBaseURL = (await options.configKV.getOrThrow('UNSPEECH_UPSTREAM')).restBaseURL const unspeechBaseURL = (await options.configKV.getOrThrow('UNSPEECH_UPSTREAM')).restBaseURL
const allFailures: Array<{ provider: string, keyId: string, status: number | 'timeout', errorMessage?: string }> = [] const allFailures: Array<HttpAttemptFailure & { provider: string }> = []
let triedUpstreams = 0 let triedUpstreams = 0
let terminalResponse: Response | undefined
// tts upstream schema has no per-upstream timeoutMs (see ttsUpstreamSchema); // tts upstream schema has no per-upstream timeoutMs (see ttsUpstreamSchema);
// the defaults bucket alone governs per-attempt timeout. // the defaults bucket alone governs per-attempt timeout.
@@ -817,7 +875,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
// so the caller can circuit-break thatpool. // so the caller can circuit-break thatpool.
async function attemptUpstream(upstream: TtsUpstream, index: number): Promise< async function attemptUpstream(upstream: TtsUpstream, index: number): Promise<
| { kind: 'ok', response: Response } | { kind: 'ok', response: Response }
| { kind: 'exhausted', sawTooManyRequests: boolean, statuses: Array<number | 'timeout'> } | { kind: 'exhausted', sawTooManyRequests: boolean, statuses: Array<number | 'timeout'>, response?: Response }
> { > {
const providerTag = deriveProviderTag(upstream.baseURL) const providerTag = deriveProviderTag(upstream.baseURL)
triedUpstreams += 1 triedUpstreams += 1
@@ -849,12 +907,13 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
kind: 'exhausted', kind: 'exhausted',
sawTooManyRequests: result.failures.some(f => f.status === 429), sawTooManyRequests: result.failures.some(f => f.status === 429),
statuses: result.failures.map(failure => failure.status), statuses: result.failures.map(failure => failure.status),
response: result.failures.at(-1)?.response,
} }
} }
async function routeGroup(group: TtsRoutingGroup): Promise< async function routeGroup(group: TtsRoutingGroup): Promise<
| { kind: 'ok', response: Response } | { kind: 'ok', response: Response }
| { kind: 'exhausted', statuses: Array<number | 'timeout'>, transitionBlocked: boolean } | { kind: 'exhausted', statuses: Array<number | 'timeout'>, transitionBlocked: boolean, response?: Response }
> { > {
const indexedUpstreams = group.upstreamIds.map((upstreamId) => { const indexedUpstreams = group.upstreamIds.map((upstreamId) => {
const index = ttsModel.upstreams.findIndex(upstream => upstream.id === upstreamId) const index = ttsModel.upstreams.findIndex(upstream => upstream.id === upstreamId)
@@ -892,10 +951,12 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
statuses.push(...result.statuses) statuses.push(...result.statuses)
const hasNextCandidate = groupCandidateIndex < indexedUpstreams.length - 1 const hasNextCandidate = groupCandidateIndex < indexedUpstreams.length - 1
if (hasNextCandidate && !failuresMatch(result.statuses, group.retryOn)) if (hasNextCandidate && !failuresMatch(result.statuses, group.retryOn))
return { kind: 'exhausted', statuses, transitionBlocked: true } return { kind: 'exhausted', statuses, transitionBlocked: true, response: result.response }
if (hasNextCandidate)
await discardUpstreamResponse(result.response)
} }
return { kind: 'exhausted', statuses, transitionBlocked: false } return { kind: 'exhausted', statuses, transitionBlocked: false, response: allFailures.at(-1)?.response }
} }
if (ttsModel.routing != null) { if (ttsModel.routing != null) {
@@ -911,8 +972,11 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|| !hasNextGroup || !hasNextGroup
|| !failuresMatch(result.statuses, group.continueOn) || !failuresMatch(result.statuses, group.continueOn)
) { ) {
if (result.response != null)
terminalResponse = result.response
break break
} }
await discardUpstreamResponse(result.response)
} }
} }
else { else {
@@ -925,12 +989,18 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
const result = await attemptUpstream(ttsModel.upstreams[i], i) const result = await attemptUpstream(ttsModel.upstreams[i], i)
if (result.kind === 'ok') if (result.kind === 'ok')
return result.response return result.response
if (i === ttsModel.upstreams.length - 1 && result.response != null)
terminalResponse = result.response
else
await discardUpstreamResponse(result.response)
} }
} }
else { else {
const result = await routeTtsAcrossPools(ttsModel.upstreams, req.modelName, attemptUpstream) const result = await routeTtsAcrossPools(ttsModel.upstreams, req.modelName, attemptUpstream)
if (result.kind === 'ok') if (result.kind === 'ok')
return result.response return result.response
if (result.response != null)
terminalResponse = result.response
} }
} }
@@ -957,6 +1027,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
} }
} }
if (terminalResponse != null)
return terminalResponse
throw mapUpstreamError( throw mapUpstreamError(
lastFailure.status, lastFailure.status,
{ {
@@ -964,7 +1037,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
triedUpstreams, triedUpstreams,
lastStatusCode: lastFailure.status, lastStatusCode: lastFailure.status,
}, },
allFailures, allFailures.map(toDiagnosticAttempt),
) )
} }
@@ -281,10 +281,12 @@ describe('createLlmRouterService', () => {
expect(ctx.upstreamModel).toBe('real/upstream-id') expect(ctx.upstreamModel).toBe('real/upstream-id')
}) })
it('multi-key fallback: k1=401 then k2=200 → returns 200 and records fallbackCount once', async () => { // https://github.com/moeru-ai/airi/pull/2333#discussion_r3820516102
it('pr #2333: multi-key fallback releases the failed chat response before returning success', async () => {
const { config, crypto } = makeConfig({ upstreams: [{ baseURL: 'https://up-a.example/v1', keyIds: ['k1', 'k2'] }] }) const { config, crypto } = makeConfig({ upstreams: [{ baseURL: 'https://up-a.example/v1', keyIds: ['k1', 'k2'] }] })
const failedResponse = failResponse(401)
const fetchImpl = vi.fn() const fetchImpl = vi.fn()
.mockResolvedValueOnce(failResponse(401)) .mockResolvedValueOnce(failedResponse)
.mockResolvedValueOnce(happyResponse({ ok: 1 })) .mockResolvedValueOnce(happyResponse({ ok: 1 }))
const metrics = makeMetrics() const metrics = makeMetrics()
@@ -299,6 +301,7 @@ describe('createLlmRouterService', () => {
const res = await router.route({ modelName: 'openai/gpt-5-mini', body: {} }) const res = await router.route({ modelName: 'openai/gpt-5-mini', body: {} })
expect(res.status).toBe(200) expect(res.status).toBe(200)
expect(failedResponse.bodyUsed).toBe(true)
expect(fetchImpl.mock.calls.length).toBe(2) expect(fetchImpl.mock.calls.length).toBe(2)
expect((metrics.fallbackCount.add as ReturnType<typeof vi.fn>).mock.calls.length).toBe(1) expect((metrics.fallbackCount.add as ReturnType<typeof vi.fn>).mock.calls.length).toBe(1)
@@ -310,6 +313,36 @@ describe('createLlmRouterService', () => {
expect((metrics.keyExhaustedCount.add as ReturnType<typeof vi.fn>).mock.calls.length).toBe(0) expect((metrics.keyExhaustedCount.add as ReturnType<typeof vi.fn>).mock.calls.length).toBe(0)
}) })
// https://github.com/moeru-ai/airi/pull/2333#discussion_r3820516102
it('pr #2333: caller abort releases earlier failed chat responses', async () => {
const { config, crypto } = makeConfig({ upstreams: [{ baseURL: 'https://up-a.example/v1', keyIds: ['k1', 'k2'] }] })
const failedResponse = failResponse(401)
const callerAbort = new AbortController()
const fetchImpl = vi.fn()
.mockResolvedValueOnce(failedResponse)
.mockImplementationOnce(async () => {
callerAbort.abort(new Error('client disconnected'))
throw callerAbort.signal.reason
})
const router = createLlmRouterService({
configKV: makeConfigKV(config),
envelopeCrypto: crypto,
gatewayMetrics: null,
fetchImpl,
redis: makeRedisStub(),
concurrencyLedger: makeLedger(),
})
await expect(router.route({
modelName: 'openai/gpt-5-mini',
body: {},
abortSignal: callerAbort.signal,
})).rejects.toThrow('client disconnected')
expect(failedResponse.bodyUsed).toBe(true)
})
it('cross-upstream fallback: upstream A keys all 401, upstream B[0] = 200 → returns 200 without terminal exhaustion', async () => { it('cross-upstream fallback: upstream A keys all 401, upstream B[0] = 200 → returns 200 without terminal exhaustion', async () => {
const { config, crypto } = makeConfig({ const { config, crypto } = makeConfig({
upstreams: [ upstreams: [
@@ -340,14 +373,20 @@ describe('createLlmRouterService', () => {
expect((metrics.fallbackCount.add as ReturnType<typeof vi.fn>).mock.calls.length).toBe(2) expect((metrics.fallbackCount.add as ReturnType<typeof vi.fn>).mock.calls.length).toBe(2)
}) })
it('full exhaustion: every upstream + every key 401 → throws 502 BAD_GATEWAY (KTD-1 last-cause = 401 → 502)', async () => { it('full exhaustion: every upstream + every key 401 → returns the final upstream response', async () => {
const { config, crypto } = makeConfig({ const { config, crypto } = makeConfig({
upstreams: [ upstreams: [
{ baseURL: 'https://up-a.example/v1', keyIds: ['kA1'] }, { baseURL: 'https://up-a.example/v1', keyIds: ['kA1'] },
{ baseURL: 'https://up-b.example/v1', keyIds: ['kB1'] }, { baseURL: 'https://up-b.example/v1', keyIds: ['kB1'] },
], ],
}) })
const fetchImpl = vi.fn(async () => failResponse(401)) const fetchImpl = vi.fn(async () => new Response('provider denied', {
status: 401,
headers: {
'content-type': 'text/plain',
'retry-after': '30',
},
}))
const metrics = makeMetrics() const metrics = makeMetrics()
const router = createLlmRouterService({ const router = createLlmRouterService({
@@ -359,16 +398,11 @@ describe('createLlmRouterService', () => {
concurrencyLedger: makeLedger(), concurrencyLedger: makeLedger(),
}) })
try { const response = await router.route({ modelName: 'openai/gpt-5-mini', body: {} })
await router.route({ modelName: 'openai/gpt-5-mini', body: {} }) expect(response.status).toBe(401)
throw new Error('expected throw') expect(response.headers.get('content-type')).toBe('text/plain')
} expect(response.headers.get('retry-after')).toBe('30')
catch (err) { await expect(response.text()).resolves.toBe('provider denied')
expect(err).toBeInstanceOf(ApiError)
expect((err as ApiError).statusCode).toBe(502)
expect((err as ApiError).errorCode).toBe('BAD_GATEWAY')
expect((err as ApiError).details).toMatchObject({ triedKeys: 2, triedUpstreams: 2, lastStatusCode: 401 })
}
const exhaustionCalls = (metrics.keyExhaustedCount.add as ReturnType<typeof vi.fn>).mock.calls const exhaustionCalls = (metrics.keyExhaustedCount.add as ReturnType<typeof vi.fn>).mock.calls
expect(exhaustionCalls.length).toBe(1) expect(exhaustionCalls.length).toBe(1)
@@ -431,6 +465,10 @@ describe('createLlmRouterService', () => {
expect(first).toMatchObject({ keyId: 'kA1', status: 401 }) expect(first).toMatchObject({ keyId: 'kA1', status: 401 })
expect(first.bodySnippet).toEqual(expect.stringContaining('key disabled')) expect(first.bodySnippet).toEqual(expect.stringContaining('key disabled'))
expect(first.errorMessage).toBeUndefined() expect(first.errorMessage).toBeUndefined()
// https://github.com/moeru-ai/airi/pull/2333#discussion_r3828016906
// `Response` owns headers and a body stream. The diagnostic cause must
// keep only the documented serializable attempt fields.
expect(first.response).toBeUndefined()
const second = (cause!.attempts as Array<Record<string, unknown>>)[1] const second = (cause!.attempts as Array<Record<string, unknown>>)[1]
expect(second).toMatchObject({ keyId: 'kB1', status: 'timeout' }) expect(second).toMatchObject({ keyId: 'kB1', status: 'timeout' })
@@ -439,7 +477,7 @@ describe('createLlmRouterService', () => {
} }
}) })
it('same-status exhaustion: all keys 429 → throws 503 + sameStatusExhaustion incremented per provider', async () => { it('same-status exhaustion: all keys 429 → returns 429 and increments sameStatusExhaustion per provider', async () => {
const { config, crypto } = makeConfig({ const { config, crypto } = makeConfig({
upstreams: [ upstreams: [
{ baseURL: 'https://up-a.example/v1', keyIds: ['kA1', 'kA2'] }, { baseURL: 'https://up-a.example/v1', keyIds: ['kA1', 'kA2'] },
@@ -458,7 +496,8 @@ describe('createLlmRouterService', () => {
concurrencyLedger: makeLedger(), concurrencyLedger: makeLedger(),
}) })
await expect(router.route({ modelName: 'openai/gpt-5-mini', body: {} })).rejects.toMatchObject({ statusCode: 503, errorCode: 'SERVICE_UNAVAILABLE' }) const response = await router.route({ modelName: 'openai/gpt-5-mini', body: {} })
expect(response.status).toBe(429)
const calls = (metrics.sameStatusExhaustion.add as ReturnType<typeof vi.fn>).mock.calls const calls = (metrics.sameStatusExhaustion.add as ReturnType<typeof vi.fn>).mock.calls
expect(calls.length).toBe(2) expect(calls.length).toBe(2)
@@ -780,10 +819,11 @@ describe('createLlmRouterService', () => {
const router = makeGroupedLlmRouter(fetchImpl) const router = makeGroupedLlmRouter(fetchImpl)
await expect(router.route({ const response = await router.route({
modelName: 'openai/gpt-5-mini', modelName: 'openai/gpt-5-mini',
body: { messages: [] }, body: { messages: [] },
})).rejects.toBeInstanceOf(ApiError) })
expect(response.status).toBe(429)
expect(calledURLs).toEqual([ expect(calledURLs).toEqual([
'https://api.stepfun.com/step_plan/v1/chat/completions', 'https://api.stepfun.com/step_plan/v1/chat/completions',
@@ -801,10 +841,11 @@ describe('createLlmRouterService', () => {
const router = makeGroupedLlmRouter(fetchImpl) const router = makeGroupedLlmRouter(fetchImpl)
await expect(router.route({ const response = await router.route({
modelName: 'openai/gpt-5-mini', modelName: 'openai/gpt-5-mini',
body: { messages: [] }, body: { messages: [] },
})).rejects.toBeInstanceOf(ApiError) })
expect(response.status).toBe(401)
expect(calledURLs).toEqual([ expect(calledURLs).toEqual([
'https://api.stepfun.com/step_plan/v1/chat/completions', 'https://api.stepfun.com/step_plan/v1/chat/completions',
@@ -823,17 +864,19 @@ describe('createLlmRouterService', () => {
// walked every key + upstream before surfacing — wasting upstream quota // walked every key + upstream before surfacing — wasting upstream quota
// and hiding the actual user-facing 400 behind a 502 mapping. // and hiding the actual user-facing 400 behind a 502 mapping.
// //
// After patch: ApiError 4xx propagates immediately; ApiError 5xx folds // After patch: ApiError 4xx propagates immediately. ApiError 5xx uses
// into the network-failure fallback path using `statusCode`; `Error & // `statusCode` and obeys `fallbackHttpCodes`. TtsUpstreamResponseError
// { status }` stays on the existing fallback policy. // stays on the existing fallback policy.
describe('routeTts adapter error handling', () => { describe('routeTts adapter error handling', () => {
function makeTtsConfig(opts: { function makeTtsConfig(opts: {
provider?: 'azure' provider?: 'azure'
upstreams?: Array<{ baseURL: string, keyIds: string[], adapterParams?: Record<string, unknown> }> upstreams?: Array<{ baseURL: string, keyIds: string[], adapterParams?: Record<string, unknown> }>
fallbackHttpCodes?: number[]
}): { config: RouterConfig, crypto: ReturnType<typeof createEnvelopeCrypto> } { }): { config: RouterConfig, crypto: ReturnType<typeof createEnvelopeCrypto> } {
const crypto = createEnvelopeCrypto({ masterKey: freshMasterKey() }) const crypto = createEnvelopeCrypto({ masterKey: freshMasterKey() })
const modelName = 'tts-test' const modelName = 'tts-test'
const upstreams = opts.upstreams ?? [{ baseURL: 'https://up-a.example', keyIds: ['kA1'] }] const upstreams = opts.upstreams ?? [{ baseURL: 'https://up-a.example', keyIds: ['kA1'] }]
const fallbackHttpCodes = opts.fallbackHttpCodes ?? [401, 429, 500, 502, 503, 504]
const upstreamConfigs = upstreams.map(u => ({ const upstreamConfigs = upstreams.map(u => ({
baseURL: u.baseURL, baseURL: u.baseURL,
keys: u.keyIds.map((id) => { keys: u.keyIds.map((id) => {
@@ -850,14 +893,14 @@ describe('createLlmRouterService', () => {
[modelName]: { [modelName]: {
provider: opts.provider ?? 'azure', provider: opts.provider ?? 'azure',
upstreams: upstreamConfigs, upstreams: upstreamConfigs,
fallbackTriggers: { httpCodes: [401, 429, 500, 502, 503, 504], onTimeout: true }, fallbackTriggers: { httpCodes: fallbackHttpCodes, onTimeout: true },
}, },
}, },
}, },
defaults: { defaults: {
perAttemptTimeoutMs: 5000, perAttemptTimeoutMs: 5000,
fullChainTimeoutMs: 10000, fullChainTimeoutMs: 10000,
fallbackHttpCodes: [401, 429, 500, 502, 503, 504], fallbackHttpCodes,
}, },
} as RouterConfig } as RouterConfig
return { config, crypto } return { config, crypto }
@@ -936,17 +979,52 @@ describe('createLlmRouterService', () => {
expect((metrics.fallbackCount.add as ReturnType<typeof vi.fn>).mock.calls.length).toBe(1) expect((metrics.fallbackCount.add as ReturnType<typeof vi.fn>).mock.calls.length).toBe(1)
}) })
it('upstream `Error & { status: 401 }` folds into the existing fallback path', async () => { // https://github.com/moeru-ai/airi/pull/2333#discussion_r3820516115
// azure adapter throws `Error & { status: number }` on upstream non-2xx it('pr #2333: apiError 5xx stops key fallback when fallbackHttpCodes excludes the status', async () => {
// (see azure.ts:189-194). 401 is in fallbackHttpCodes so we must try const { config, crypto } = makeTtsConfig({
// the next key. upstreams: [{ baseURL: 'https://az.example', keyIds: ['kA1', 'kA2'], adapterParams: { region: 'eastasia' } }],
fallbackHttpCodes: [401, 429],
})
const fetchImpl = vi.fn(async () => {
throw new TypeError('network unreachable')
})
const router = createLlmRouterService({
configKV: makeConfigKV(config),
envelopeCrypto: crypto,
gatewayMetrics: null,
fetchImpl,
redis: makeRedisStub(),
concurrencyLedger: makeLedger(),
})
let caught: unknown
try {
await router.routeTts({
modelName: 'tts-test',
input: { text: 'hi', voice: 'en-US-AvaMultilingualNeural' },
})
}
catch (error) {
caught = error
}
expect(caught).toBeInstanceOf(ApiError)
expect((caught as ApiError).statusCode).toBe(502)
expect(fetchImpl).toHaveBeenCalledTimes(1)
})
it('upstream 401 folds into the existing fallback path', async () => {
// The Azure adapter throws TtsUpstreamResponseError on upstream non-2xx.
// 401 is in fallbackHttpCodes so we must try the next key.
const { config, crypto } = makeTtsConfig({ upstreams: [{ baseURL: 'https://az.example', keyIds: ['kA1', 'kA2'], adapterParams: { region: 'eastasia' } }] }) const { config, crypto } = makeTtsConfig({ upstreams: [{ baseURL: 'https://az.example', keyIds: ['kA1', 'kA2'], adapterParams: { region: 'eastasia' } }] })
let callIdx = 0 let callIdx = 0
const failedResponse = failResponse(401)
const fetchImpl = vi.fn(async () => { const fetchImpl = vi.fn(async () => {
callIdx += 1 callIdx += 1
if (callIdx === 1) if (callIdx === 1)
return failResponse(401) return failedResponse
return new Response(new Uint8Array([0x01]), { status: 200, headers: { 'content-type': 'audio/mpeg' } }) return new Response(new Uint8Array([0x01]), { status: 200, headers: { 'content-type': 'audio/mpeg' } })
}) })
const metrics = makeMetrics() const metrics = makeMetrics()
@@ -966,6 +1044,7 @@ describe('createLlmRouterService', () => {
}) })
expect(res.status).toBe(200) expect(res.status).toBe(200)
expect(failedResponse.bodyUsed).toBe(true)
expect(fetchImpl).toHaveBeenCalledTimes(2) expect(fetchImpl).toHaveBeenCalledTimes(2)
const fallbackCalls = (metrics.fallbackCount.add as ReturnType<typeof vi.fn>).mock.calls const fallbackCalls = (metrics.fallbackCount.add as ReturnType<typeof vi.fn>).mock.calls
expect(fallbackCalls.length).toBe(1) expect(fallbackCalls.length).toBe(1)
@@ -995,10 +1074,12 @@ describe('createLlmRouterService', () => {
concurrencyLedger: makeLedger(), concurrencyLedger: makeLedger(),
}) })
await expect(router.routeTts({ const response = await router.routeTts({
modelName: 'tts-test', modelName: 'tts-test',
input: { text: 'hi', voice: 'en-US-AvaMultilingualNeural' }, input: { text: 'hi', voice: 'en-US-AvaMultilingualNeural' },
})).rejects.toMatchObject({ statusCode: 502 }) })
expect(response.status).toBe(451)
await expect(response.json()).resolves.toEqual({ error: 'bad' })
const exhaustionCalls = (metrics.keyExhaustedCount.add as ReturnType<typeof vi.fn>).mock.calls const exhaustionCalls = (metrics.keyExhaustedCount.add as ReturnType<typeof vi.fn>).mock.calls
expect(exhaustionCalls.length).toBe(1) expect(exhaustionCalls.length).toBe(1)
@@ -1323,10 +1404,11 @@ describe('createLlmRouterService', () => {
const router = makeGroupedStepfunRouter(fetchImpl) const router = makeGroupedStepfunRouter(fetchImpl)
await expect(router.routeTts({ const response = await router.routeTts({
modelName: 'stepfun/stepaudio-2.5-tts', modelName: 'stepfun/stepaudio-2.5-tts',
input: { text: '你好' }, input: { text: '你好' },
})).rejects.toBeInstanceOf(ApiError) })
expect(response.status).toBe(429)
expect(calledProfiles).toEqual(['step-plan', 'step-plan']) expect(calledProfiles).toEqual(['step-plan', 'step-plan'])
expect(calledProfiles).not.toContain('default') expect(calledProfiles).not.toContain('default')
@@ -1342,10 +1424,11 @@ describe('createLlmRouterService', () => {
const router = makeGroupedStepfunRouter(fetchImpl) const router = makeGroupedStepfunRouter(fetchImpl)
await expect(router.routeTts({ const response = await router.routeTts({
modelName: 'stepfun/stepaudio-2.5-tts', modelName: 'stepfun/stepaudio-2.5-tts',
input: { text: '你好' }, input: { text: '你好' },
})).rejects.toBeInstanceOf(ApiError) })
expect(response.status).toBe(401)
expect(calledProfiles).toEqual(['step-plan']) expect(calledProfiles).toEqual(['step-plan'])
}) })
@@ -1638,10 +1721,11 @@ describe('createLlmRouterService', () => {
const router = makePoolRouter(config, crypto, ledger, fetchImpl) const router = makePoolRouter(config, crypto, ledger, fetchImpl)
await expect(router.routeTts({ const response = await router.routeTts({
modelName: 'tts-pool', modelName: 'tts-pool',
input: { text: 'hi' }, input: { text: 'hi' },
})).rejects.toBeInstanceOf(ApiError) })
expect(response.status).toBe(402)
expect(selectedAppIds).toEqual(['plan-b']) expect(selectedAppIds).toEqual(['plan-b'])
}) })
@@ -1728,7 +1812,8 @@ describe('createLlmRouterService', () => {
const fetchImpl = vi.fn(async () => failResponse(429)) as unknown as typeof fetch const fetchImpl = vi.fn(async () => failResponse(429)) as unknown as typeof fetch
const router = makePoolRouter(config, crypto, ledger, fetchImpl) const router = makePoolRouter(config, crypto, ledger, fetchImpl)
await expect(router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } })).rejects.toBeInstanceOf(ApiError) const response = await router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } })
expect(response.status).toBe(429)
expect(markSaturated).toHaveBeenCalledWith('app-1', expect.any(Number)) expect(markSaturated).toHaveBeenCalledWith('app-1', expect.any(Number))
}) })
@@ -1743,7 +1828,8 @@ describe('createLlmRouterService', () => {
const fetchImpl = vi.fn(async () => failResponse(500)) as unknown as typeof fetch const fetchImpl = vi.fn(async () => failResponse(500)) as unknown as typeof fetch
const router = makePoolRouter(config, crypto, ledger, fetchImpl) const router = makePoolRouter(config, crypto, ledger, fetchImpl)
await expect(router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } })).rejects.toBeInstanceOf(ApiError) const response = await router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } })
expect(response.status).toBe(500)
expect(markSaturated).not.toHaveBeenCalled() expect(markSaturated).not.toHaveBeenCalled()
}) })
@@ -28,6 +28,11 @@ const SAFE_RESPONSE_HEADERS = new Set([
'cache-control', 'cache-control',
]) ])
const SAFE_ERROR_RESPONSE_HEADERS = new Set([
'content-type',
'retry-after',
])
function asRecord(value: unknown): Record<string, unknown> | undefined { function asRecord(value: unknown): Record<string, unknown> | undefined {
if (typeof value !== 'object' || value == null || Array.isArray(value)) if (typeof value !== 'object' || value == null || Array.isArray(value))
return undefined return undefined
@@ -199,7 +204,7 @@ export function createOpenAiSpeechService(deps: OpenAiSpeechServiceDeps) {
.warn('tts speech delivered with upstream error status') .warn('tts speech delivered with upstream error status')
return new Response(response.body, { return new Response(response.body, {
status: response.status, status: response.status,
headers: buildSafeResponseHeaders(response), headers: buildSafeErrorResponseHeaders(response),
}) })
} }
@@ -383,3 +388,12 @@ function buildSafeResponseHeaders(response: Response): Headers {
}) })
return headers return headers
} }
function buildSafeErrorResponseHeaders(response: Response): Headers {
const headers = new Headers()
response.headers.forEach((value, key) => {
if (SAFE_ERROR_RESPONSE_HEADERS.has(key.toLowerCase()))
headers.set(key, value)
})
return headers
}