From c7c0c85b3e57a976c3d711ff2c3e27467e87188d Mon Sep 17 00:00:00 2001 From: umaru Date: Sat, 8 Aug 2026 09:47:57 +0800 Subject: [PATCH] =?UTF-8?q?refactor(proxy):=20=E5=AE=8C=E6=88=90=20HTTP=20?= =?UTF-8?q?adapter=20cutover=20(#264)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../architecture/cliproxy-integration.md | 2 +- docs/guide/architecture/failover-circuit.md | 27 +- docs/guide/architecture/overview.md | 32 +- docs/guide/architecture/request-lifecycle.md | 13 +- docs/guide/architecture/security.md | 2 +- docs/guide/architecture/upstream-model.md | 68 +-- docs/guide/deployment/https-proxy.md | 2 +- docs/guide/usage/circuit-breaker-config.md | 8 +- docs/guide/usage/client-keys.md | 4 +- docs/guide/usage/invoke-models.md | 12 +- docs/guide/usage/logs-stats.md | 4 +- docs/guide/usage/model-routing.md | 12 +- docs/guide/usage/request-recording.md | 20 +- docs/guide/usage/troubleshooting.md | 40 +- .../v1/[...path]/proxy-request-lifecycle.ts | 453 +++++++----------- .../v1/[...path]/proxy-stream-lifecycle.ts | 190 +++++++- src/app/api/proxy/v1/[...path]/route.ts | 19 +- .../api/proxy/proxy-stream-lifecycle.test.ts | 60 ++- tests/unit/api/proxy/route-adapter.test.ts | 37 ++ tests/unit/api/proxy/route.test.ts | 59 +-- 20 files changed, 600 insertions(+), 464 deletions(-) create mode 100644 tests/unit/api/proxy/route-adapter.test.ts diff --git a/docs/guide/architecture/cliproxy-integration.md b/docs/guide/architecture/cliproxy-integration.md index 49869b64..bf836aec 100644 --- a/docs/guide/architecture/cliproxy-integration.md +++ b/docs/guide/architecture/cliproxy-integration.md @@ -141,7 +141,7 @@ CPA 调整对外约定时，路径后缀与默认路由能力的改动集中在 ## 转发路径中的 CPA 分支 -CPA 上游在请求生命周期里只有一处特殊处理，即单账号映射上游的模型前缀注入，发生在 `src/app/api/proxy/v1/[...path]/route.ts:1519-1532`： +CPA 上游在请求生命周期里只有一处特殊处理，即单账号映射上游的模型前缀注入，发生在 `src/app/api/proxy/v1/[...path]/proxy-execution.ts` 的 `forwardWithFailover` 上游调用前： ```ts let cliproxyModelOverride: string | undefined; diff --git a/docs/guide/architecture/failover-circuit.md b/docs/guide/architecture/failover-circuit.md index dce2e0a2..3eb38fbb 100644 --- a/docs/guide/architecture/failover-circuit.md +++ b/docs/guide/architecture/failover-circuit.md @@ -64,22 +64,11 @@ AutoRouter 把「上游会失败」当作常态。一次客户端请求可能触 ## 单次请求内的故障转移循环 -入口函数 `forwardWithFailover`，源码 `src/app/api/proxy/v1/[...path]/route.ts:1295-1760`。签名： +入口函数 `forwardWithFailover`，源码 `src/app/api/proxy/v1/[...path]/proxy-execution.ts`。签名： ```ts -// route.ts:1295-1320（节选） -async function forwardWithFailover( - request, - routeCapability, - path, - requestId, - candidateUpstreamIds: string[], - requestModel, - affinityContext, - compensationHeaders, - onQueueStateChange?, - config: FailoverConfig = DEFAULT_FAILOVER_CONFIG -); +// proxy-execution.ts +async function forwardWithFailover(input: ProxyExecutionInput): Promise; ``` 默认配置在 `src/lib/services/failover-config.ts:44-48`： @@ -94,7 +83,7 @@ export const DEFAULT_FAILOVER_CONFIG: FailoverConfig = { 主循环每一轮做三件事： -1. 调用 `selectFromUpstreamCandidates(candidateUpstreamIds, failedUpstreamIds, affinityContext)`，把已经失败的上游排除（`route.ts:1371` 维护 `failedUpstreamIds` 数组）； +1. 调用 `selectFromUpstreamCandidates(candidateUpstreamIds, failedUpstreamIds, affinityContext)`，把已经失败的上游排除； 2. 调用 `forwardRequest(...)` 实际转发； 3. 根据结果决定下一步： - 成功 → `markHealthy` + `recordSuccess` + 返回响应 @@ -105,7 +94,7 @@ export const DEFAULT_FAILOVER_CONFIG: FailoverConfig = { 代理层把两类错误判定为可故障转移： -**异常类（`isFailoverableError`，`route.ts:844-869`）**： +**异常类（`isFailoverableError`，`proxy-execution.ts`）**： - `CircuitBreakerOpenError` - `FirstByteTimeoutError` / `StreamIdleTimeoutError` / `UpstreamNoContentStreamError` @@ -117,7 +106,7 @@ export const DEFAULT_FAILOVER_CONFIG: FailoverConfig = { - 状态码非 2xx 且不在 `excludeStatusCodes` 中 -默认 `excludeStatusCodes` 为空数组，意味着**所有 4xx（包括 401 / 403 / 404 / 429）都会触发故障转移**。`getErrorType()` 会区分 `http_429` 和通用 `http_4xx`（`route.ts:829-830`），但并不影响是否触发转移。如果不希望客户端的 401 把所有上游试一遍，需要在 `FailoverConfig.excludeStatusCodes` 里配置 `[401, 403]` 等。 +默认 `excludeStatusCodes` 为空数组，意味着**所有 4xx（包括 401 / 403 / 404 / 429）都会触发故障转移**。`getErrorType()` 会区分 `http_429` 和通用 `http_4xx`（`proxy-execution.ts`），但并不影响是否触发转移。如果不希望客户端的 401 把所有上游试一遍，需要在 `FailoverConfig.excludeStatusCodes` 里配置 `[401, 403]` 等。 ### 失败是否记入熔断器：FailureRule @@ -130,13 +119,13 @@ export const DEFAULT_FAILOVER_CONFIG: FailoverConfig = { | `bodyPattern` | 响应体正则 | | `headerName` + `headerPattern` | 响应头名 + 值正则 | -源码 `src/lib/services/upstream-failure-rules.ts:12-18`。当 `matchFailureRule()` 命中一条规则时，本次失败仍然会触发故障转移，但 `circuitBreakerRecorded = false`（`route.ts:1555-1556, 1714-1715`），不写入 `circuit_breaker_states.failure_count`。 +源码 `src/lib/services/upstream-failure-rules.ts:12-18`。当 `matchFailureRule()` 命中一条规则时，本次失败仍然会触发故障转移，但 `circuitBreakerRecorded = false`（`proxy-execution.ts`），不写入 `circuit_breaker_states.failure_count`。 典型用法：上游对应 OAuth 受控的 CLIProxyAPI auth-file，正常会偶发 401 触发后台 refresh，不希望把上游打到熔断；可以加一条 `statusCodes: [401], bodyPattern: "token expired"` 的规则。上游层 `upstreams.failure_rule_config.useGlobalRules`（默认 `true`）控制是否同时参与全局规则匹配（`upstream-failure-rules.ts:353`）。 ### 并发已满与队列等待 -当 `selectFromUpstreamCandidates` 抛出 `AllCandidatesConcurrencyFullError` 并携带 `waitableCandidate` 时，主循环不会立即返回失败，而是调用 `resumeQueuedUpstreamSelection`（`route.ts:1409-1452`），内部通过 `upstreamQueueAdmission` 等待该上游的并发槽位释放。等待时长由 `upstream.queue_policy` 控制，超时会抛 `UpstreamQueueWaitTimeoutError`，此时不再尝试其他上游，直接返回 503 / 504。 +当 `selectFromUpstreamCandidates` 抛出 `AllCandidatesConcurrencyFullError` 并携带 `waitableCandidate` 时，主循环不会立即返回失败，而是调用 `resumeQueuedUpstreamSelection`（`proxy-execution.ts`），内部通过 `upstreamQueueAdmission` 等待该上游的并发槽位释放。等待时长由 `upstream.queue_policy` 控制，超时会抛 `UpstreamQueueWaitTimeoutError`，此时不再尝试其他上游，直接返回 503 / 504。 ### 故障转移决策日志 diff --git a/docs/guide/architecture/overview.md b/docs/guide/architecture/overview.md index a1efe45b..020fe431 100644 --- a/docs/guide/architecture/overview.md +++ b/docs/guide/architecture/overview.md @@ -45,21 +45,21 @@ AutoRouter 是一个 Next.js 全栈应用：同一个进程同时承担「管理 代码组织遵循 Next.js App Router 的常规分层，运行期逻辑集中在 `src/lib/services/`： -| 路径 | 职责 | -| ----------------------------------------- | --------------------------------------------------------------------------- | -| `src/app/api/proxy/v1/[...path]/route.ts` | 唯一的代理入口，GET/POST/PUT/DELETE/PATCH 都委托给同一个 `handleProxy` 函数 | -| `src/app/api/admin/` | 管理 API：上游、密钥、熔断、日志、统计、计费、流量录制、CLIProxy 等 | -| `src/app/api/health/route.ts` | 公开健康探针，不需要鉴权 | -| `src/app/[locale]/(dashboard)/` | 管理后台页面集合（需要登录） | -| `src/app/[locale]/(auth)/login/` | 登录页（独立布局，不挂 dashboard 框架） | -| `src/lib/services/` | 全部运行期业务逻辑模块 | -| `src/lib/db/` | Drizzle ORM schema 与数据库 client | -| `src/lib/utils/` | 通用工具：配置加载、鉴权 helper、加密、CORS 等 | -| `src/components/` | 管理后台 React 组件（shadcn/ui 基础） | -| `src/hooks/` | TanStack Query 包装的数据获取 hooks | -| `src/i18n/`、`src/messages/` | next-intl 配置与中英文翻译 | - -`src/app/api/proxy/v1/[...path]/route.ts` 在文件末尾把所有 HTTP 方法都导向同一个内部函数（`POST` 位于第 4147 行、`handleProxy` 位于第 2440 行），后文「请求生命周期」会逐步展开它的内部流程。 +| 路径 | 职责 | +| ----------------------------------------- | ------------------------------------------------------------------------------------- | +| `src/app/api/proxy/v1/[...path]/route.ts` | 唯一的代理入口，GET/POST/PUT/DELETE/PATCH 都委托给 `executeProxyRequest` 生命周期入口 | +| `src/app/api/admin/` | 管理 API：上游、密钥、熔断、日志、统计、计费、流量录制、CLIProxy 等 | +| `src/app/api/health/route.ts` | 公开健康探针，不需要鉴权 | +| `src/app/[locale]/(dashboard)/` | 管理后台页面集合（需要登录） | +| `src/app/[locale]/(auth)/login/` | 登录页（独立布局，不挂 dashboard 框架） | +| `src/lib/services/` | 全部运行期业务逻辑模块 | +| `src/lib/db/` | Drizzle ORM schema 与数据库 client | +| `src/lib/utils/` | 通用工具：配置加载、鉴权 helper、加密、CORS 等 | +| `src/components/` | 管理后台 React 组件（shadcn/ui 基础） | +| `src/hooks/` | TanStack Query 包装的数据获取 hooks | +| `src/i18n/`、`src/messages/` | next-intl 配置与中英文翻译 | + +`src/app/api/proxy/v1/[...path]/route.ts` 在文件末尾把所有 HTTP 方法都导向同一个生命周期入口（`executeProxyRequest` 位于 `proxy-request-lifecycle.ts`），后文「请求生命周期」会逐步展开它的内部流程。 ## 服务模块清单 @@ -171,7 +171,7 @@ AutoRouter 是一个 Next.js 全栈应用：同一个进程同时承担「管理 | `/api/health` | 无 | 健康探针 | | `/[locale]/...` 页面 | 浏览器侧 sessionStorage Token | 管理后台 UI | -代理入口的全部 HTTP 方法都委托给 `handleProxy`；管理 API 的每个路由独立鉴权；健康探针完全公开。next-intl 中间件位于 `src/proxy.ts`（注意：是 `src/proxy.ts`，不是 Next.js 默认惯用的 `src/middleware.ts`），其 matcher 显式排除 `/_next`、`/api`、带扩展名的资源路径，因此中间件**不会**拦截任何 API 请求，所有 API 鉴权都发生在 route handler 自身内部。 +代理入口的全部 HTTP 方法都委托给 `executeProxyRequest`；管理 API 的每个路由独立鉴权；健康探针完全公开。next-intl 中间件位于 `src/proxy.ts`（注意：是 `src/proxy.ts`，不是 Next.js 默认惯用的 `src/middleware.ts`），其 matcher 显式排除 `/_next`、`/api`、带扩展名的资源路径，因此中间件**不会**拦截任何 API 请求，所有 API 鉴权都发生在 route handler 自身内部。 ## 国际化与路由分组 diff --git a/docs/guide/architecture/request-lifecycle.md b/docs/guide/architecture/request-lifecycle.md index e058cd00..fe5c3ae9 100644 --- a/docs/guide/architecture/request-lifecycle.md +++ b/docs/guide/architecture/request-lifecycle.md @@ -5,7 +5,7 @@ outline: deep # 请求生命周期 -这一页跟踪一次客户端请求从进入 AutoRouter、完成鉴权与上游准入、发送到上游，再到响应、日志、计费和流量录制落地的完整流程。代理请求现在由三个边界清晰的模块协作：`src/app/api/proxy/v1/[...path]/route.ts` 只负责 HTTP 方法与参数适配，`proxy-request-lifecycle.ts` 的 `handleProxy` 负责生命周期编排，`proxy-execution.ts` 的 `forwardWithFailover` 负责候选选择、队列准入、上游调用、失败转移和资源释放。 +这一页跟踪一次客户端请求从进入 AutoRouter、完成鉴权与上游准入、发送到上游，再到响应、日志、计费和流量录制落地的完整流程。代理请求由多个边界清晰的模块协作：`route.ts` 只负责 HTTP 方法与参数适配，`proxy-request-lifecycle.ts` 的 `executeProxyRequest` 负责生命周期编排，`proxy-execution.ts` 的 `forwardWithFailover` 负责候选选择、队列准入、上游调用与失败转移，`proxy-non-stream-lifecycle.ts` / `proxy-stream-lifecycle.ts` 负责终态响应和日志、计费、录制收口。 示例以最常见的 `POST /api/proxy/v1/chat/completions` 为基准，其他协议（Anthropic `/v1/messages`、Gemini `/v1beta/models/:generateContent`、OpenAI `/v1/responses` 等）的差异在相应阶段标出。 @@ -17,11 +17,12 @@ outline: deep ```ts export async function POST(request: NextRequest, context: RouteContext) { - return handleProxy(request, context); + const { path } = await context.params; + return executeProxyRequest(request, path.join("/")); } ``` -`route.ts` 不再直接编排鉴权、路由、上游调用、日志、计费或 recording。阅读代理行为时，以 `proxy-request-lifecycle.ts` 的 `handleProxy` 为主时序，以 `proxy-execution.ts` 的 `forwardWithFailover` 为上游执行子流程。 +`route.ts` 不再直接编排鉴权、路由、上游调用、日志、计费或 recording。阅读代理行为时，以 `proxy-request-lifecycle.ts` 的 `executeProxyRequest` 为主时序，以 `proxy-execution.ts` 的 `forwardWithFailover` 为上游执行子流程，并在 `proxy-non-stream-lifecycle.ts` 与 `proxy-stream-lifecycle.ts` 查看终态副作用。 ## 阶段二：CORS 与 OPTIONS @@ -35,7 +36,7 @@ export async function POST(request: NextRequest, context: RouteContext) { 2. `x-api-key`：Anthropic SDK 的默认 header。 3. `x-goog-api-key`：Gemini SDK 的默认 header。 -提取后，`handleProxy` 按 key prefix 找候选记录并用 `verifyApiKey` 做 bcrypt 比对，再检查过期与用户状态。 +提取后，`executeProxyRequest` 按 key prefix 找候选记录并用 `verifyApiKey` 做 bcrypt 比对，再检查过期与用户状态。 | 场景 | HTTP 响应 | 说明 | | ------------------------ | ---------------------------------------- | ---------------------------- | @@ -68,7 +69,7 @@ export async function POST(request: NextRequest, context: RouteContext) { ## 阶段五：候选过滤与上游选路 -`handleProxy` 先读取活跃上游快照，再根据 Key 的 `accessMode` 构建候选集合： +`executeProxyRequest` 先读取活跃上游快照，再根据 Key 的 `accessMode` 构建候选集合： - `restricted`：只允许 `apiKeyUpstreams` 关联表中的上游。 - `unrestricted`：允许所有活跃上游，但仍受 capability、model rule、健康和熔断状态限制。 @@ -169,7 +170,7 @@ export async function POST(request: NextRequest, context: RouteContext) { ▼ [2] CORS / OPTIONS（当前没有自定义 preflight handler） ▼ -[3] handleProxy 鉴权 +[3] executeProxyRequest 鉴权 ├ 缺失 / 无效 / 过期 / disabled key → 401 └ 记录拒绝日志，不访问上游 ▼ diff --git a/docs/guide/architecture/security.md b/docs/guide/architecture/security.md index 1478ea2a..c7f58494 100644 --- a/docs/guide/architecture/security.md +++ b/docs/guide/architecture/security.md @@ -83,7 +83,7 @@ const keyValueEncrypted = encrypt(keyValue); // Fernet ### 转发时的验证 -代理路由 `src/app/api/proxy/v1/[...path]/route.ts:2452-2473` 用前缀查候选行，再对候选逐条 bcrypt 比对： +代理请求生命周期 `src/app/api/proxy/v1/[...path]/proxy-request-lifecycle.ts` 的 `executeProxyRequest` 用前缀查候选行，再对候选逐条 bcrypt 比对： ```ts const keyPrefix = getKeyPrefix(keyValue); diff --git a/docs/guide/architecture/upstream-model.md b/docs/guide/architecture/upstream-model.md index 402f1114..5327e7bc 100644 --- a/docs/guide/architecture/upstream-model.md +++ b/docs/guide/architecture/upstream-model.md @@ -39,7 +39,7 @@ outline: deep - `codex_cli_responses` → `openai_responses` - `claude_code_messages` → `anthropic_messages` -降级行为在 `src/app/api/proxy/v1/[...path]/route.ts:2663` 通过双候选池实现：先按 CLI 能力构建主池，再按 fallback 能力构建副池，由 `shouldPreferGenericFallbackPool` 决定使用哪个池。 +降级行为在 `proxy-request-lifecycle.ts` 通过双候选池实现：先按 CLI 能力构建主池，再按 fallback 能力构建副池，由 `shouldPreferGenericFallbackPool` 决定使用哪个池。 ## upstreams 表关键字段 @@ -74,7 +74,7 @@ interface UpstreamModelRule { - `regex`：`new RegExp(rule.value).test(model)` 全字段正则匹配 ::: warning model_redirects 与 model_rules 的 alias **不改写转发 body** -两者解析出的「目标模型名」只用于**过滤候选**、**写日志** 和 **计费价格解析** 三件事，**不会**改写客户端请求 body 里的 `model` 字段。`forwardRequest` 把原始 model 原样发给上游（`src/lib/services/proxy-client.ts:1004, 1116`），唯一会改写 body 的路径是 CLIProxyAPI 上游：当 `selectedUpstream.cliproxyAuthFileName` 存在时，代理层构造 `cliproxyModelOverride` 传给 `forwardRequest`（`route.ts:1519-1530, 1540`），由 `applyModelOverride` 改写 body。 +两者解析出的「目标模型名」只用于**过滤候选**、**写日志**和**计费价格解析**三件事，**不会**改写客户端请求 body 里的 `model` 字段。`forwardRequest` 把原始 model 原样发给上游（`src/lib/services/proxy-client.ts:1004, 1116`），唯一会改写 body 的路径是 CLIProxyAPI 上游：当 `selectedUpstream.cliproxyAuthFileName` 存在时，`proxy-execution.ts` 构造 `cliproxyModelOverride` 传给 `forwardRequest`，由 `applyModelOverride` 改写 body。 这意味着：给一个普通 OpenAI 上游配置 `model_redirects: { "gpt-4o-mini": "gpt-4o" }`，客户端发 `gpt-4o-mini`，候选筛选与日志会按 `gpt-4o` 来，但实际打到上游的 body 里仍是 `gpt-4o-mini`。需要真正的服务端 model 改写时，应当在客户端层面解决，或者走 CLIProxyAPI 集成。 ::: @@ -120,7 +120,7 @@ API Key 的加解密统一通过 `src/lib/utils/encryption.ts` 提供的 `encryp ## 候选池构建：第一阶段（按 RouteCapability + 模型规则） -候选池的构建发生在 `handleProxy`（`src/app/api/proxy/v1/[...path]/route.ts:2440`）内部，按「能力 → API Key 授权 → 模型规则」三层过滤，最终交给 `selectFromUpstreamCandidates`。 +候选池的构建发生在 `executeProxyRequest`（`src/app/api/proxy/v1/[...path]/proxy-request-lifecycle.ts`）内部，按「能力 → API Key 授权 → 模型规则」三层过滤，最终交给 `selectFromUpstreamCandidates`。 ::: tip 关于 routeByModel `src/lib/services/model-router.ts:306` 的 `routeByModel(model)` 实现了一套基于模型名前缀（`claude-` / `gpt-` / `gemini-`）推断 provider type 再过滤候选的算法，但**当前运行期没有任何生产路径调用它**——全仓库 `routeByModel(` 仅匹配定义本身。代理路径采用的是下文描述的 `resolveRouteCapabilityCandidatePool` + `filterCandidatesByModelRules`，按客户端**请求路径**解析出的 `RouteCapability` 与 `model_rules` 进行匹配，与模型名前缀无关。阅读源码时如果落到 `routeByModel` 上，可以视为历史代码。 @@ -128,7 +128,7 @@ API Key 的加解密统一通过 `src/lib/utils/encryption.ts` 提供的 `encryp ### 步骤 1：按 RouteCapability + API Key 授权构建候选池 -`resolveRouteCapabilityCandidatePool`（`route.ts:662`）签名： +`resolveRouteCapabilityCandidatePool`（`proxy-request-lifecycle.ts`）签名： ```ts function resolveRouteCapabilityCandidatePool( @@ -139,9 +139,9 @@ function resolveRouteCapabilityCandidatePool( ): RouteCapabilityCandidatePool; ``` -`activeUpstreams` 是数据库查出的全部 `is_active=true` 上游（`route.ts:2654`）；`allowedUpstreamIdSet` 在 `restricted` 模式下取 API Key 绑定的 `api_key_upstreams` 集合，`unrestricted` 模式下取全集（`route.ts:2657-2661`）。 +`activeUpstreams` 是数据库查出的全部 `is_active=true` 上游；`allowedUpstreamIdSet` 在 `restricted` 模式下取 API Key 绑定的 `api_key_upstreams` 集合，`unrestricted` 模式下取全集。 -过滤逻辑（`route.ts:668-670`）： +过滤逻辑位于 `proxy-request-lifecycle.ts` 的 `resolveRouteCapabilityCandidatePool`： ```ts const capabilityCandidates = activeUpstreams.filter((upstream) => @@ -149,17 +149,17 @@ const capabilityCandidates = activeUpstreams.filter((upstream) => ); ``` -随后再用 `allowedUpstreamIdSet` 做授权过滤（`route.ts:671-673`），得到 `authorizedCapabilityCandidates`，并把这一层结果命名输出在 `RouteCapabilityCandidatePool`（`route.ts:654-660`）： +随后再用 `allowedUpstreamIdSet` 做授权过滤，得到 `authorizedCapabilityCandidates`，并把这一层结果命名输出在 `RouteCapabilityCandidatePool`： - `capabilityCandidates`：能力匹配但不限授权 - `authorizedCapabilityCandidates`：能力匹配 + API Key 授权 - `candidateUpstreamIds`：上一层 ID 列表，是后续函数的实际输入 -主候选池在 `route.ts:2663` 构建。如果客户端命中的是 CLI 窄能力（`codex_cli_responses` / `claude_code_messages`），代理还会在 `route.ts:2669` 用 `getFallbackRouteCapability` 解析出的通用能力构建第二个 fallback 池，由 `shouldPreferGenericFallbackPool` 决定使用哪个。 +主候选池在 `executeProxyRequest` 中构建。如果客户端命中的是 CLI 窄能力（`codex_cli_responses` / `claude_code_messages`），代理还会用 `getFallbackRouteCapability` 解析出的通用能力构建第二个 fallback 池，由 `shouldPreferGenericFallbackPool` 决定使用哪个池。 ### 步骤 2：按 model_rules 过滤候选 -`filterCandidatesByModelRules`（`route.ts:592`）以请求 body 里的 `model` 字段为输入： +`filterCandidatesByModelRules`（`proxy-request-lifecycle.ts`）以请求 body 里的 `model` 字段为输入： ```ts function filterCandidatesByModelRules( @@ -168,7 +168,7 @@ function filterCandidatesByModelRules( ): { allowed: Upstream[]; excluded: RoutingExcluded[] }; ``` -行为（`route.ts:595-622`）： +行为（`proxy-request-lifecycle.ts` 的 `filterCandidatesByModelRules`）： - `originalModel` 为 `null`（请求 body 没有 `model` 字段）→ 全部放行，不过滤 - 否则对每个候选调用 `resolvePathRoutingModelForUpstream(originalModel, candidate)`： @@ -176,11 +176,11 @@ function filterCandidatesByModelRules( - 未命中且上游有显式规则（`hasExplicitRules: true`）→ 加入 `excluded`，理由 `"model_not_allowed"` - 未命中且上游没有任何规则（`hasExplicitRules: false`）→ **仍加入 `allowed`**（视为「不限制」） -这步调用在 `route.ts:2755`，紧跟主候选池构建之后；fallback 池切换时第二次调用在 `route.ts:3068`。 +这步在 `executeProxyRequest` 的主候选池构建之后执行；fallback 池切换时会再次执行。 ### 步骤 3：resolvePathRoutingModelForUpstream 与规则合并 -每个候选上游被 `filterCandidatesByModelRules` 调用时，最终落到 `resolvePathRoutingModelForUpstream`（`route.ts:558`），它内部调用 `matchUpstreamModelRules` 完成实际匹配，返回： +每个候选上游被 `filterCandidatesByModelRules` 调用时，最终落到 `resolvePathRoutingModelForUpstream`（`proxy-request-lifecycle.ts`），它内部调用 `matchUpstreamModelRules` 完成实际匹配，返回： ```ts { @@ -197,7 +197,7 @@ function filterCandidatesByModelRules( ### 步骤 4：resolvedModel 的真实用途 -`resolvePathRoutingModelForUpstream` 返回的 `resolvedModel` 在四处被消费（`route.ts:2853, 3085, 3148, 3903`）： +`resolvePathRoutingModelForUpstream` 返回的 `resolvedModel` 在生命周期编排的多个阶段被消费： 1. 决定 API Key 配额检查时用哪个 model 名（计费维度对齐） 2. 写入 `request_logs` 与 `RoutingDecisionLog.resolved_model` @@ -208,7 +208,7 @@ function filterCandidatesByModelRules( ### 步骤 5：候选 ID 列表交给 load-balancer -走到这里得到 `candidateUpstreamIds`（已通过 capability、API Key 授权、model_rules 三重过滤），由 `handleProxy` 在 `route.ts:3045` / `route.ts:3100`（fallback 路径）传给 `forwardWithFailover`，后者在 `route.ts:1386` 调用 `selectFromUpstreamCandidates` 进入第二阶段。 +走到这里得到 `candidateUpstreamIds`（已通过 capability、API Key 授权、model_rules 三重过滤），由 `executeProxyRequest` 传给 `forwardWithFailover`，后者调用 `selectFromUpstreamCandidates` 进入第二阶段。 ## load-balancer 选上游：第二阶段（按 tier + 加权） @@ -243,7 +243,7 @@ score = 1.0 - min(latencyMs / 500, 0.5) // 至少 0.1 effectiveWeight = upstream.weight * score ``` -最近一次记录的 `latency_ms`（来自 `upstream_health` 表）越大、分越低。但要注意：当前 `markHealthy` 调用点写入的 latency 固定为 `100`（`src/lib/services/health-checker.ts` + `route.ts:1601, 2072`），不是实测值。因此 `score` 在当前实现里基本恒为 1.0，加权采样近似等价于按 `upstream.weight` 加权随机。 +最近一次记录的 `latency_ms`（来自 `upstream_health` 表）越大、分越低。但要注意：当前 `markHealthy` 调用点位于 `proxy-execution.ts`，写入的 latency 固定为 `100`，不是实测值。因此 `score` 在当前实现里基本恒为 1.0，加权采样近似等价于按 `upstream.weight` 加权随机。 加权抽样完成后输出的 `selectedUpstream` 即为本次实际转发目标。 @@ -262,16 +262,16 @@ effectiveWeight = upstream.weight * score | `NoHealthyUpstreamsError` | 所有 tier 全部过滤后仍为空 | | `AllCandidatesConcurrencyFullError` | 候选池存在但全部 `concurrency_full`，可能携带等待句柄 | -错误类定义在 `load-balancer.ts:29, 39, 49`。`AllCandidatesConcurrencyFullError` 携带的 `waitableCandidate` 会被代理入口拿去做队列等待（`route.ts:1409-1469`），等待超时则抛 `UpstreamQueueWaitTimeoutError` 转 504，详见 [失败转移与熔断](./failover-circuit)。 +错误类定义在 `load-balancer.ts`。`AllCandidatesConcurrencyFullError` 携带的 `waitableCandidate` 会被 `proxy-execution.ts` 拿去做队列等待，等待超时则抛 `UpstreamQueueWaitTimeoutError` 转 504，详见 [失败转移与熔断](./failover-circuit)。 ## 健康状态与路由的关系 `upstream_health` 表（`schema-pg.ts:133-152`）记录 `is_healthy`、`latency_ms`、`failure_count`、`error_message` 等。代码里有两处「健康写入」入口： -| 写入函数 | 触发点 | -| ------------------------------------ | ------------------------------------------------------------------------------------------------- | -| `markHealthy(upstreamId, latencyMs)` | 请求成功（`route.ts:1601` 非流式；`route.ts:2072` 流式完成） | -| `markUnhealthy(upstreamId, reason)` | HTTP 非 2xx（`route.ts:1559`）、网络/超时错误（`route.ts:1722`）、流式中途错误（`route.ts:2103`） | +| 写入函数 | 触发点 | +| ------------------------------------ | ------------------------------------------------------------------------------------------ | +| `markHealthy(upstreamId, latencyMs)` | 请求成功（非流式与流式均由 `proxy-execution.ts` 的执行流程触发） | +| `markUnhealthy(upstreamId, reason)` | HTTP 非 2xx、网络 / 超时错误、流式中途错误（均由 `proxy-execution.ts` 的故障转移流程触发） | ::: warning is_healthy 不直接参与路由 `load-balancer.ts:1033` 的 `filterByExclusions` 注释明确写着： @@ -283,19 +283,19 @@ effectiveWeight = upstream.weight * score ## 调用链一览 -| 入口 | 行号 | 作用 | -| ------------------------------------------------------------------------------ | --------- | ---------------------------------- | -| `src/app/api/proxy/v1/[...path]/route.ts` `handleProxy` | 2440 | 代理主流程容器 | -| ↳ `resolveRouteCapability(method, path, headers)` | 2504 | 路径 → RouteCapability | -| ↳ `resolveRouteCapabilityCandidatePool` | 2663 | 按主能力 + API Key 授权构建候选池 | -| ↳ `getFallbackRouteCapability` + 副候选池 | 2669-2678 | CLI 能力降级路径 | -| ↳ `filterCandidatesByModelRules` | 2755 | 按 `model_rules` 过滤候选 | -| ↳ `forwardWithFailover(... candidateUpstreamIds ...)` | 3045 | 故障转移主循环 | -| `src/app/api/proxy/v1/[...path]/route.ts` `resolvePathRoutingModelForUpstream` | 558 | 实际匹配规则、产出 `resolvedModel` | -| `src/lib/services/upstream-model-rules.ts` `normalizeUpstreamModelRules` | 189 | model_rules / 旧字段统一规范化 | -| `src/lib/services/upstream-model-rules.ts` `matchUpstreamModelRules` | 326 | 三种规则类型的实际匹配 | -| `src/lib/services/load-balancer.ts` `selectFromUpstreamCandidates` | 675 | tier 过滤 + 加权抽样 | -| ↳ `performTieredSelection` | 983 | 内部 tier 循环 | -| ↳ `selectWeightedWithHealthScore` | 485 | 加权抽样实现 | +| 入口 | 行号 | 作用 | +| --------------------------------------------------------------------------------- | ---- | ---------------------------------- | +| `src/app/api/proxy/v1/[...path]/proxy-request-lifecycle.ts` `executeProxyRequest` | — | 代理主流程容器 | +| ↳ `resolveRouteCapability(method, path, headers)` | — | 路径 → RouteCapability | +| ↳ `resolveRouteCapabilityCandidatePool` | — | 按主能力 + API Key 授权构建候选池 | +| ↳ `getFallbackRouteCapability` + 副候选池 | — | CLI 能力降级路径 | +| ↳ `filterCandidatesByModelRules` | — | 按 `model_rules` 过滤候选 | +| ↳ `forwardWithFailover(... candidateUpstreamIds ...)` | — | 故障转移主循环 | +| `proxy-request-lifecycle.ts` `resolvePathRoutingModelForUpstream` | — | 实际匹配规则、产出 `resolvedModel` | +| `src/lib/services/upstream-model-rules.ts` `normalizeUpstreamModelRules` | 189 | `model_rules` / 旧字段统一规范化 | +| `src/lib/services/upstream-model-rules.ts` `matchUpstreamModelRules` | 326 | 三种规则类型的实际匹配 | +| `src/lib/services/load-balancer.ts` `selectFromUpstreamCandidates` | 675 | tier 过滤 + 加权抽样 | +| ↳ `performTieredSelection` | 983 | 内部 tier 循环 | +| ↳ `selectWeightedWithHealthScore` | 485 | 加权抽样实现 | 读源码时按这条链顺着走即可。后续上游被选中后的转发、SSE 处理、失败重试由 [请求生命周期](./request-lifecycle) 和 [失败转移与熔断](./failover-circuit) 接力描述。 diff --git a/docs/guide/deployment/https-proxy.md b/docs/guide/deployment/https-proxy.md index c3c14f6b..910f8894 100644 --- a/docs/guide/deployment/https-proxy.md +++ b/docs/guide/deployment/https-proxy.md @@ -259,5 +259,5 @@ handle /api/* { - `docker-compose.yml`：端口映射默认值 `${PORT:-3331}:3000` - `src/lib/utils/config.ts`：`corsOrigins` 解析逻辑，确认当前没有运行期 CORS 注入 -- `src/app/api/proxy/v1/[...path]/route.ts`：`/api/proxy/v1/*` 在 `stream: true` 下走 SSE 路径 +- `src/app/api/proxy/v1/[...path]/route.ts` 与 `proxy-request-lifecycle.ts`：`/api/proxy/v1/*` 的 HTTP 方法由 adapter 统一委托，`stream: true` 由生命周期进入 SSE 路径 - `src/app/api/admin/*` 与 `src/lib/utils/api-auth.ts`：管理 API 用 Bearer Token 鉴权而非 cookie diff --git a/docs/guide/usage/circuit-breaker-config.md b/docs/guide/usage/circuit-breaker-config.md index 7d190d25..7590fd53 100644 --- a/docs/guide/usage/circuit-breaker-config.md +++ b/docs/guide/usage/circuit-breaker-config.md @@ -88,7 +88,7 @@ API 字段（管理 API 层接收以秒为单位的输入并转换为毫秒存 `matchFailureRule`（`upstream-failure-rules.ts:342`）按 `priority` 升序找第一条命中规则，返回 `MatchedFailureRule | null`。返回非 null 时： - **failover 仍发生**：请求会换下一条上游继续重试。 -- **熔断不计数**：`route.ts:1549-1557` 显式判断 `matchedFailureRule === null`，命中规则时跳过 `recordFailure(upstream, errorType)`。 +- **熔断不计数**：`proxy-execution.ts` 显式判断 `matchedFailureRule === null`，命中规则时跳过 `recordFailure(upstream, errorType)`。 也就是说失败规则的语义是「这次失败已经被规则解释了，不再算作上游故障」，而不是「这次失败不算失败」。 @@ -153,11 +153,11 @@ POST body 字段对应 `match` 结构（`upstream-failure-rules.ts:16-22`、`fai 熔断与 failover 共用同一次 HTTP 失败事件，但处于两个独立的代码路径： - **failover**：「换一个上游重试」。触发条件由 `src/lib/services/failover-config.ts:57-73` 决定，默认任何非 2xx 都触发，可通过 `excludeStatusCodes` 排除；策略可选 `exhaust_all`（默认）或 `max_attempts`（默认 10 次，`failover-config.ts:44-48`）。 -- **熔断计数**：「这条上游不健康」。由 `recordFailure` 写入，受 `shouldRecordCircuitBreakerFailure(path)`（`route.ts:800-803`）与 `matchedFailureRule === null`（`route.ts:1549-1557`）两个条件共同控制。 +- **熔断计数**：「这条上游不健康」。由 `recordFailure` 写入，受 `shouldRecordCircuitBreakerFailure(path)`（`proxy-execution.ts`）与 `matchedFailureRule === null`（`proxy-execution.ts`）两个条件共同控制。 -`shouldRecordCircuitBreakerFailure` 维护一个路径白名单 `CIRCUIT_BREAKER_NEUTRAL_PATHS = {"messages/count_tokens"}`（`route.ts:793`）。命中白名单的路径即使失败也不计入熔断（这种 token 计数类请求不代表上游真实健康度）。 +`shouldRecordCircuitBreakerFailure` 维护一个路径白名单 `CIRCUIT_BREAKER_NEUTRAL_PATHS = {"messages/count_tokens"}`（`proxy-execution.ts`）。命中白名单的路径即使失败也不计入熔断（这种 token 计数类请求不代表上游真实健康度）。 -`matchedFailureRule` 在三处出现：HTTP 错误分支（`route.ts:1549-1556`）、流式错误 settlement 分支（`:1708-1712`）、网络 / 超时 settlement 分支（`:1948-1951`）。**例外**：流式 runtime 错误分支（`:1632-1635`）不检查 failure rule，直接按白名单决定。 +`matchedFailureRule` 在 HTTP 错误、流式错误和网络 / 超时 settlement 中参与判断，均位于 `proxy-execution.ts` 的故障转移循环；流式 runtime 错误则由 `proxy-stream-lifecycle.ts` 负责终态收口，并按路径白名单决定是否计入熔断。 ## 排查清单 diff --git a/docs/guide/usage/client-keys.md b/docs/guide/usage/client-keys.md index 2a7e1cbf..913e9313 100644 --- a/docs/guide/usage/client-keys.md +++ b/docs/guide/usage/client-keys.md @@ -54,7 +54,7 @@ outline: deep | `rpm_limit` | null | 每分钟请求数（RPM）上限。正整数；null 表示不限制请求数 | | `tpm_limit` | null | 每分钟 Token 数（TPM）上限。正整数；null 表示不限制已计量的响应 Token 数 | -过期判定（`src/app/api/proxy/v1/[...path]/route.ts:2469`）发生在每次代理请求鉴权时：`expiresAt && expiresAt < new Date()` 即返回 401。无需周期任务介入。 +过期判定（`src/app/api/proxy/v1/[...path]/proxy-request-lifecycle.ts` 的 `executeProxyRequest`）发生在每次代理请求鉴权时：`expiresAt && expiresAt < new Date()` 即返回 401。无需周期任务介入。 `spending_rules` 与上游的 `spending_rules` 含义类似，但作用对象是「该 Key 的累计消费」而非「该上游的累计消费」。 @@ -178,7 +178,7 @@ curl -X POST http://:3331/api/proxy/v1/chat/completions \ }' ``` -AutoRouter 也支持额外两种 header 名称（`src/app/api/proxy/v1/[...path]/route.ts:2255`）： +AutoRouter 也支持额外两种 header 名称（由 `proxy-request-lifecycle.ts` 的 `extractProxyApiKey` 解析）： ``` Authorization: Bearer diff --git a/docs/guide/usage/invoke-models.md b/docs/guide/usage/invoke-models.md index 389be65f..24464da4 100644 --- a/docs/guide/usage/invoke-models.md +++ b/docs/guide/usage/invoke-models.md @@ -26,7 +26,7 @@ Content-Type: application/json ## 鉴权 header 支持的三种形式 -AutoRouter 按以下顺序尝试解析客户端 Key（`src/app/api/proxy/v1/[...path]/route.ts:2255`）： +AutoRouter 按以下顺序尝试解析客户端 Key（`src/app/api/proxy/v1/[...path]/proxy-request-lifecycle.ts` 的 `extractProxyApiKey`）： ``` Authorization: Bearer @@ -58,7 +58,7 @@ AutoRouter 把客户端请求路径解析为「路由能力」，再从声明了 ## 流式与非流式 -OpenAI 协议下用请求体的 `stream` 字段切换（`src/app/api/proxy/v1/[...path]/route.ts:2415`）： +OpenAI 协议下用请求体的 `stream` 字段切换（`proxy-request-lifecycle.ts` 的 `isStreamRequest`）： | `stream` 值 | 行为 | | --------------- | ------------------------------------------------------------------------------------------------------------------------ | @@ -206,16 +206,16 @@ print(response.text) ## 响应行为：透传 + 改写 -正常 2xx 响应：AutoRouter 把上游响应体透传给调用方，响应 header 在 `src/app/api/proxy/v1/[...path]/route.ts:3198` 处由 `new Headers(result.headers)` 拷贝得到。这里的 `result.headers` 并不是上游响应的原始 header，已经经过 `src/lib/services/proxy-client.ts` 的两道处理： +正常 2xx 响应：AutoRouter 把上游响应体透传给调用方，非流式终态由 `proxy-non-stream-lifecycle.ts` 的 `settleNonStreamRequest` 适配，流式终态由 `proxy-stream-lifecycle.ts` 的 `createStreamResponse` 适配。这里的 `result.headers` 并不是上游响应的原始 header，已经经过 `src/lib/services/proxy-client.ts` 的两道处理： 1. **去 hop-by-hop**：`proxy-client.ts:1168-1173` 的 inline 循环按 `HOP_BY_HOP_HEADERS` 集合过滤上游响应头，剔除 `connection`、`keep-alive`、`transfer-encoding` 等不应跨连接传递的字段（与 `filterHeaders` 处理请求侧 inbound header 是两段不同代码，不要混淆）。 2. **去解压元数据**：当 undici 已经自动解压响应体时，`proxy-client.ts:1177-1180` 会同时删除 `content-encoding` 与 `content-length`，避免响应体长度与声明值不一致导致下游再解压时报 `Z_DATA_ERROR`。 -也就是说调用方拿到的不是 1:1 的上游 header 副本。SSE 流式分支额外强制写入 `Content-Type: text/event-stream`、`Cache-Control: no-cache`、`Connection: keep-alive` 三个标准头（`route.ts:3563-3565`）。代理层**不会**追加 `X-AutoRouter-Request-Id` / `X-AutoRouter-Upstream-Id` 之类的自定义头；本次请求的 ID 与命中上游 ID 通过管理后台的「请求日志」回查。响应体本身格式与上游完全一致，调用方不需要任何兼容层。 +也就是说调用方拿到的不是 1:1 的上游 header 副本。`createStreamResponse` 额外强制写入 `Content-Type: text/event-stream`、`Cache-Control: no-cache`、`Connection: keep-alive` 三个标准头。代理层**不会**追加 `X-AutoRouter-Request-Id` / `X-AutoRouter-Upstream-Id` 之类的自定义头；本次请求的 ID 与命中上游 ID 通过管理后台的「请求日志」回查。响应体本身格式与上游完全一致，调用方不需要任何兼容层。 错误响应分两类，调用方需要分别识别： -**鉴权阶段**（`src/app/api/proxy/v1/[...path]/route.ts:2452-2479`）：发生在统一错误包装之前，响应体格式较朴素： +**鉴权阶段**（`proxy-request-lifecycle.ts` 的 `executeProxyRequest`）：发生在统一错误包装之前，响应体格式较朴素： ```json { "error": "Missing API key" } @@ -236,7 +236,7 @@ print(response.text) 状态码与错误码映射关系定义在 `src/lib/services/unified-error.ts`；以上仅列最常见者，完整枚举以源文件 `UnifiedErrorCode` 与 `STATUS_CODE_MAP` 为准。 -failover 只在「首字节前」对调用方无感：上游在返回响应头时如果命中可重试条件（5xx、连接超时等），AutoRouter 会按 [`docs/circuit-breaker.md`](/circuit-breaker) 中的逻辑自动尝试下一条候选，仅当全部候选都失败时才返回最终错误。一旦 SSE 流的第一块数据已经吐出（`result.isStream === true`、`src/app/api/proxy/v1/[...path]/route.ts:1592-1651`），后续的流中断不会再换上游，调用方会看到一条提前结束的 SSE 流，需要自行处理「上游 stream 中断」错误。两类失败都会写入 `requestLogs`，可在 `/api/admin/logs` 看到本次请求的 `failover_history` 字段，记录每次尝试的上游 ID、错误类型与时间戳。 +failover 只在「首字节前」对调用方无感：上游在返回响应头时如果命中可重试条件（5xx、连接超时等），AutoRouter 会按 [`docs/circuit-breaker.md`](/circuit-breaker) 中的逻辑自动尝试下一条候选，仅当全部候选都失败时才返回最终错误。一旦 SSE 的第一块数据已经吐出，`proxy-stream-lifecycle.ts` 的流式收口不会再换上游，调用方会看到一条提前结束的 SSE 流，需要自行处理「上游 stream 中断」错误。两类失败都会写入 `requestLogs`，可在 `/api/admin/logs` 看到本次请求的 `failover_history` 字段，记录每次尝试的上游 ID、错误类型与时间戳。 ## 模型字段的写法约束 diff --git a/docs/guide/usage/logs-stats.md b/docs/guide/usage/logs-stats.md index 9b98edef..63d0a06a 100644 --- a/docs/guide/usage/logs-stats.md +++ b/docs/guide/usage/logs-stats.md @@ -87,7 +87,7 @@ Token 数据由 `extractNormalizedUsage`（`src/lib/services/proxy-client.ts:468 ``` client request ↓ -proxy route 决策完毕（route.ts:2965） +统一代理生命周期完成路由与准入决策（`proxy-request-lifecycle.ts` 的 `executeProxyRequest`） ↓ logRequestStart() — INSERT 一行，status_code=NULL，duration_ms=NULL ↓ @@ -100,7 +100,7 @@ calculateAndPersistRequestBillingSnapshot() — 在 request_billing_snapshots publishRequestLogLiveUpdate() — 广播 SSE 事件给 /api/admin/logs/live 订阅者 ``` -部分非流式入口直接调 `logRequest()`（`request-logger.ts:504-557`）一次性 INSERT，跳过 in-progress 中间态。 +部分非流式终态由 `proxy-non-stream-lifecycle.ts` 的 `settleNonStreamRequest` 统一落日志；鉴权、准入拒绝和流式终态也由同一生命周期模块按对应失败阶段写入。 ### duration_ms 与 routing_duration_ms 的 clamp diff --git a/docs/guide/usage/model-routing.md b/docs/guide/usage/model-routing.md index a642cdfb..cca8a148 100644 --- a/docs/guide/usage/model-routing.md +++ b/docs/guide/usage/model-routing.md @@ -87,7 +87,7 @@ AutoRouter 选择上游的决策依据并非「模型名前缀映射」这类预 ### 规则匹配出口：resolvePathRoutingModelForUpstream -`resolvePathRoutingModelForUpstream(originalModel, upstream)`（`src/app/api/proxy/v1/[...path]/route.ts:558`）是路由层使用的统一出口。内部调用 `matchUpstreamModelRules`（`upstream-model-rules.ts:326`），返回四个字段： +`resolvePathRoutingModelForUpstream(originalModel, upstream)`（`src/app/api/proxy/v1/[...path]/proxy-request-lifecycle.ts`）是路由层使用的统一出口。内部调用 `matchUpstreamModelRules`（`upstream-model-rules.ts:326`），返回四个字段： | 字段 | 含义 | | ------------------ | ------------------------------------------------------------ | @@ -98,10 +98,10 @@ AutoRouter 选择上游的决策依据并非「模型名前缀映射」这类预 ### 「未显式拒绝即默认放行」语义 -整体过滤逻辑在 `filterCandidatesByModelRules`（`route.ts:592-625`）： +整体过滤逻辑在 `filterCandidatesByModelRules`（`proxy-request-lifecycle.ts`）中： ```ts -// 摘自 route.ts:592-625 +// 摘自 proxy-request-lifecycle.ts if (!originalModel) return { allowed: candidates, excluded: [] }; // 模型缺失 → 全部放行 for (const candidate of candidates) { const modelResolution = resolvePathRoutingModelForUpstream(originalModel, candidate); @@ -135,7 +135,7 @@ for (const candidate of candidates) { "gemini-" → "google" ``` -但这个函数**不再被主代理路由 `src/app/api/proxy/v1/[...path]/route.ts` 调用**（全仓 grep 无 `routeByModel` 在主路由中的引用）。它现在只在两处出现： +但这个函数**不再被主代理生命周期 `src/app/api/proxy/v1/[...path]/proxy-request-lifecycle.ts` 调用**（全仓 grep 无 `routeByModel` 在主路由中的引用）。它现在只在两处出现： - `model-router.ts:310` 的旧版 `routeByModel`——已不在主路由路径上。 - `src/lib/services/billing-cost-service.ts:445`——计费时用来区分输入 token 计算口径。 @@ -151,9 +151,9 @@ for (const candidate of candidates) { 客户端 Key 的 `allowed_models` 字段（`schema-pg.ts:55`）是另一层白名单，在候选筛选**之前**生效： -`isModelAllowedByApiKey(requestedModel, allowedModels)`（`src/lib/api-key-models.ts:16`）：`allowedModels` 为空或 null 直接放行；否则做精确字符串 `includes` 检查，命中失败的请求直接返回错误码 `API_KEY_MODEL_NOT_ALLOWED`（`route.ts:2513`）。 +`isModelAllowedByApiKey(requestedModel, allowedModels)`（`src/lib/api-key-models.ts:16`）：`allowedModels` 为空或 null 直接放行；否则做精确字符串 `includes` 检查，命中失败的请求直接返回错误码 `API_KEY_MODEL_NOT_ALLOWED`（`proxy-request-lifecycle.ts`）。 -`getApiKeyVisibleModelList`（`route.ts:627`）仅在 `GET /v1/models` 这种返回模型列表的请求里触发：对 Key 的 `allowedModels` 做过滤，保留其中**能被至少一个候选上游接受**的模型名（用 `resolvePathRoutingModelForUpstream(model, candidate).matched` 判断），返回交集。 +`getApiKeyVisibleModelList`（`proxy-request-lifecycle.ts`）仅在 `GET /v1/models` 这种返回模型列表的请求里触发：对 Key 的 `allowedModels` 做过滤，保留其中**能被至少一个候选上游接受**的模型名（用 `resolvePathRoutingModelForUpstream(model, candidate).matched` 判断），返回交集。 叠加规则三条： diff --git a/docs/guide/usage/request-recording.md b/docs/guide/usage/request-recording.md index fd6dbfe0..36e4a172 100644 --- a/docs/guide/usage/request-recording.md +++ b/docs/guide/usage/request-recording.md @@ -26,7 +26,7 @@ outline: deep shouldRecordTraffic(outcome) === enabled && (mode === "all" || mode === outcome); ``` -每次代理请求单独调一次 `getTrafficRecordingSettings()`（`route.ts:2487`，每请求新查 DB，无 in-memory 缓存），所以**改设置立即生效，不需要重启**。 +每次代理请求由 `proxy-request-lifecycle.ts` 的 `executeProxyRequest` 调用 `getTrafficRecordingSettings()`，每请求新查 DB，无 in-memory 缓存，所以**改设置立即生效，不需要重启**。 入口：管理后台 **系统 → 流量录制**（`/system/traffic-recording`，页面文件 `src/app/[locale]/(dashboard)/system/traffic-recording/page.tsx`）。 @@ -90,19 +90,15 @@ shouldRecordTraffic(outcome) === enabled && (mode === "all" || mode === outcome) ### 入口与执行时机 -`src/app/api/proxy/v1/[...path]/route.ts`： +录制由以下生命周期边界协作完成： -| 行 | 行为 | -| --------- | --------------------------------------------------------------------------------------------------- | -| 2487 | `await getTrafficRecordingSettings()` —— 每请求一次 DB 查询 | -| 2488-2490 | 计算 `shouldRecordSuccess` / `shouldRecordFailure` / `recorderEnabled` | -| 2491 | `recorderEnabled === true` 时才 `await readRequestBody(request)` 把请求体读进内存 | -| 3208 | `teeStreamForRecording(originalStream)` —— `ReadableStream.tee()` 分叉流，一路给 client，一路给录制 | -| 3603 | 流式成功路径：`return recordTrafficFixture(...)`，落盘在后台 `.then()` 里，client 响应已先行返回 | -| 3802 | 非流式成功路径：`void recordTrafficFixture(...).catch(...)` 显式 fire-and-forget | -| 4040 | 失败路径：`void recordTrafficFixture(...).catch(...)` 同上 | +| 模块 / 函数 | 行为 | +| ---------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------- | +| `proxy-request-lifecycle.ts` / `executeProxyRequest` | 读取录制设置；仅在成功或失败录制启用时读取 inbound body，并把录制上下文传给终态模块 | +| `proxy-non-stream-lifecycle.ts` / `settleNonStreamRequest` | 统一处理非流式成功与失败的请求日志、计费快照和 fixture | +| `proxy-stream-lifecycle.ts` / `settleStreamFailureRequest`、`createStreamResponse` | 统一处理流开始前失败、流式成功、下游取消和流中错误的日志、计费快照与 fixture | -**所有落盘均为 fire-and-forget**，client 端不阻塞等磁盘写入。读取请求体只在 `recorderEnabled === true` 时才发生，关闭录制时**不会**多产生 body 读取开销。 +所有 fixture 写入都通过 `recordTrafficFixture` 异步执行，client 响应不等待磁盘写入。读取请求体只在 `recorderEnabled === true` 时发生，关闭录制时**不会**多产生 body 读取开销。 ### 脱敏规则 diff --git a/docs/guide/usage/troubleshooting.md b/docs/guide/usage/troubleshooting.md index 554bb38c..8aaf3e88 100644 --- a/docs/guide/usage/troubleshooting.md +++ b/docs/guide/usage/troubleshooting.md @@ -11,30 +11,30 @@ outline: deep ## 一、客户端 Key 相关 -| 客户端看到的响应 | 触发位置 | 根因 / 排查方向 | -| ------------------------------------------------------------------------------------ | ----------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- | -| `401 {"error":"Missing API key"}` | `route.ts:2446-2449` | 三个 header（`Authorization` / `x-api-key` / `x-goog-api-key`）都没值。检查客户端 SDK 是否真的把 Key 注入到请求 | -| `401 {"error":"Invalid API key"}` | `route.ts:2454-2473` | 按 keyPrefix 找不到激活 key，或 hash 校验失败。先在管理后台用前缀搜确认 key 存在且 active | -| `401 {"error":"API key has expired"}` | `route.ts:2463-2465` | `candidate.expiresAt < new Date()`。如要延期，到管理后台改 `expires_at` | -| `403 {error:{code:"API_KEY_MODEL_NOT_ALLOWED", ...}}` | `route.ts:2507-2542` | Key 的 `allowedModels` 列表不含请求模型。要么把模型加进 allowedModels，要么换 Key | -| `403 {error:{code:"NO_AUTHORIZED_UPSTREAMS"}}` | `route.ts:2726-2745`、`load-balancer.ts:39-44` | Restricted 模式 Key 未绑定任何能匹配的上游，或绑定上游全被 model rule 排除。检查 Key→Upstream 绑定与上游 model_rules | -| `429 {error:{code:"API_KEY_QUOTA_EXCEEDED", user_hint:"当前密钥已达到消费限额..."}}` | `route.ts:163-280`、`api-key-quota-tracker.ts:62-133` | Key 已超 spending quota；仅 streaming + 可定价模型触发主动拒绝。In-memory tracker 同步周期：80% 以下 5 min，80%+ 紧急 1 min。涨额度后等下次同步生效 | +| 客户端看到的响应 | 触发位置 | 根因 / 排查方向 | +| ------------------------------------------------------------------------------------ | -------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- | +| `401 {"error":"Missing API key"}` | `proxy-request-lifecycle.ts` 的 `extractProxyApiKey` | 三个 header（`Authorization` / `x-api-key` / `x-goog-api-key`）都没值。检查客户端 SDK 是否真的把 Key 注入到请求 | +| `401 {"error":"Invalid API key"}` | `proxy-request-lifecycle.ts` 的 `executeProxyRequest` | 按 keyPrefix 找不到激活 key，或 hash 校验失败。先在管理后台用前缀搜确认 key 存在且 active | +| `401 {"error":"API key has expired"}` | `proxy-request-lifecycle.ts` 的 `executeProxyRequest` | `candidate.expiresAt < new Date()`。如要延期，到管理后台改 `expires_at` | +| `403 {error:{code:"API_KEY_MODEL_NOT_ALLOWED", ...}}` | `proxy-request-lifecycle.ts` 的 `executeProxyRequest` | Key 的 `allowedModels` 列表不含请求模型。要么把模型加进 allowedModels，要么换 Key | +| `403 {error:{code:"NO_AUTHORIZED_UPSTREAMS"}}` | `proxy-request-lifecycle.ts`、`load-balancer.ts` | Restricted 模式 Key 未绑定任何能匹配的上游，或绑定上游全被 model rule 排除。检查 Key→Upstream 绑定与上游 model_rules | +| `429 {error:{code:"API_KEY_QUOTA_EXCEEDED", user_hint:"当前密钥已达到消费限额..."}}` | `proxy-request-lifecycle.ts`、`api-key-quota-tracker.ts` | Key 已超 spending quota；仅 streaming + 可定价模型触发主动拒绝。In-memory tracker 同步周期：80% 以下 5 min，80%+ 紧急 1 min。涨额度后等下次同步生效 | 排查 Key 维度问题，最快路径是 `/logs?api_key_id=` 看最近一批请求的 status_code 与 `error_message`。 ## 二、路由 / 候选上游 -| 响应 | 触发位置 | 根因 / 排查方向 | -| -------------------------------------------------------------------------------- | ---------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `503 {error:{code:"NO_UPSTREAMS_CONFIGURED"}}`（reason：路径不支持） | `route.ts:2547-2625` | `resolveRouteCapability` 返回 null。请求 method+path 不在已知 capability 列表里 | -| `503 {error:{code:"NO_UPSTREAMS_CONFIGURED"}}`（reason：池为空） | `route.ts:2703-2724` | 活跃上游中没有一条声明匹配的 `route_capabilities`。检查上游列表 + `is_active` | -| `503 {error:{code:"NO_UPSTREAMS_CONFIGURED", reason:"NO_HEALTHY_CANDIDATES"}}` | `route.ts:2749-2813`、`route.ts:591-624` | Key 绑定上游或 capability 池在 model rule 过滤后为空。检查 `model_rules` 与拼写 | -| `503 {error:{code:"ALL_UPSTREAMS_UNAVAILABLE", reason:"UPSTREAM_CIRCUIT_OPEN"}}` | `load-balancer.ts`、`route.ts` | 所有候选被熔断拦截。`user_hint` 只带最快恢复秒数（各候选剩余时间的最小值，上游身份不下发），被熔断的具体上游看 routing_decision.excluded 的 `circuit_open` 条目。检查上游服务 / 密钥余额，或在管理后台手动关闭熔断。例外：`GET /v1/models` 在该场景下会回退本地模型目录返回 200 | -| `503 {error:{code:"ALL_UPSTREAMS_UNAVAILABLE", reason:"NO_HEALTHY_CANDIDATES"}}` | `load-balancer.ts`、`:243-303` | 所有 tier 遍历完毕但非熔断导致（如 quota 耗尽、显式排除）。看每个上游的 circuit_breaker_states 与上游绑定的 quota | -| `503 {error:{code:"ALL_UPSTREAMS_UNAVAILABLE", reason:"CONCURRENCY_FULL"}}` | `load-balancer.ts:49-62, 1129-1144` | `max_concurrency` 全打满且未启用队列 | -| `504 {error:{code:"QUEUE_WAIT_TIMEOUT"}}` | `upstream-queue-admission.ts:89-118` | 进入队列但 `timeout_ms` 内未拿到槽位 | -| `499 {error:{code:"CLIENT_DISCONNECTED"}}` | `upstream-queue-admission.ts:89-118` | 客户端在排队期间断开了连接 | -| `503 {error:{code:"QUEUE_FULL", reason:"queue_full"}}` | `upstream-queue-admission.ts:189-197` | 队列已达 `max_queue_length`，直接拒绝 | +| 响应 | 触发位置 | 根因 / 排查方向 | +| -------------------------------------------------------------------------------- | ----------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `503 {error:{code:"NO_UPSTREAMS_CONFIGURED"}}`（reason：路径不支持） | `route-capability-matcher.ts`、`proxy-request-lifecycle.ts` | `resolveRouteCapability` 返回 null。请求 method+path 不在已知 capability 列表里 | +| `503 {error:{code:"NO_UPSTREAMS_CONFIGURED"}}`（reason：池为空） | `proxy-request-lifecycle.ts` | 活跃上游中没有一条声明匹配的 `route_capabilities`。检查上游列表 + `is_active` | +| `503 {error:{code:"NO_UPSTREAMS_CONFIGURED", reason:"NO_HEALTHY_CANDIDATES"}}` | `proxy-request-lifecycle.ts`、`load-balancer.ts` | Key 绑定上游或 capability 池在 model rule 过滤后为空。检查 `model_rules` 与拼写 | +| `503 {error:{code:"ALL_UPSTREAMS_UNAVAILABLE", reason:"UPSTREAM_CIRCUIT_OPEN"}}` | `load-balancer.ts`、`proxy-execution.ts` | 所有候选被熔断拦截。`user_hint` 只带最快恢复秒数（各候选剩余时间的最小值，上游身份不下发），被熔断的具体上游看 routing_decision.excluded 的 `circuit_open` 条目。检查上游服务 / 密钥余额，或在管理后台手动关闭熔断。例外：`GET /v1/models` 在该场景下会回退本地模型目录返回 200 | +| `503 {error:{code:"ALL_UPSTREAMS_UNAVAILABLE", reason:"NO_HEALTHY_CANDIDATES"}}` | `load-balancer.ts`、`proxy-execution.ts` | 所有 tier 遍历完毕但非熔断导致（如 quota 耗尽、显式排除）。看每个上游的 circuit_breaker_states 与上游绑定的 quota | +| `503 {error:{code:"ALL_UPSTREAMS_UNAVAILABLE", reason:"CONCURRENCY_FULL"}}` | `load-balancer.ts:49-62, 1129-1144` | `max_concurrency` 全打满且未启用队列 | +| `504 {error:{code:"QUEUE_WAIT_TIMEOUT"}}` | `upstream-queue-admission.ts:89-118` | 进入队列但 `timeout_ms` 内未拿到槽位 | +| `499 {error:{code:"CLIENT_DISCONNECTED"}}` | `upstream-queue-admission.ts:89-118` | 客户端在排队期间断开了连接 | +| `503 {error:{code:"QUEUE_FULL", reason:"queue_full"}}` | `upstream-queue-admission.ts:189-197` | 队列已达 `max_queue_length`，直接拒绝 | ### HALF_OPEN 探针失败循环（症状不直接报错，但成功率低） @@ -54,7 +54,7 @@ outline: deep | 客户端看到 | 错误码 | 根因 | | -------------------------------------------------------------------- | ----------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------- | -| 流途中收到 `event: error` 后断流，data 含 `"code":"REQUEST_TIMEOUT"` | `REQUEST_TIMEOUT`（HTTP 504） | `streamIdleTimeout` 内未收到新数据块。`route.ts:2082-2084`、`proxy-client.ts` 的 `StreamIdleTimeoutError` | +| 流途中收到 `event: error` 后断流，data 含 `"code":"REQUEST_TIMEOUT"` | `REQUEST_TIMEOUT`（HTTP 504） | `streamIdleTimeout` 内未收到新数据块。`proxy-execution.ts` 的流读取与 `proxy-client.ts` 的 `StreamIdleTimeoutError` | | 流途中收到 `event: error`，data 含 `"code":"STREAM_ERROR"` | `STREAM_ERROR`（HTTP 502） | 流中其他读取异常（连接重置 / 协议错误等） | | 流开始前失败 | 走普通 5xx | 还能 failover，参见上一节 | | 流开始**后**中断 | 仅 SSE error event | **不可重试**——AutoRouter 不会做 mid-stream failover，因为已经向客户端发送了头与部分 body。详见 [请求生命周期](../architecture/request-lifecycle) 第六阶段 | diff --git a/src/app/api/proxy/v1/[...path]/proxy-request-lifecycle.ts b/src/app/api/proxy/v1/[...path]/proxy-request-lifecycle.ts index 96a0d667..12141c98 100644 --- a/src/app/api/proxy/v1/[...path]/proxy-request-lifecycle.ts +++ b/src/app/api/proxy/v1/[...path]/proxy-request-lifecycle.ts @@ -1,4 +1,3 @@ -import { NextRequest, NextResponse } from "next/server"; import { extractApiKey, getKeyPrefix, verifyApiKey } from "@/lib/utils/auth"; import { db, apiKeys, apiKeyUpstreams, upstreams, users, type Upstream } from "@/lib/db"; import { eq, and } from "drizzle-orm"; @@ -63,12 +62,7 @@ import type { RoutingQueueLog, RoutingSelectionReason, } from "@/types/api"; -import { - shouldRecordFixture, - readRequestBody, - buildFixture, - recordTrafficFixture, -} from "@/lib/services/traffic-recorder"; +import { shouldRecordFixture, readRequestBody } from "@/lib/services/traffic-recorder"; import { getTrafficRecordingSettings } from "@/lib/services/traffic-recording-service"; import { extractSessionId, @@ -103,6 +97,7 @@ import { computeAffinityTokens, createStreamResponse, resolveEffectiveServiceTier, + settleStreamFailureRequest, type StreamLifecycleContext, type StreamLifecycleTerminal, } from "./proxy-stream-lifecycle"; @@ -123,8 +118,6 @@ import { const log = createLogger("proxy-route"); -export type RouteContext = { params: Promise<{ path: string[] }> }; - async function persistBillingSnapshotSafely(input: { requestLogId: string; apiKeyId: string | null; @@ -155,11 +148,119 @@ async function persistBillingSnapshotSafely(input: { log.error({ err: error, requestId: input.requestId }, "failed to persist billing snapshot"); } } +type FailureFixture = NonStreamFailureTerminal["fixture"]; + +async function buildFailureFixtureMetadata(input: { + request: Request; + requestId: string; + matchedRouteCapability: RouteCapability; + compensationHeaders: CompensationHeader[]; + selectedCandidate: Upstream | null; + attributionFailoverAttempt?: FailoverAttempt; + errorStatusCode: number; + downstreamBody: unknown; +}): Promise { + const { + request, + requestId, + matchedRouteCapability, + compensationHeaders, + selectedCandidate, + attributionFailoverAttempt, + errorStatusCode, + downstreamBody, + } = input; + const fallbackOutboundHeaders = filterHeaders(new Headers(request.headers)).filtered; + applyCompensationHeaders(fallbackOutboundHeaders, compensationHeaders); + const fallbackProviderType = + selectedCandidate != null + ? resolveUpstreamProvider(selectedCandidate, matchedRouteCapability) + : getProviderByRouteCapability(matchedRouteCapability); + const fallbackUpstream = { + id: selectedCandidate?.id ?? "unknown", + name: selectedCandidate?.name ?? "unknown", + providerType: fallbackProviderType, + baseUrl: selectedCandidate?.baseUrl ?? "unknown", + }; + let outboundHeaders: Headers | Record = fallbackOutboundHeaders; + let upstreamForFixture = fallbackUpstream; + + if (attributionFailoverAttempt?.upstream_id) { + const attemptProvider = + attributionFailoverAttempt.upstream_provider_type === "openai" || + attributionFailoverAttempt.upstream_provider_type === "anthropic" || + attributionFailoverAttempt.upstream_provider_type === "google" + ? attributionFailoverAttempt.upstream_provider_type + : fallbackProviderType; + upstreamForFixture = { + id: attributionFailoverAttempt.upstream_id, + name: attributionFailoverAttempt.upstream_name, + providerType: attemptProvider, + baseUrl: + attributionFailoverAttempt.upstream_base_url ?? selectedCandidate?.baseUrl ?? "unknown", + }; + + try { + const attemptedUpstream = await db.query.upstreams.findFirst({ + where: eq(upstreams.id, attributionFailoverAttempt.upstream_id), + }); + if (attemptedUpstream) { + const attemptedUpstreamForProxy = prepareUpstreamForProxy(attemptedUpstream); + outboundHeaders = injectAuthHeader(fallbackOutboundHeaders, attemptedUpstreamForProxy); + upstreamForFixture = { + id: attemptedUpstream.id, + name: attemptedUpstream.name, + providerType: resolveUpstreamProvider(attemptedUpstream, matchedRouteCapability), + baseUrl: attemptedUpstreamForProxy.baseUrl, + }; + } + } catch (error) { + log.warn( + { err: error, requestId }, + "failed to resolve attempted upstream for failure fixture" + ); + } + } else if (selectedCandidate) { + try { + const upstreamForProxy = prepareUpstreamForProxy(selectedCandidate); + outboundHeaders = injectAuthHeader(fallbackOutboundHeaders, upstreamForProxy); + upstreamForFixture = { + id: selectedCandidate.id, + name: selectedCandidate.name, + providerType: resolveUpstreamProvider(selectedCandidate, matchedRouteCapability), + baseUrl: upstreamForProxy.baseUrl, + }; + } catch (error) { + log.warn( + { err: error, requestId }, + "failed to build upstream auth headers for failure fixture" + ); + } + } + + return { + providerType: fallbackProviderType, + responseSource: attributionFailoverAttempt?.status_code != null ? "upstream" : "gateway", + upstream: upstreamForFixture, + outboundHeaders, + response: { + statusCode: attributionFailoverAttempt?.status_code ?? errorStatusCode, + headers: attributionFailoverAttempt?.response_headers ?? {}, + bodyJson: attributionFailoverAttempt?.response_body_json ?? null, + bodyText: + attributionFailoverAttempt?.response_body_json == null + ? (attributionFailoverAttempt?.response_body_text ?? null) + : null, + }, + downstreamBody, + }; +} async function shouldRejectExceededApiKeyQuotaBeforeProxy(input: { quotaStatus: ReturnType; model: string | null; requestedStream: boolean; + requestId: string; }): Promise { if (!input.quotaStatus?.isExceeded) { @@ -227,7 +328,7 @@ async function logRejectedRequest(input: { apiKeyName?: string | null; apiKeyPrefix?: string | null; userId?: string | null; - request: NextRequest; + request: Request; path: string; model: string | null; reasoningEffort?: ReasoningEffort | null; @@ -306,7 +407,7 @@ async function logApiKeyQuotaRejectedRequest(input: { apiKeyName: string | null; apiKeyPrefix: string | null; userId: string | null; - request: NextRequest; + request: Request; path: string; model: string | null; reasoningEffort: ReasoningEffort | null; @@ -359,7 +460,7 @@ async function logApiKeyAdmissionRejectedRequest(input: { apiKeyName: string | null; apiKeyPrefix: string | null; userId: string | null; - request: NextRequest; + request: Request; path: string; model: string | null; reasoningEffort: ReasoningEffort | null; @@ -405,7 +506,7 @@ async function logLocalApiKeyModelListRequest(input: { apiKeyName: string | null; apiKeyPrefix: string | null; userId: string | null; - request: NextRequest; + request: Request; path: string; requestId: string; startTime: number; @@ -753,7 +854,7 @@ interface RequestContext { type AuthSource = "authorization" | "x-api-key" | "x-goog-api-key" | "none"; -function extractProxyApiKey(request: NextRequest): { +function extractProxyApiKey(request: Request): { keyValue: string | null; authSource: AuthSource; } { @@ -907,7 +1008,7 @@ function extractReasoningEffortFromBody( return null; } -async function extractRequestContext(request: NextRequest, path: string): Promise { +async function extractRequestContext(request: Request, path: string): Promise { const modelFromPath = extractGeminiModelFromPath(path); const requestUrl = new URL(request.url); @@ -954,16 +1055,13 @@ async function extractRequestContext(request: NextRequest, path: string): Promis } /** - * Handle all HTTP methods for proxy + * Execute the unified proxy lifecycle for an adapted HTTP request. */ -export async function handleProxy(request: NextRequest, context: RouteContext): Promise { +export async function executeProxyRequest(request: Request, path: string): Promise { const requestId = randomUUID().slice(0, 8); const startTime = Date.now(); let routingDurationMs: number | null = null; - // Extract path - const { path: pathSegments } = await context.params; - const path = pathSegments.join("/"); const requestUrl = new URL(request.url); const authRequestIsStream = isStreamRequest({}, path, requestUrl); const logAuthRejected = async (input: { @@ -1001,7 +1099,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): if (!keyValue) { log.debug({ requestId, authSource }, "proxy auth: missing supported API key header"); await logAuthRejected({ errorMessage: "Missing API key" }); - return NextResponse.json({ error: "Missing API key" }, { status: 401 }); + return Response.json({ error: "Missing API key" }, { status: 401 }); } log.debug({ requestId, authSource }, "proxy auth: extracted API key"); @@ -1024,7 +1122,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): apiKeyPrefix: candidate.keyPrefix, userId: candidate.userId, }); - return NextResponse.json({ error: "API key has expired" }, { status: 401 }); + return Response.json({ error: "API key has expired" }, { status: 401 }); } validApiKey = candidate; break; @@ -1033,7 +1131,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): if (!validApiKey) { await logAuthRejected({ errorMessage: "Invalid API key", apiKeyPrefix: keyPrefix }); - return NextResponse.json({ error: "Invalid API key" }, { status: 401 }); + return Response.json({ error: "Invalid API key" }, { status: 401 }); } // Deactivating a user cascades to their keys at the proxy boundary: a key @@ -1056,7 +1154,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): apiKeyPrefix: validApiKey.keyPrefix, userId: validApiKey.userId, }); - return NextResponse.json({ error: "API key is disabled" }, { status: 401 }); + return Response.json({ error: "API key is disabled" }, { status: 401 }); } } @@ -1806,6 +1904,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): inboundBody, trafficRecordingSettings, shouldRecordSuccess, + shouldRecordFailure, getCompensationHeaders: () => compensationHeaders, getQueueStatePersistence: () => queueStatePersistence, awaitRequestLogReady, @@ -1815,6 +1914,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): requestLogId = value; }, persistBillingSnapshot: persistBillingSnapshotSafely, + settlement: { response: null }, }; try { // Prepare affinity context if session ID is available @@ -2318,98 +2418,19 @@ export async function handleProxy(request: NextRequest, context: RouteContext): if (!requestedStream) { const failureResponse = createUnifiedErrorResponse(errorCode, errorDetails); - let failureFixture: NonStreamFailureTerminal["fixture"]; - if (shouldRecordFailure && inboundBody && didSendUpstream) { - const fallbackOutboundHeaders = filterHeaders(new Headers(request.headers)).filtered; - applyCompensationHeaders(fallbackOutboundHeaders, compensationHeaders); - const fallbackProviderType = - selectedCandidate != null - ? resolveUpstreamProvider(selectedCandidate, matchedRouteCapability) - : getProviderByRouteCapability(matchedRouteCapability); - const fallbackUpstream = { - id: selectedCandidate?.id ?? "unknown", - name: selectedCandidate?.name ?? "unknown", - providerType: fallbackProviderType, - baseUrl: selectedCandidate?.baseUrl ?? "unknown", - }; - let outboundHeaders: Headers | Record = fallbackOutboundHeaders; - let upstreamForFixture = fallbackUpstream; - - if (attributionFailoverAttempt?.upstream_id) { - const attemptProvider = - attributionFailoverAttempt.upstream_provider_type === "openai" || - attributionFailoverAttempt.upstream_provider_type === "anthropic" || - attributionFailoverAttempt.upstream_provider_type === "google" - ? attributionFailoverAttempt.upstream_provider_type - : fallbackProviderType; - upstreamForFixture = { - id: attributionFailoverAttempt.upstream_id, - name: attributionFailoverAttempt.upstream_name, - providerType: attemptProvider, - baseUrl: - attributionFailoverAttempt.upstream_base_url ?? - selectedCandidate?.baseUrl ?? - "unknown", - }; - - try { - const attemptedUpstream = await db.query.upstreams.findFirst({ - where: eq(upstreams.id, attributionFailoverAttempt.upstream_id), - }); - if (attemptedUpstream) { - const attemptedUpstreamForProxy = prepareUpstreamForProxy(attemptedUpstream); - outboundHeaders = injectAuthHeader( - fallbackOutboundHeaders, - attemptedUpstreamForProxy - ); - upstreamForFixture = { - id: attemptedUpstream.id, - name: attemptedUpstream.name, - providerType: resolveUpstreamProvider(attemptedUpstream, matchedRouteCapability), - baseUrl: attemptedUpstreamForProxy.baseUrl, - }; - } - } catch (recorderBuildError) { - log.warn( - { err: recorderBuildError, requestId }, - "failed to resolve attempted upstream for non-stream failure fixture" - ); - } - } else if (selectedCandidate) { - try { - const upstreamForProxy = prepareUpstreamForProxy(selectedCandidate); - outboundHeaders = injectAuthHeader(fallbackOutboundHeaders, upstreamForProxy); - upstreamForFixture = { - id: selectedCandidate.id, - name: selectedCandidate.name, - providerType: resolveUpstreamProvider(selectedCandidate, matchedRouteCapability), - baseUrl: upstreamForProxy.baseUrl, - }; - } catch (recorderBuildError) { - log.warn( - { err: recorderBuildError, requestId }, - "failed to build upstream auth headers for non-stream failure fixture" - ); - } - } - - failureFixture = { - providerType: fallbackProviderType, - responseSource: attributionFailoverAttempt?.status_code != null ? "upstream" : "gateway", - upstream: upstreamForFixture, - outboundHeaders, - response: { - statusCode: attributionFailoverAttempt?.status_code ?? errorStatusCode, - headers: attributionFailoverAttempt?.response_headers ?? {}, - bodyJson: attributionFailoverAttempt?.response_body_json ?? null, - bodyText: - attributionFailoverAttempt?.response_body_json == null - ? (attributionFailoverAttempt?.response_body_text ?? null) - : null, - }, - downstreamBody: downstreamErrorBody, - }; - } + const failureFixture = + shouldRecordFailure && inboundBody && didSendUpstream + ? await buildFailureFixtureMetadata({ + request, + requestId, + matchedRouteCapability, + compensationHeaders, + selectedCandidate, + attributionFailoverAttempt, + errorStatusCode, + downstreamBody: downstreamErrorBody, + }) + : undefined; if (error instanceof ClientDisconnectedError) { log.warn({ requestId }, "client disconnected, no response sent"); @@ -2437,181 +2458,43 @@ export async function handleProxy(request: NextRequest, context: RouteContext): }); } - if (shouldRecordFailure && inboundBody && didSendUpstream) { - const fallbackOutboundHeaders = filterHeaders(new Headers(request.headers)).filtered; - applyCompensationHeaders(fallbackOutboundHeaders, compensationHeaders); - const fallbackProviderType = - selectedCandidate != null - ? resolveUpstreamProvider(selectedCandidate, matchedRouteCapability) - : getProviderByRouteCapability(matchedRouteCapability); - const fallbackUpstream = { - id: didSendUpstream ? (selectedCandidate?.id ?? "unknown") : "unknown", - name: didSendUpstream ? (selectedCandidate?.name ?? "unknown") : "not-sent", - providerType: fallbackProviderType, - baseUrl: didSendUpstream ? (selectedCandidate?.baseUrl ?? "unknown") : "unknown", - }; - let outboundHeaders: Headers | Record = didSendUpstream - ? fallbackOutboundHeaders - : {}; - let upstreamForFixture = fallbackUpstream; - - if (didSendUpstream && attributionFailoverAttempt?.upstream_id) { - const attemptProvider = - attributionFailoverAttempt.upstream_provider_type === "openai" || - attributionFailoverAttempt.upstream_provider_type === "anthropic" || - attributionFailoverAttempt.upstream_provider_type === "google" - ? attributionFailoverAttempt.upstream_provider_type - : fallbackProviderType; - upstreamForFixture = { - id: attributionFailoverAttempt.upstream_id, - name: attributionFailoverAttempt.upstream_name, - providerType: attemptProvider, - baseUrl: - attributionFailoverAttempt.upstream_base_url ?? selectedCandidate?.baseUrl ?? "unknown", - }; - - try { - const attemptedUpstream = await db.query.upstreams.findFirst({ - where: eq(upstreams.id, attributionFailoverAttempt.upstream_id), - }); - if (attemptedUpstream) { - const attemptedUpstreamForProxy = prepareUpstreamForProxy(attemptedUpstream); - outboundHeaders = injectAuthHeader(fallbackOutboundHeaders, attemptedUpstreamForProxy); - upstreamForFixture = { - id: attemptedUpstream.id, - name: attemptedUpstream.name, - providerType: resolveUpstreamProvider(attemptedUpstream, matchedRouteCapability), - baseUrl: attemptedUpstreamForProxy.baseUrl, - }; - } - } catch (recorderBuildError) { - log.warn( - { err: recorderBuildError, requestId }, - "failed to resolve attempted upstream for failure fixture" - ); - } - } else if (didSendUpstream && selectedCandidate) { - try { - const upstreamForProxy = prepareUpstreamForProxy(selectedCandidate); - outboundHeaders = injectAuthHeader(fallbackOutboundHeaders, upstreamForProxy); - upstreamForFixture = { - id: selectedCandidate.id, - name: selectedCandidate.name, - providerType: resolveUpstreamProvider(selectedCandidate, matchedRouteCapability), - baseUrl: upstreamForProxy.baseUrl, - }; - } catch (recorderBuildError) { - log.warn( - { err: recorderBuildError, requestId }, - "failed to build upstream auth headers for failure fixture" - ); - } - } - - const failureFixture = buildFixture({ - requestId, - startTime, - providerType: fallbackProviderType, - route: path, - model: resolvedModel, - inboundRequest: { - method: request.method, - path, - headers: request.headers, - bodyText: inboundBody.text, - bodyJson: inboundBody.json, - }, - upstream: upstreamForFixture, - outboundHeaders, - response: { - statusCode: attributionFailoverAttempt?.status_code ?? errorStatusCode, - headers: attributionFailoverAttempt?.response_headers ?? {}, - bodyJson: attributionFailoverAttempt?.response_body_json ?? null, - bodyText: - attributionFailoverAttempt?.response_body_json == null - ? (attributionFailoverAttempt?.response_body_text ?? null) - : null, - }, - outboundRequestSent: didSendUpstream, - outboundResponseSource: - didSendUpstream && attributionFailoverAttempt?.status_code != null - ? "upstream" - : "gateway", - downstreamResponse: { - statusCode: errorStatusCode, - headers: { "content-type": "application/json" }, - bodyJson: downstreamErrorBody, - }, - failoverHistory: failoverHistory.length > 0 ? failoverHistory : null, - redactSensitive: trafficRecordingSettings.redactSensitive, - }); - - void recordTrafficFixture(failureFixture, { - requestLogId, - apiKeyId: validApiKey.id, - upstreamId: actualUpstreamId, - method: request.method, - path, - model: resolvedModel, - statusCode: errorStatusCode, - outcome: "failure", - }).catch((recordError) => - log.error({ err: recordError, requestId }, "failed to record error fixture") - ); - } - - // Log failed request (internal logging with full details) - await queueStatePersistence; - const failureLogFields = { - ...failureLogBaseFields, - upstreamId: actualUpstreamId, - model: resolvedModel, - statusCode: errorStatusCode, - errorMessage, - }; - await awaitRequestLogReady(); - let persistedLogId: string | null = requestLogId; - if (requestLogId) { - const updatedLog = await updateRequestLog(requestLogId, failureLogFields); - persistedLogId = updatedLog?.id ?? requestLogId; - } else { - const createdLog = await logRequest({ - apiKeyId: validApiKey.id, - method: request.method, - path, - ...failureLogFields, - }); - persistedLogId = createdLog.id; - } - - if (persistedLogId && didSendUpstream) { - await persistBillingSnapshotSafely({ - requestLogId: persistedLogId, - apiKeyId: validApiKey.id, - upstreamId: actualUpstreamId, - model: resolvedModel, - requestedServiceTier, - effectiveServiceTier: null, - usage: { - promptTokens: 0, - completionTokens: 0, - totalTokens: 0, - cacheReadTokens: 0, - cacheWriteTokens: 0, - }, - requestId, - }); - } + const failureResponse = createUnifiedErrorResponse(errorCode, errorDetails); + const failureFixture = + shouldRecordFailure && inboundBody && didSendUpstream + ? await buildFailureFixtureMetadata({ + request, + requestId, + matchedRouteCapability, + compensationHeaders, + selectedCandidate, + attributionFailoverAttempt, + errorStatusCode, + downstreamBody: downstreamErrorBody, + }) + : undefined; - // Handle client disconnect silently (no response needed) if (error instanceof ClientDisconnectedError) { log.warn({ requestId }, "client disconnected, no response sent"); - return createUnifiedErrorResponse(errorCode, errorDetails); } - if (errorCode === "SERVICE_UNAVAILABLE") { log.error({ err: error, requestId }, "proxy error"); } - return createUnifiedErrorResponse(errorCode, errorDetails); + + return settleStreamFailureRequest(streamLifecycleContext, { + response: failureResponse, + errorStatusCode, + errorMessage, + actualUpstreamId, + resolvedModel, + didSendUpstream, + failoverHistory, + routingDecision: failureRoutingDecisionLog, + routingType, + priorityTier, + routingDurationMs, + sessionIdCompensated, + headerDiff: failureHeaderDiff, + ...(failureFixture ? { fixture: failureFixture } : {}), + }); } } diff --git a/src/app/api/proxy/v1/[...path]/proxy-stream-lifecycle.ts b/src/app/api/proxy/v1/[...path]/proxy-stream-lifecycle.ts index 79cd7d95..d33756f3 100644 --- a/src/app/api/proxy/v1/[...path]/proxy-stream-lifecycle.ts +++ b/src/app/api/proxy/v1/[...path]/proxy-stream-lifecycle.ts @@ -75,6 +75,7 @@ export interface StreamLifecycleContext { inboundBody: InboundBody | null; trafficRecordingSettings: Pick; shouldRecordSuccess: boolean; + shouldRecordFailure: boolean; getCompensationHeaders: () => CompensationHeader[]; getQueueStatePersistence: () => Promise; awaitRequestLogReady: () => Promise; @@ -91,6 +92,9 @@ export interface StreamLifecycleContext { usage: StreamBillingUsage; requestId: string; }) => Promise; + settlement: { + response: Response | null; + }; } export interface StreamLifecycleTerminal { @@ -109,6 +113,40 @@ export interface StreamLifecycleTerminal { routingDurationMs: number | null; } +interface StreamFailureLifecycleTerminal { + response: Response; + errorStatusCode: number; + errorMessage: string; + actualUpstreamId: string | null; + resolvedModel: string | null; + didSendUpstream: boolean; + failoverHistory: FailoverAttempt[]; + routingDecision: RoutingDecisionLog; + routingType: "tiered" | "direct" | "provider_type" | null; + priorityTier: number | null; + routingDurationMs: number | null; + sessionIdCompensated: boolean; + headerDiff: HeaderDiff | null; + fixture?: { + providerType: string; + responseSource: "upstream" | "gateway"; + upstream: { + id: string; + name: string; + providerType: string; + baseUrl: string; + }; + outboundHeaders: Headers | Record; + response: { + statusCode: number; + headers: Headers | Record; + bodyText?: string | null; + bodyJson?: unknown | null; + }; + downstreamBody: unknown; + }; +} + type StreamLogFields = Omit; type StreamTerminalOutcome = "success" | "failure" | "disconnect"; type StreamOutcome = @@ -353,14 +391,15 @@ async function persistTerminalRequestLog( async function persistZeroUsageBilling( context: StreamLifecycleContext, - terminal: StreamLifecycleTerminal, - requestLogId: string + requestLogId: string, + upstreamId: string | null, + model: string | null ): Promise { await context.persistBillingSnapshot({ requestLogId, apiKeyId: context.apiKeyId, - upstreamId: terminal.upstream.id, - model: terminal.resolvedModel, + upstreamId, + model, requestedServiceTier: context.requestedServiceTier, effectiveServiceTier: null, usage: { @@ -374,6 +413,101 @@ async function persistZeroUsageBilling( }); } +function buildRequestFailureLogFields( + context: StreamLifecycleContext, + terminal: StreamFailureLifecycleTerminal, + durationMs: number +): StreamLogFields { + return { + ...context.apiKeySnapshot, + upstreamId: terminal.actualUpstreamId, + method: context.request.method, + path: context.path, + model: terminal.resolvedModel, + reasoningEffort: context.reasoningEffort, + requestedServiceTier: context.requestedServiceTier, + effectiveServiceTier: null, + promptTokens: 0, + completionTokens: 0, + totalTokens: 0, + statusCode: terminal.errorStatusCode, + durationMs, + errorMessage: terminal.errorMessage, + routingType: terminal.routingType, + priorityTier: terminal.priorityTier, + failoverAttempts: terminal.failoverHistory.length, + failoverHistory: terminal.failoverHistory.length > 0 ? terminal.failoverHistory : null, + routingDecision: terminal.routingDecision, + thinkingConfig: context.thinkingConfig, + sessionId: context.sessionId, + isStream: true, + routingDurationMs: terminal.routingDurationMs, + sessionIdCompensated: terminal.sessionIdCompensated, + headerDiff: terminal.headerDiff, + }; +} + +function buildAndRecordFailureFixture( + context: StreamLifecycleContext, + terminal: StreamFailureLifecycleTerminal, + inboundBody: InboundBody, + requestLogId: string | null +): void { + if (!context.shouldRecordFailure || !terminal.fixture || !terminal.didSendUpstream) { + return; + } + + try { + const fixture = buildFixture({ + requestId: context.requestId, + startTime: context.startTime, + providerType: terminal.fixture.providerType, + route: context.path, + model: terminal.resolvedModel, + inboundRequest: { + method: context.request.method, + path: context.path, + headers: context.request.headers, + bodyText: inboundBody.text, + bodyJson: inboundBody.json, + }, + upstream: terminal.fixture.upstream, + outboundHeaders: terminal.fixture.outboundHeaders, + response: terminal.fixture.response, + outboundRequestSent: true, + outboundResponseSource: terminal.fixture.responseSource, + downstreamResponse: { + statusCode: terminal.errorStatusCode, + headers: { "content-type": "application/json" }, + bodyJson: terminal.fixture.downstreamBody, + }, + failoverHistory: terminal.failoverHistory.length > 0 ? terminal.failoverHistory : null, + redactSensitive: context.trafficRecordingSettings.redactSensitive, + }); + + void recordTrafficFixture(fixture, { + requestLogId, + apiKeyId: context.apiKeyId, + upstreamId: terminal.actualUpstreamId, + method: context.request.method, + path: context.path, + model: terminal.resolvedModel, + statusCode: terminal.errorStatusCode, + outcome: "failure", + }).catch((error) => + log.error( + { err: error, requestId: context.requestId }, + "failed to record stream failure fixture" + ) + ); + } catch (error) { + log.error( + { err: error, requestId: context.requestId }, + "failed to build stream failure fixture" + ); + } +} + function buildAndRecordSuccessFixture( context: StreamLifecycleContext, terminal: StreamLifecycleTerminal, @@ -434,6 +568,40 @@ function buildAndRecordSuccessFixture( } } +/** + * Settle a stream request failure discovered before an SSE response is returned. + * The lifecycle owns the terminal response, request log, zero-usage billing and failure recording. + */ +export async function settleStreamFailureRequest( + context: StreamLifecycleContext, + terminal: StreamFailureLifecycleTerminal +): Promise { + if (context.settlement.response) { + return context.settlement.response; + } + + const persistedLogId = await persistTerminalRequestLog( + context, + buildRequestFailureLogFields(context, terminal, Date.now() - context.startTime) + ); + + if (persistedLogId && terminal.didSendUpstream) { + await persistZeroUsageBilling( + context, + persistedLogId, + terminal.actualUpstreamId, + terminal.resolvedModel + ); + } + + if (context.inboundBody) { + buildAndRecordFailureFixture(context, terminal, context.inboundBody, persistedLogId); + } + + context.settlement.response = terminal.response; + return terminal.response; +} + function wrapStreamWithDownstreamSettlement( stream: ReadableStream, abortSignal: AbortSignal | undefined, @@ -616,7 +784,12 @@ export function createStreamResponse( buildDisconnectLogFields(context, terminal, Date.now() - context.startTime) ); if (requestLogId) { - await persistZeroUsageBilling(context, terminal, requestLogId); + await persistZeroUsageBilling( + context, + requestLogId, + terminal.upstream.id, + terminal.resolvedModel + ); } }); }; @@ -634,7 +807,12 @@ export function createStreamResponse( ); const requestLogId = await persistTerminalRequestLog(context, failureFields.fields); if (requestLogId) { - await persistZeroUsageBilling(context, terminal, requestLogId); + await persistZeroUsageBilling( + context, + requestLogId, + terminal.upstream.id, + terminal.resolvedModel + ); } }); }; diff --git a/src/app/api/proxy/v1/[...path]/route.ts b/src/app/api/proxy/v1/[...path]/route.ts index f7dcac6b..fd292523 100644 --- a/src/app/api/proxy/v1/[...path]/route.ts +++ b/src/app/api/proxy/v1/[...path]/route.ts @@ -1,41 +1,48 @@ import { NextRequest } from "next/server"; -import { handleProxy, type RouteContext } from "./proxy-request-lifecycle"; +import { executeProxyRequest } from "./proxy-request-lifecycle"; // Edge runtime for streaming support export const runtime = "nodejs"; export const dynamic = "force-dynamic"; +export type RouteContext = { params: Promise<{ path: string[] }> }; + +async function handleProxyRoute(request: NextRequest, context: RouteContext): Promise { + const { path } = await context.params; + return executeProxyRequest(request, path.join("/")); +} + /** * Handle proxied GET requests through the unified proxy pipeline. */ export async function GET(request: NextRequest, context: RouteContext) { - return handleProxy(request, context); + return handleProxyRoute(request, context); } /** * Handle proxied POST requests through the unified proxy pipeline. */ export async function POST(request: NextRequest, context: RouteContext) { - return handleProxy(request, context); + return handleProxyRoute(request, context); } /** * Handle proxied PUT requests through the unified proxy pipeline. */ export async function PUT(request: NextRequest, context: RouteContext) { - return handleProxy(request, context); + return handleProxyRoute(request, context); } /** * Handle proxied DELETE requests through the unified proxy pipeline. */ export async function DELETE(request: NextRequest, context: RouteContext) { - return handleProxy(request, context); + return handleProxyRoute(request, context); } /** * Handle proxied PATCH requests through the unified proxy pipeline. */ export async function PATCH(request: NextRequest, context: RouteContext) { - return handleProxy(request, context); + return handleProxyRoute(request, context); } diff --git a/tests/unit/api/proxy/proxy-stream-lifecycle.test.ts b/tests/unit/api/proxy/proxy-stream-lifecycle.test.ts index 917a87a7..c184470a 100644 --- a/tests/unit/api/proxy/proxy-stream-lifecycle.test.ts +++ b/tests/unit/api/proxy/proxy-stream-lifecycle.test.ts @@ -79,11 +79,12 @@ vi.mock("@/app/api/proxy/v1/[...path]/proxy-execution", () => ({ resolveUpstreamProvider: vi.fn(() => "openai"), })); -const { createStreamResponse } = +const { createStreamResponse, settleStreamFailureRequest } = await import("@/app/api/proxy/v1/[...path]/proxy-stream-lifecycle"); type StreamLifecycleContext = Parameters[0]; type StreamLifecycleTerminal = Parameters[1]; +type StreamFailureLifecycleTerminal = Parameters[1]; const ROUTING_DECISION: RoutingDecisionLog = { original_model: "gpt-4.1", @@ -142,6 +143,7 @@ function makeContext(signal: AbortSignal): StreamLifecycleContext { }, trafficRecordingSettings: { redactSensitive: true }, shouldRecordSuccess: true, + shouldRecordFailure: false, getCompensationHeaders: () => [], getQueueStatePersistence: vi.fn(async () => undefined), awaitRequestLogReady: vi.fn(async () => requestLogId), @@ -151,6 +153,7 @@ function makeContext(signal: AbortSignal): StreamLifecycleContext { requestLogId = value; }, persistBillingSnapshot: mocks.persistBillingSnapshot, + settlement: { response: null }, }; } @@ -415,3 +418,58 @@ describe("createStreamResponse", () => { expect(upstreamCancelled).toBe(true); }); }); +describe("settleStreamFailureRequest", () => { + it("settles request failure side effects once before returning the response", async () => { + const context = makeContext(new AbortController().signal); + context.shouldRecordFailure = true; + const response = Response.json({ error: { code: "SERVICE_UNAVAILABLE" } }, { status: 503 }); + const terminal: StreamFailureLifecycleTerminal = { + response, + errorStatusCode: 503, + errorMessage: "upstream failed", + actualUpstreamId: UPSTREAM.id, + resolvedModel: "gpt-4.1", + didSendUpstream: true, + failoverHistory: [], + routingDecision: ROUTING_DECISION, + routingType: "tiered", + priorityTier: 0, + routingDurationMs: 4, + sessionIdCompensated: false, + headerDiff: null, + fixture: { + providerType: "openai", + responseSource: "upstream", + upstream: { + id: UPSTREAM.id, + name: UPSTREAM.name, + providerType: "openai", + baseUrl: UPSTREAM.baseUrl, + }, + outboundHeaders: {}, + response: { + statusCode: 500, + headers: {}, + bodyJson: { error: { message: "upstream failed" } }, + }, + downstreamBody: { error: { code: "SERVICE_UNAVAILABLE" } }, + }, + }; + + await expect(settleStreamFailureRequest(context, terminal)).resolves.toBe(response); + await expect(settleStreamFailureRequest(context, terminal)).resolves.toBe(response); + + expect(mocks.updateRequestLog).toHaveBeenCalledTimes(1); + expect(mocks.persistBillingSnapshot).toHaveBeenCalledTimes(1); + expect(mocks.buildFixture).toHaveBeenCalledTimes(1); + expect(mocks.recordTrafficFixture).toHaveBeenCalledTimes(1); + expect(mocks.updateRequestLog).toHaveBeenCalledWith( + "log-1", + expect.objectContaining({ + statusCode: 503, + isStream: true, + errorMessage: "upstream failed", + }) + ); + }); +}); diff --git a/tests/unit/api/proxy/route-adapter.test.ts b/tests/unit/api/proxy/route-adapter.test.ts new file mode 100644 index 00000000..ad81856f --- /dev/null +++ b/tests/unit/api/proxy/route-adapter.test.ts @@ -0,0 +1,37 @@ +import { describe, expect, it, vi } from "vitest"; +import { NextRequest } from "next/server"; + +const { executeProxyRequest } = vi.hoisted(() => ({ + executeProxyRequest: vi.fn(async (request: Request, path: string) => + Response.json({ method: request.method, path }) + ), +})); + +vi.mock("@/app/api/proxy/v1/[...path]/proxy-request-lifecycle", () => ({ + executeProxyRequest, +})); + +const route = await import("@/app/api/proxy/v1/[...path]/route"); + +describe("proxy HTTP adapter", () => { + it.each([ + ["GET", route.GET], + ["POST", route.POST], + ["PUT", route.PUT], + ["DELETE", route.DELETE], + ["PATCH", route.PATCH], + ])("delegates %s through the shared request lifecycle", async (method, handler) => { + executeProxyRequest.mockClear(); + const request = new NextRequest("http://localhost/api/proxy/v1/chat/completions", { + method, + }); + + const response = await handler(request, { + params: Promise.resolve({ path: ["chat", "completions"] }), + }); + + expect(response.status).toBe(200); + expect(executeProxyRequest).toHaveBeenCalledTimes(1); + expect(executeProxyRequest).toHaveBeenCalledWith(request, "chat/completions"); + }); +}); diff --git a/tests/unit/api/proxy/route.test.ts b/tests/unit/api/proxy/route.test.ts index 1c75d8a2..61924708 100644 --- a/tests/unit/api/proxy/route.test.ts +++ b/tests/unit/api/proxy/route.test.ts @@ -557,10 +557,7 @@ describe("proxy route upstream selection", () => { request: NextRequest, context: { params: Promise<{ path: string[] }> } ) => Promise; - let handleProxy: ( - request: NextRequest, - context: { params: Promise<{ path: string[] }> } - ) => Promise; + let executeProxyRequest: (request: Request, path: string) => Promise; let GET: ( request: NextRequest, context: { params: Promise<{ path: string[] }> } @@ -601,7 +598,7 @@ describe("proxy route upstream selection", () => { const { db } = await import("@/lib/db"); POST = routeModule.POST; const lifecycleModule = await import("@/app/api/proxy/v1/[...path]/proxy-request-lifecycle"); - handleProxy = lifecycleModule.handleProxy; + executeProxyRequest = lifecycleModule.executeProxyRequest; GET = routeModule.GET; vi.mocked(db.query.upstreams.findMany).mockResolvedValue(DEFAULT_ACTIVE_UPSTREAMS); vi.mocked(db.query.upstreamHealth.findMany).mockResolvedValue([]); @@ -695,7 +692,8 @@ describe("proxy route upstream selection", () => { const { calculateAndPersistRequestBillingSnapshot } = await import("@/lib/services/billing-cost-service"); const { buildFixture, recordTrafficFixture } = await import("@/lib/services/traffic-recorder"); - const { handleProxy } = await import("@/app/api/proxy/v1/[...path]/proxy-request-lifecycle"); + const { executeProxyRequest } = + await import("@/app/api/proxy/v1/[...path]/proxy-request-lifecycle"); const upstream = DEFAULT_ACTIVE_UPSTREAMS[0]; const responseBody = { id: "lifecycle-success", object: "chat.completion" }; const lifecycleEvents: string[] = []; @@ -768,7 +766,7 @@ describe("proxy route upstream selection", () => { process.env.RECORDER_ENABLED = "true"; process.env.RECORDER_MODE = "success"; - const response = await handleProxy( + const response = await executeProxyRequest( new NextRequest("http://localhost/api/proxy/v1/chat/completions", { method: "POST", headers: { @@ -780,7 +778,7 @@ describe("proxy route upstream selection", () => { messages: [{ role: "user", content: "hello" }], }), }), - { params: Promise.resolve({ path: ["chat", "completions"] }) } + "chat/completions" ); expect(response.status).toBe(200); @@ -815,7 +813,8 @@ describe("proxy route upstream selection", () => { }); it("logs a lifecycle rejection without billing or recording when capability is missing", async () => { const { db } = await import("@/lib/db"); - const { handleProxy } = await import("@/app/api/proxy/v1/[...path]/proxy-request-lifecycle"); + const { executeProxyRequest } = + await import("@/app/api/proxy/v1/[...path]/proxy-request-lifecycle"); const { logRequest } = await import("@/lib/services/request-logger"); const { calculateAndPersistRequestBillingSnapshot } = await import("@/lib/services/billing-cost-service"); @@ -832,8 +831,7 @@ describe("proxy route upstream selection", () => { isActive: true, }, ]); - - const response = await handleProxy( + const response = await executeProxyRequest( new NextRequest("http://localhost/api/proxy/v1/custom/not-matched", { method: "POST", headers: { @@ -842,7 +840,7 @@ describe("proxy route upstream selection", () => { }, body: JSON.stringify({ model: "gpt-5.2", input: "hello" }), }), - { params: Promise.resolve({ path: ["custom", "not-matched"] }) } + "custom/not-matched" ); expect(response.status).toBe(503); @@ -861,7 +859,8 @@ describe("proxy route upstream selection", () => { expect(recordTrafficFixture).not.toHaveBeenCalled(); }); it("logs missing API key rejection without billing or recording", async () => { - const { handleProxy } = await import("@/app/api/proxy/v1/[...path]/proxy-request-lifecycle"); + const { executeProxyRequest } = + await import("@/app/api/proxy/v1/[...path]/proxy-request-lifecycle"); const { logRequest } = await import("@/lib/services/request-logger"); const { calculateAndPersistRequestBillingSnapshot } = await import("@/lib/services/billing-cost-service"); @@ -870,13 +869,13 @@ describe("proxy route upstream selection", () => { process.env.RECORDER_ENABLED = "true"; process.env.RECORDER_MODE = "all"; - const response = await handleProxy( + const response = await executeProxyRequest( new NextRequest("http://localhost/api/proxy/v1/chat/completions?alt=sse", { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ model: "gpt-5.2", messages: [] }), }), - { params: Promise.resolve({ path: ["chat", "completions"] }) } + "chat/completions" ); expect(response.status).toBe(401); @@ -3803,9 +3802,7 @@ describe("proxy route upstream selection", () => { }), }); - const response = await handleProxy(request, { - params: Promise.resolve({ path: ["v1", "messages"] }), - }); + const response = await executeProxyRequest(request, "v1/messages"); expect(response.status).toBe(200); expect(markUnhealthy).not.toHaveBeenCalledWith("up-anthropic-1", expect.any(String)); @@ -3905,11 +3902,8 @@ describe("proxy route upstream selection", () => { }), }); - const response = await handleProxy(request, { - params: Promise.resolve({ path: ["v1", "messages"] }), - }); + const response = await executeProxyRequest(request, "v1/messages"); const payload = (await response.json()) as { error: { reason?: string; user_hint?: string } }; - expect(response.status).toBe(503); expect(payload.error.reason).toBe("CONCURRENCY_FULL"); expect(payload.error.user_hint).toContain("并发上限"); @@ -4042,9 +4036,7 @@ describe("proxy route upstream selection", () => { }), }); - const response = await handleProxy(request, { - params: Promise.resolve({ path: ["v1", "messages"] }), - }); + const response = await executeProxyRequest(request, "v1/messages"); expect(response.status).toBe(200); expect(vi.mocked(upstreamQueueAdmission.enqueueWait)).toHaveBeenCalledWith( @@ -4171,7 +4163,7 @@ describe("proxy route upstream selection", () => { }; }); - const response = await handleProxy( + const response = await executeProxyRequest( new NextRequest("http://localhost/api/proxy/v1/messages", { method: "POST", signal: controller.signal, @@ -4184,7 +4176,7 @@ describe("proxy route upstream selection", () => { messages: [{ role: "user", content: "hi" }], }), }), - { params: Promise.resolve({ path: ["v1", "messages"] }) } + "v1/messages" ); const data = await response.json(); @@ -4701,9 +4693,7 @@ describe("proxy route upstream selection", () => { }), }); - const response = await handleProxy(request, { - params: Promise.resolve({ path: ["v1", "messages"] }), - }); + const response = await executeProxyRequest(request, "v1/messages"); expect(response.status).toBe(200); expect(vi.mocked(reselectQueuedUpstreamOnce)).toHaveBeenCalledWith( @@ -4819,11 +4809,8 @@ describe("proxy route upstream selection", () => { }), }); - const response = await handleProxy(request, { - params: Promise.resolve({ path: ["v1", "messages"] }), - }); + const response = await executeProxyRequest(request, "v1/messages"); const data = await response.json(); - expect(response.status).toBe(504); expect(data).toEqual({ error: expect.objectContaining({ @@ -4958,7 +4945,7 @@ describe("proxy route upstream selection", () => { waitPromise: queueWaitPromise, }); - const responsePromise = handleProxy( + const responsePromise = executeProxyRequest( new NextRequest("http://localhost/api/proxy/v1/messages", { method: "POST", signal: controller.signal, @@ -4971,7 +4958,7 @@ describe("proxy route upstream selection", () => { messages: [{ role: "user", content: "hi" }], }), }), - { params: Promise.resolve({ path: ["v1", "messages"] }) } + "v1/messages" ); await expect .poll(() => vi.mocked(upstreamQueueAdmission.enqueueWait).mock.calls.length)