diff --git a/docs/guide/architecture/request-lifecycle.md b/docs/guide/architecture/request-lifecycle.md index 6e46c85d..e058cd00 100644 --- a/docs/guide/architecture/request-lifecycle.md +++ b/docs/guide/architecture/request-lifecycle.md @@ -5,164 +5,159 @@ outline: deep # 请求生命周期 -这一页跟踪一次客户端请求从打到 AutoRouter 入口、到上游响应回到调用方手中的完整流程。所有引用都指向 `master` 分支上的源码与行号，可以照着读、照着改。示例以最常见的 `POST /api/proxy/v1/chat/completions` 为基准，其他协议（Anthropic `/v1/messages`、Gemini `/v1beta/models/:generateContent`、OpenAI `/v1/responses` 等）的差异在每一阶段单独标出。 +这一页跟踪一次客户端请求从进入 AutoRouter、完成鉴权与上游准入、发送到上游，再到响应、日志、计费和流量录制落地的完整流程。代理请求现在由三个边界清晰的模块协作：`src/app/api/proxy/v1/[...path]/route.ts` 只负责 HTTP 方法与参数适配，`proxy-request-lifecycle.ts` 的 `handleProxy` 负责生命周期编排，`proxy-execution.ts` 的 `forwardWithFailover` 负责候选选择、队列准入、上游调用、失败转移和资源释放。 + +示例以最常见的 `POST /api/proxy/v1/chat/completions` 为基准，其他协议（Anthropic `/v1/messages`、Gemini `/v1beta/models/:generateContent`、OpenAI `/v1/responses` 等）的差异在相应阶段标出。 ## 阶段一：HTTP 方法分发 入口文件：`src/app/api/proxy/v1/[...path]/route.ts`。 -文件末尾导出全部 5 个 HTTP 方法，每个方法只做一件事——把请求委托给同一个内部函数 `handleProxy`： - -| 导出 | 行号 | -| -------- | ---- | -| `GET` | 4140 | -| `POST` | 4147 | -| `PUT` | 4154 | -| `DELETE` | 4161 | -| `PATCH` | 4168 | +文件导出 `GET`、`POST`、`PUT`、`DELETE`、`PATCH` 五个 HTTP 方法。每个方法只把请求和动态路径参数委托给同一个生命周期入口： ```ts -// route.ts:4147 export async function POST(request: NextRequest, context: RouteContext) { return handleProxy(request, context); } ``` -`handleProxy` 自身从 `route.ts:2440` 开始，是后续所有阶段的容器函数。阅读源码时把它当成「主时序图」即可。 +`route.ts` 不再直接编排鉴权、路由、上游调用、日志、计费或 recording。阅读代理行为时，以 `proxy-request-lifecycle.ts` 的 `handleProxy` 为主时序，以 `proxy-execution.ts` 的 `forwardWithFailover` 为上游执行子流程。 ## 阶段二：CORS 与 OPTIONS -代理入口**没有**显式导出 `OPTIONS` handler，也没有独立的 `cors.ts` 工具文件。环境变量 `CORS_ORIGINS` 解析后保存在 `src/lib/utils/config.ts` 的 `corsOrigins` 字段中（默认 `["http://localhost:3000"]`），但当前**仅此一处引用**——全仓没有任何代码读取该字段后输出 `Access-Control-Allow-Origin` 或允许请求头等响应头（grep `Access-Control-Allow` 无匹配）。也就是说：`CORS_ORIGINS` 在当前实现里没有运行期效果，把某个 origin 加入该列表并不能让代理通过浏览器的 preflight。如果一定要让浏览器侧 SDK 直连代理，需要在代理前置一层反向代理（Nginx / Caddy / Traefik 等）由它来注入 CORS 头；典型部署中，代理仍被服务端调用方使用，浏览器不直接访问。 +代理入口没有显式导出 `OPTIONS` handler，也没有独立的 `cors.ts` 工具文件。环境变量 `CORS_ORIGINS` 会在 `src/lib/utils/config.ts` 中解析，但当前没有代码据此输出 `Access-Control-Allow-*` 响应头，因此把 origin 加入该列表不会改变代理的浏览器 preflight 行为。浏览器侧 SDK 如需直连，应在代理前置一层 Nginx、Caddy 或 Traefik，由前置层处理 CORS。 ## 阶段三：客户端鉴权 -提取客户端 Key 的函数：`extractProxyApiKey`，`route.ts:2255`。三种 header 按以下顺序判定，先命中先用： +客户端 Key 提取函数是 `proxy-request-lifecycle.ts` 的 `extractProxyApiKey`。三个 header 按以下顺序判定，先命中先用： -```ts -// route.ts:2253-2268（节选） -const fromAuthorization = extractApiKey(request.headers.get("authorization")); -if (fromAuthorization) return { keyValue: fromAuthorization, authSource: "authorization" }; +1. `Authorization`：兼容 `Bearer ` 与裸字符串。 +2. `x-api-key`：Anthropic SDK 的默认 header。 +3. `x-goog-api-key`：Gemini SDK 的默认 header。 -const fromApiKey = extractApiKey(request.headers.get("x-api-key")); -if (fromApiKey) return { keyValue: fromApiKey, authSource: "x-api-key" }; +提取后，`handleProxy` 按 key prefix 找候选记录并用 `verifyApiKey` 做 bcrypt 比对，再检查过期与用户状态。 -const fromGoogleApiKey = extractApiKey(request.headers.get("x-goog-api-key")); -if (fromGoogleApiKey) return { keyValue: fromGoogleApiKey, authSource: "x-goog-api-key" }; -``` +| 场景 | HTTP 响应 | 说明 | +| ------------------------ | ---------------------------------------- | ---------------------------- | +| 缺少支持的 Key header | 401 `{ "error": "Missing API key" }` | 不进入上游路由 | +| Key 不存在或 hash 不匹配 | 401 `{ "error": "Invalid API key" }` | 不进入上游路由 | +| Key 已过期 | 401 `{ "error": "API key has expired" }` | 不进入上游路由 | +| Key 所有者已停用 | 401 `{ "error": "API key is disabled" }` | ownerless key 不受此检查影响 | -`extractApiKey` 同时识别 `Bearer ` 与裸字符串两种写法。任意一种 header 都能通过，目的是兼容 OpenAI SDK（`Authorization: Bearer`）、Anthropic SDK（`x-api-key`）与 Gemini SDK（`x-goog-api-key`）的默认行为。 +这些早期鉴权错误保留简单的顶层 `error` 字符串格式，并通过 `logRejectedRequest` 写入拒绝日志；它们不产生上游请求、计费快照或 traffic fixture。 -提取到候选 Key 后，鉴权依次执行以下检查： +## 阶段四：路由能力、模型与 API Key 准入 -1. **存在性**（`route.ts:2452`）：`keyValue` 为空 → `{ "error": "Missing API key" }` HTTP 401。 -2. **bcrypt 比对**（`route.ts:2466`）：以 prefix 找出候选记录，调用 `verifyApiKey(keyValue, candidate.keyHash)`（内部 `bcrypt.compare`）。比对失败 → `{ "error": "Invalid API key" }` HTTP 401（`route.ts:2478`）。 -3. **过期判定**（`route.ts:2469`）：`candidate.expiresAt && candidate.expiresAt < new Date()` → `{ "error": "API key has expired" }` HTTP 401。 +鉴权通过后，`extractRequestContext` 从请求体和路径提取模型、session ID、stream 标志、reasoning effort 与 service tier。`resolveRouteCapability` 将 method、path 和 client profile 映射为 `RouteCapability`。 -注意这三类早期错误响应体里**只有一个 `error` 字符串字段**，没有 `code` 或 `error_code`，与后续路由阶段的统一错误格式不同。客户端如果要按机器可读规则区分原因，需要解析这个字符串本身。 +模型来源按协议不同： -## 阶段四：路由能力解析与模型提取 +- OpenAI / Anthropic：`bodyJson.model`。 +- Gemini：`extractGeminiModelFromPath(path)` 从 URL 中提取模型。 +- 最终模型：请求体模型优先，否则使用路径模型。 -`handleProxy` 在鉴权通过后立刻把请求映射为一个 `RouteCapability`，所有后续上游筛选都基于这个枚举值。 +没有模型时 AutoRouter 不会凭空拒绝请求；如果上游返回 400，该错误来自上游，而不是网关的统一错误层。 -**路径 → 能力映射**：`resolveRouteCapability(method, path, headers)`，`src/lib/services/route-capability-matcher.ts:307`。内部分两步： +在读取上游候选前，生命周期模块还会处理 API Key 维度的准入： -1. `matchProtocolFamily`（`route-capability-matcher.ts:171`）：按 URL 路径段匹配基础协议族，例如 `chat/completions` → `openai_chat_compatible`，`messages` → `anthropic_messages`，`responses` → `openai_responses`，`v1beta/models/:generateContent` → `gemini_native_generate`。 -2. `resolveFinalCapability`（`route-capability-matcher.ts:218`）：再结合请求头中的 client profile 做升级。例如 Claude Code CLI 的特征 header 会把 `anthropic_messages` 升级为 `claude_code_messages`，Codex CLI 会把 `openai_responses` 升级为 `codex_cli_responses`。 +- `API_KEY_RATE_LIMITED`：Key 的 RPM / TPM 限制已达到，返回 429，并带 `Retry-After`。 +- `API_KEY_QUOTA_EXCEEDED`：消费规则已超过额度，返回 429。 +- `API_KEY_MODEL_NOT_ALLOWED`：Key 不允许请求该模型，返回 403。 -`RouteCapability` 的全部取值定义在 `src/lib/route-capabilities.ts:1`： +这些拒绝都设置 `did_send_upstream: false`，并记录 `failure_stage: "auth_filter"` 或对应的准入阶段。 -``` -"anthropic_messages" | "claude_code_messages" | -"openai_responses" | "codex_cli_responses" | -"openai_chat_compatible" | "openai_extended" | -"gemini_native_generate" | "gemini_code_assist_internal" -``` +## 阶段五：候选过滤与上游选路 -**模型提取**：`extractRequestContext`，`route.ts:2396`。单次解析请求体，按协议族取值： +`handleProxy` 先读取活跃上游快照，再根据 Key 的 `accessMode` 构建候选集合： -- OpenAI / Anthropic：`bodyJson.model`（`route.ts:2414`）。 -- Gemini：`extractGeminiModelFromPath(path)`（`route.ts:2397`、`route-capability-matcher.ts:279`），从 URL 路径段 `v1beta/models/:generateContent` 中取出 ``。 -- 最终：`model = modelFromBody ?? modelFromPath`（`route.ts:2419`）。 +- `restricted`：只允许 `apiKeyUpstreams` 关联表中的上游。 +- `unrestricted`：允许所有活跃上游，但仍受 capability、model rule、健康和熔断状态限制。 -当请求体里 `bodyJson.model` 是 string 时直接采用，否则 `modelFromBody` 为 `null`（`route.ts:2414`）。当 `modelFromBody` 与 `modelFromPath` 都为 `null` 时，最终 `model` 字段也是 `null`，AutoRouter **不会**在本地拒绝该请求：`filterCandidatesByModelRules`（`route.ts:592`）在 `originalModel` 为 null 时直接返回全部候选（`route.ts:596-601`），请求仍会进入阶段五并被转发到选中的上游。若调用方因此收到 400，错误来自上游侧的响应，而非 AutoRouter 的统一错误层。 +候选集合随后经过： -## 阶段五：候选过滤与上游选路 +1. `filterCandidatesByModelRules`：按上游 `model_rules` 过滤模型。 +2. `filterByCircuitBreaker`：跳过 `OPEN` 或尚未到探测时间的 `HALF_OPEN` 上游。 +3. `selectFromUpstreamCandidates`：按 tier、权重、健康和 session affinity 选择候选。 +4. 转发前再次申请熔断器准入；期间变为 `OPEN` 的候选会被拒绝或触发失败转移。 -进入上游选路前要先确定候选集合。`handleProxy` 在 `route.ts:2634-2659` 附近做受限模式过滤： +如果路径不支持 capability、Key 没有授权上游或候选集合为空，请求在发送上游前结束。常见统一错误会包含 `request_id`、`reason`、`did_send_upstream` 和用户可读的 `user_hint`。 -```ts -// route.ts:2634-2659（节选） -const accessMode = validApiKey.accessMode ?? "restricted"; -const allowedUpstreamIds = - accessMode === "restricted" - ? storedAllowedUpstreamIds // 来自 apiKeyUpstreams 关联表 - : activeUpstreams.map((u) => u.id); // unrestricted: 全部活跃上游 -``` +## 阶段六：上游调用前的拒绝与资源释放 -`storedAllowedUpstreamIds` 来自 `apiKeyUpstreams` 表，是该客户端 Key 创建或编辑时绑定的上游集合。受限模式下未绑定的上游一律不可见；非受限模式下任何活跃上游都可被命中（具体能否承接当前请求，仍由路由能力与模型可用性进一步过滤）。 +这是代理生命周期的关键边界：只有 `forwardRequest` 真正开始调用上游后，才把请求视为已发送。所有此前的失败都必须满足： -接下来在候选内做选路。整套逻辑分为三层： +- 不发送上游请求。 +- 不写入成功计费快照。 +- 不写入 traffic recording fixture。 +- 写入正确的拒绝阶段、原因、耗时与队列状态。 -1. **熔断状态过滤**（`src/lib/services/load-balancer.ts:243`，`filterByCircuitBreaker`）： - - `OPEN` 状态且距离开启时间 `< openDuration` → 跳过（`load-balancer.ts:273-279`）。 - - `HALF_OPEN` 状态且距离上次探测 `< probeInterval` → 跳过（`load-balancer.ts:289-295`）。 - - 其余进入下一步。 -2. **模型匹配**（`route.ts filterCandidatesByModelRules`）：根据每个候选上游的 `model_rules` 决定是否承接当前模型，不匹配的上游加入排除列表。 -3. **加权随机选择**（`src/lib/services/load-balancer.ts:485`，`selectWeightedWithHealthScore`）：当前实现只用一种策略——加权随机叠加延时分数。有效权重 = `upstream.weight * latencyScore`，`latencyPenalty = min(latencyMs / 500, 0.5)`（`load-balancer.ts:496`）。当所有候选 `totalWeight == 0` 时退化为纯随机（`load-balancer.ts:510`）。 +### 常见拒绝结果 -选中候选后转发前再申请一次熔断器准入（`src/lib/services/circuit-breaker.ts:160`，`acquireCircuitBreakerPermit`）。若期间状态已切换到 `OPEN`，直接抛 `CircuitBreakerOpenError`（`circuit-breaker.ts:183`），由失败转移逻辑接住（见下一阶段）。 +| 阶段 / 原因 | HTTP 响应 | `did_send_upstream` | +| -------------------------------- | ------------------------------ | ------------------- | +| 无匹配 capability 或无可用候选 | 503 | `false` | +| Key 未授权任何可用上游 | 403 `NO_AUTHORIZED_UPSTREAMS` | `false` | +| 所有候选并发已满且未进入队列 | 503，reason `CONCURRENCY_FULL` | `false` | +| 等待队列超时 | 504 `QUEUE_WAIT_TIMEOUT` | `false` | +| 队列已满 | 503，队列拒绝 | `false` | +| 客户端在 dispatch 前取消 | 499 `CLIENT_DISCONNECTED` | `false` | +| 读取请求体或 dispatch 前准备失败 | 503 | `false` | -熔断器自身是个三态机：`CLOSED`（默认）/ `OPEN`（熔断中，拒绝新流量）/ `HALF_OPEN`（半开，按 `probeInterval` 节奏放探测请求）。状态枚举定义在 `circuit-breaker.ts:13-17`，状态持久化在 `circuitBreakerStates` 表中。状态机的完整行为详见 [`docs/circuit-breaker.md`](/circuit-breaker)。 +### 队列与并发生命周期 -## 阶段六：上游转发与流式包装 +`upstream-queue-admission.ts` 的 `UpstreamQueueAdmissionService` 管理每个上游的 active reservation 与等待队列： -转发函数：`forwardRequest(request, upstream, path, requestId, ...)`，`src/lib/services/proxy-client.ts:1004`。流程如下： +1. `enqueueWait` 创建等待项，并注册 timeout 和 abort listener。 +2. 等待成功后，释放的并发槽位通过 reservation handoff 交给排队请求，状态变为 `resumed`。 +3. 超时或客户端取消时，服务移除等待项，清理 timer 与 abort listener，并分别产生 `timed_out` 或 `aborted` 状态。 +4. dispatch 前的取消、重新选路失败和准备阶段异常通过 `releaseConnection` 释放已获得的槽位。 +5. `createReleaseConnectionOnce` 保证同一个上游 reservation 不会被重复释放。 -1. **header 处理**：调用 `filterHeaders`（`proxy-client.ts:234`）剔除 hop-by-hop header；调用 `injectAuthHeader`（`proxy-client.ts:255`）按上游配置注入正确的鉴权 header（部分上游用 `Authorization`、部分用 `x-api-key` 或 `x-goog-api-key`）。 -2. **发起请求**：通过 `fetch` 把改写后的请求体发到上游（`proxy-client.ts:1149`）。 -3. **响应类型判定**：上游响应若带 `content-type: text/event-stream`，进入 SSE 流式分支；否则按非流式整体回传。 +请求日志的 `routing_decision.queue` 会记录 `waiting`、`resumed`、`timed_out` 或 `aborted`，便于区分“没有容量”和“请求主动取消”。 -SSE 分支的处理（`proxy-client.ts:1185` 起）： +### 取消语义 -- `createSSETransformer`：把 chunk 解析为标准 `data: ...\n\n` 事件。 -- `stream.tee()`：分出两路，一路给客户端、一路给日志侧用于提取 token 计数与 TTFT。 -- `waitForFirstStreamContent`（`proxy-client.ts:1230`）：实现 first-byte 超时，避免上游长时间不吐第一块。 +`request.signal` 从入口一路传给 `forwardWithFailover` 和上游 `fetch`： -回到 `handleProxy`，给客户端的那一路再被包一层 `wrapStreamWithConnectionTracking`（`route.ts:1981`）： +- dispatch 前取消：停止选路 / 排队，不调用上游，返回 499，并释放相关 reservation。 +- dispatch 后取消：取消上游 fetch，释放当前连接并结束本次请求；不会把已经向客户端发送过响应头或 body 的流切换到另一条上游。 -- 每次 `read()` 与 `streamIdleTimeoutMs` 超时 promise 竞争（`route.ts:2004-2007`）。 -- `abortSignal.abort` 触发（典型场景：客户端关连接）时，调用 `reader.cancel` 并释放上游侧并发槽位（`route.ts:2038-2039`）。 -- 流正常完成后释放槽位（`route.ts:2063`），并 fire-and-forget 调 `markHealthy` 与 `recordSuccess` 通知健康与熔断模块（`route.ts:2072-2073`）。 +## 阶段七：上游转发与失败转移 -**失败转移分两类，行为不一样**： +`proxy-execution.ts` 的 `forwardWithFailover` 调用 `proxy-client.ts` 的 `forwardRequest`。转发过程包括： -- **首字节前的失败（可重试）**（`route.ts:1544` 起）：上游返回响应头时如果 `shouldTriggerFailover(result.statusCode, config)` 为真（典型：5xx、特定错误码、连接超时），记录此次失败、释放连接、调 `markUnhealthy` 与 `recordFailure`，向本次请求的 `failoverHistory` 数组追加一条记录（`route.ts:1559`），把当前上游加入「已失败」集合，`continue` 重新进入阶段五选下一条候选。当且仅当全部候选都失败时，才向调用方返回最终错误。这一阶段的重试对调用方完全无感。 -- **流开始后的中断（不可重试）**（`route.ts:1603-1667`）：一旦 `result.isStream === true`，函数直接 `return` 包装好的流给调用方（`route.ts:1657`），中途读流失败由 `wrapStreamWithConnectionTracking` 的回调（`route.ts:1618-1649`）交给 `settleStreamRuntimeFailureForCircuitBreaker` 处理——只更新日志、记录熔断失败、释放连接，**不会**回到阶段五选另一条上游接着吐 chunk。调用方此时看到的是一条提前结束的 SSE 流，需要自行处理「上游 stream 中断」这一错误。 +1. `filterHeaders` 移除 hop-by-hop 请求头。 +2. `injectAuthHeader` 按上游 provider 注入 `Authorization`、`x-api-key` 或 `x-goog-api-key`。 +3. `fetch` 使用下游 `AbortSignal` 发起上游请求。 +4. 非流式响应整体返回；SSE 响应经过 transformer 和流式 tracking 后返回。 -`failoverHistory` 数组在请求结束时随日志一起写入 `requestLogs.failoverHistory` 字段，可在管理后台「请求日志」详情页查看每一次尝试的 upstream_id、错误类型、状态码与时间戳。流式中断的失败记录入口不在这个数组，而是写入流式日志更新（阶段七的 `metricsPromise.then(...)` 路径）。 +首字节前的 5xx、连接错误和可配置失败规则可以触发 failover：当前上游会记录 `failoverHistory`、更新健康 / 熔断状态、释放 reservation，再回到阶段五选择下一条候选。 -## 阶段七：日志、计费、响应回写 +流已经开始后不再 failover。中途读取失败只更新日志、记录熔断失败并释放连接，客户端需要自行重新建立请求。 -**请求日志**：`src/lib/services/request-logger.ts`。 +请求体读取失败或 fetch 尚未真正开始时，`proxy-client.ts` 会通过 request metadata 标记 `fetchStarted: false`。生命周期模块据此把错误归类为 gateway rejection，而不是上游失败，不会误标记上游 unhealthy。 -- `logRequestStart`（`request-logger.ts:364`）：请求进入时**同步 await** 写入一行 `requestLogs`，token / latency 字段初始为 0，`statusCode` 字段初始为 null。 -- `updateRequestLog`（`request-logger.ts:412`）：请求结束或失败时 await 更新同一行（非流式路径在 `route.ts:3669` 与 `route.ts:4051`）。SSE 流式路径下，token 与 TTFT 在 `metricsPromise.then(...)` 内异步算完后再更新（`route.ts:3471`），失败用 `.catch` 兜底为 fire-and-forget。 -- `logRequest`（`request-logger.ts:504`）：无 `requestLogId` 时的兜底单次 INSERT，用于异常分支。 +## 阶段八：日志、计费、录制与响应回写 -**计费**：`src/lib/services/billing-cost-service.ts`。 +### 请求日志 -- 入口：`calculateAndPersistRequestBillingSnapshot`（`billing-cost-service.ts:431`），由 `route.ts:136` 的 `persistBillingSnapshotSafely` 封装做错误兜底。 -- 时机：日志写入后立即 **await**——非流式在 `route.ts:3739-3748`，流式在 `metricsPromise.then(...)` 内（`route.ts:3530-3545`）。 -- 写入：`requestBillingSnapshots` 表，使用 Drizzle 的 `onConflictDoUpdate`（`billing-cost-service.ts:118`）实现幂等 upsert，对同一 `request_log_id` 多次写入安全。 +- 在候选集合准备完成后，`logRequestStart` 创建 `in-progress` 日志，供管理后台实时展示。 +- 早于该节点发生的鉴权、Key 准入和 capability 拒绝，直接通过 `logRejectedRequest` 写入终态日志。 +- 后续失败或成功通过 `updateRequestLog` 更新同一行；没有可用的起始日志时，使用 `logRequest` 兜底插入。 +- 日志中的 `routing_decision` 会记录候选、排除原因、`failure_stage`、`actual_upstream_id`、`did_send_upstream` 和队列状态。 -**响应 header 回写**：`route.ts:3198` 用 `new Headers(result.headers)` 拷贝得到响应 header，但 `result.headers` 不是上游原始 header 的 1:1 副本，已经经过 `proxy-client.ts` 两道处理——`proxy-client.ts:1170-1173` 的 inline 循环按 `HOP_BY_HOP_HEADERS` 集合过滤上游响应头去掉 hop-by-hop 字段（与请求侧 `filterHeaders` 是两段不同代码）；当 undici 解压响应体时 `proxy-client.ts:1177-1179` 再删 `content-encoding` 与 `content-length`。SSE 分支额外强制写入 `Content-Type: text/event-stream`、`Cache-Control: no-cache`、`Connection: keep-alive`（`route.ts:3563-3565`）。代理层**不会**追加任何 AutoRouter 专属 header（既无 `X-AutoRouter-Request-Id`，也无 `X-AutoRouter-Upstream-Id`）。请求 ID 与命中上游 ID 只通过管理后台「请求日志」回查。 +### 计费与流量录制 -**统一错误格式**：路由阶段及之后的所有错误经 `src/lib/services/unified-error.ts` 包装，响应体形如 `{ error: { code, message, ... } }`，状态码与错误码的映射定义在 `unified-error.ts` 的 `STATUS_CODE_MAP`。注意阶段三的鉴权早期错误**不经过**这一层，格式更朴素（只有顶层 `error` 字段，无 `code`）。 +计费入口是 `billing-cost-service.ts` 的 `calculateAndPersistRequestBillingSnapshot`，以 request log ID 做幂等 upsert。录制入口是 `traffic-recorder.ts` 的 `recordTrafficFixture`。 -**流量录制**：`src/lib/services/traffic-recorder.ts`。 +两者的生命周期边界不同： -- 决策：`shouldRecordFixture(outcome, settings)`（`traffic-recorder.ts:158`）依据 `trafficRecordingSettings` 表的运行期配置（`enabled` + `mode`）判断当前请求是否录制。该开关现为 DB 运行期配置，详见 [`.env` 配置参考](../deployment/env-reference) 中的 RECORDER 章节。 -- 时机：鉴权通过后立即按需读入请求体快照（`route.ts:2491`，`recorderEnabled ? await readRequestBody(request) : null`）；响应完成后在日志写入后 `void recordTrafficFixture(...).catch(...)` 异步落盘（`route.ts:3802` 与 `route.ts:4040`），错误不阻塞调用方响应。 +- `did_send_upstream === false`：不写计费快照，不构建或写入成功 / 失败 fixture。 +- `did_send_upstream === true`：成功请求按正常路径计费并按设置录制；已发送上游但最终失败的请求可以写失败计费快照和 failure fixture，用于真实故障排查。 + +### 响应格式 + +路由阶段及之后的错误由 `src/lib/services/unified-error.ts` 统一映射为 `{ error: { code, message, ... } }`。鉴权早期错误保留阶段三的简单 `{ error: string }` 兼容格式。非流式成功响应返回上游 body；SSE 响应额外设置 `Content-Type: text/event-stream`、`Cache-Control: no-cache` 与 `Connection: keep-alive`。 ## 时序总览 @@ -170,44 +165,39 @@ SSE 分支的处理（`proxy-client.ts:1185` 起）： 客户端 │ POST /api/proxy/v1/chat/completions ▼ -[1] 方法分发 ──────────► handleProxy（route.ts:2440） +[1] route.ts 方法适配 + ▼ +[2] CORS / OPTIONS（当前没有自定义 preflight handler） ▼ -[2] CORS / OPTIONS（无自定义 handler；CORS_ORIGINS 当前无运行期效果） +[3] handleProxy 鉴权 + ├ 缺失 / 无效 / 过期 / disabled key → 401 + └ 记录拒绝日志，不访问上游 ▼ -[3] 鉴权 - ├ 缺 key → 401 { error: "Missing API key" } - ├ bcrypt 失败 → 401 { error: "Invalid API key" } - └ 已过期 → 401 { error: "API key has expired" } +[4] capability + model + Key 准入 + ├ rate limit / quota / model permission → 429 / 403 + └ did_send_upstream = false ▼ -[4] 路由能力 + 模型解析 - route-capability-matcher.ts → RouteCapability - bodyJson.model 或 URL 路径 +[5] 候选过滤 + 熔断 + 并发选路 + ├ 无候选 / 未授权 / 并发满 → 403 / 503 + └ queue admission → waiting / resumed / timed_out / aborted ▼ -[5] 候选过滤 + 选路 - 受限模式 → apiKeyUpstreams 过滤 - 熔断状态 → filterByCircuitBreaker - 模型匹配 → filterCandidatesByModelRules - 加权随机 → selectWeightedWithHealthScore - 申请准入 → acquireCircuitBreakerPermit（OPEN 抛 CircuitBreakerOpenError） +[6] 上游调用前拒绝与资源释放 + ├ queue / reservation / abort listener cleanup + └ 不计费、不录制、不发送上游 ▼ -[6] 转发 - proxy-client.forwardRequest → 上游 - SSE → tee + wrapStreamWithConnectionTracking - 失败 → 记 failoverHistory，回到 [5] 选下一条 +[7] forwardWithFailover → proxy-client.forwardRequest + ├ 成功 → 非流式响应或 SSE + └ 首字节前失败 → 记录 failoverHistory 后回到 [5] ▼ -[7] 日志 / 计费 / 响应 - requestLogs 更新 - requestBillingSnapshots upsert - 上游 header 透传 + SSE 强制写三个标准头 - traffic-recorder fire-and-forget +[8] request log / billing / recording / response ▼ -客户端 ← 2xx 响应体（与上游一致）或统一错误格式 +客户端 ← 2xx 响应体、SSE 或统一错误响应 ``` ## 不在本页范围内 - 客户端 Key 的创建与可见性配置：见 [创建客户端 API Key](../usage/client-keys)。 -- 上游配置字段与能力声明：见 [添加第一个上游](../usage/first-upstream)。 +- 上游配置字段与 capability 声明：见 [添加第一个上游](../usage/first-upstream)。 - 各类 SDK 调用样例：见 [通过 AutoRouter 调用模型](../usage/invoke-models)。 - 熔断器与失败转移的状态机细节：见 [`docs/circuit-breaker.md`](/circuit-breaker)。 -- 模型路由规则与多上游同模型的调度细节：后续「模型路由规则」「负载均衡与权重」专题文档。 +- 请求日志筛选与统计查询：见 [请求日志与统计](../usage/logs-stats)。 diff --git a/src/app/api/proxy/v1/[...path]/proxy-execution.ts b/src/app/api/proxy/v1/[...path]/proxy-execution.ts index 2d17c504..3aee6bf4 100644 --- a/src/app/api/proxy/v1/[...path]/proxy-execution.ts +++ b/src/app/api/proxy/v1/[...path]/proxy-execution.ts @@ -6,6 +6,7 @@ import { UpstreamEmptyResponseError, UpstreamNoContentStreamError, type CompensationHeader, + getProxyRequestErrorMetadata, type HeaderDiff, type ProxyResult, } from "@/lib/services/proxy-client"; @@ -110,6 +111,7 @@ export interface FailoverErrorWithHistory extends Error { failoverHistory?: FailoverAttempt[]; concurrencyExcludedCandidates?: RoutingExcluded[]; didSendUpstream?: boolean; + lastDispatchedFailoverAttempt?: FailoverAttempt; headerDiff?: HeaderDiff | null; queue?: RoutingQueueLog | null; } @@ -117,6 +119,7 @@ export interface FailoverErrorWithHistory extends Error { interface FailoverContext { failoverHistory: FailoverAttempt[]; didSendUpstream: boolean; + lastDispatchedFailoverAttempt?: FailoverAttempt; concurrencyExcludedCandidates: RoutingExcluded[]; headerDiff?: HeaderDiff | null; queue?: RoutingQueueLog | null; @@ -130,6 +133,7 @@ function attachFailoverContext( enrichedError.failoverHistory = [...context.failoverHistory]; enrichedError.concurrencyExcludedCandidates = [...context.concurrencyExcludedCandidates]; enrichedError.didSendUpstream = context.didSendUpstream; + enrichedError.lastDispatchedFailoverAttempt = context.lastDispatchedFailoverAttempt; enrichedError.headerDiff = context.headerDiff ?? null; enrichedError.queue = context.queue ?? enrichedError.queue ?? null; return enrichedError; @@ -149,22 +153,6 @@ export function withQueueStreamFlag( }; } -function extractHeaderDiffFromError(error: unknown): HeaderDiff | null { - if (!error || typeof error !== "object" || !("headerDiff" in error)) { - return null; - } - const candidate = (error as { headerDiff?: unknown }).headerDiff; - if (!candidate || typeof candidate !== "object") { - return null; - } - - return candidate as HeaderDiff; -} - -function isSyntheticFailoverAttempt(attempt: FailoverAttempt): boolean { - return attempt.error_type === "concurrency_full"; -} - const CIRCUIT_BREAKER_NEUTRAL_PATHS = new Set(["messages/count_tokens"]); function normalizeCircuitBreakerPath(path: string): string { @@ -177,18 +165,6 @@ function shouldRecordCircuitBreakerFailure(path: string): boolean { return !CIRCUIT_BREAKER_NEUTRAL_PATHS.has(normalizedPath); } -export function getLastSentFailoverAttempt( - failoverHistory: FailoverAttempt[] -): FailoverAttempt | undefined { - for (let index = failoverHistory.length - 1; index >= 0; index -= 1) { - const attempt = failoverHistory[index]; - if (!isSyntheticFailoverAttempt(attempt)) { - return attempt; - } - } - return undefined; -} - /** * Determine error type for failover logging. */ @@ -291,11 +267,22 @@ export function resolveFailureStage( if (isNoAuthorizedUpstreamsError(error)) { return "auth_filter"; } + if (isQueueWaitTimeoutError(error) || isQueueWaitAbortedError(error)) { + return "candidate_selection"; + } if (isDownstreamStreamingError(error)) { return "downstream_streaming"; } - if (error instanceof ClientDisconnectedError && didSendUpstream) { - return "downstream_streaming"; + if (error instanceof ClientDisconnectedError) { + if (error.failureStage) { + return error.failureStage; + } + if (error.queue?.status === "aborted") { + return "candidate_selection"; + } + if (didSendUpstream) { + return "downstream_streaming"; + } } if (!didSendUpstream) { return "candidate_selection"; @@ -349,18 +336,9 @@ export function getCircuitBlockedCandidates(error: unknown): CircuitBlockedCandi } export function resolveDidSendUpstream( - error: FailoverErrorWithHistory | null | undefined, - lastSentFailoverAttempt: FailoverAttempt | undefined + error: FailoverErrorWithHistory | null | undefined ): boolean { - const attachedDidSend = - typeof error?.didSendUpstream === "boolean" ? error.didSendUpstream : undefined; - - // Prefer positive evidence: if any non-synthetic failover attempt exists, upstream was sent. - if (lastSentFailoverAttempt != null) { - return true; - } - - return attachedDidSend === true; + return error?.didSendUpstream === true; } export function getUserHint( @@ -787,10 +765,35 @@ export async function forwardWithFailover( const concurrencyExcludedCandidates: RoutingExcluded[] = []; let lastError: Error | null = null; let didSendUpstream = false; + let lastDispatchedFailoverAttempt: FailoverAttempt | undefined; let affinityHit = false; let affinityMigrated = false; let finalSelectionReason: RoutingSelectionReason | null = null; let queueLifecycle: RoutingQueueLog | null = null; + let queueReservationPending = false; + let didDispatchCurrentAttempt = false; + const markQueueAborted = () => { + if (!queueReservationPending || !queueLifecycle || queueLifecycle.status === "aborted") { + return; + } + queueLifecycle = { + ...queueLifecycle, + status: "aborted", + resumed_at: null, + }; + }; + + const attachCurrentFailoverContext = ( + error: T, + queue: RoutingQueueLog | null = queueLifecycle + ): T & FailoverErrorWithHistory => + attachFailoverContext(error, { + failoverHistory, + didSendUpstream, + lastDispatchedFailoverAttempt, + concurrencyExcludedCandidates, + queue, + }); const circuitBlockedCandidates: CircuitBlockedCandidate[] = []; @@ -818,23 +821,45 @@ export async function forwardWithFailover( } } }; - - // Clone the request body once for potential retries - const requestClone = request.clone(); - const requestBodyBuffer = await requestClone.arrayBuffer(); + let requestBodyBuffer: ArrayBuffer; + try { + requestBodyBuffer = await awaitBeforeDispatch( + request.signal, + async () => { + const requestClone = request.clone(); + return requestClone.arrayBuffer(); + }, + () => {} + ); + } catch (error) { + const bodyReadError = + error instanceof Error ? error : new Error(typeof error === "string" ? error : String(error)); + const clientDisconnected = request.signal.aborted + ? new ClientDisconnectedError( + "Client disconnected before upstream dispatch", + "candidate_selection" + ) + : bodyReadError; + throw attachCurrentFailoverContext(clientDisconnected); + } // Loop until we succeed, exhaust all upstreams, or hit max attempts let attemptCount = 0; while (true) { + didDispatchCurrentAttempt = false; // Check if downstream client has disconnected if (request.signal.aborted) { + markQueueAborted(); + queueReservationPending = false; log.warn({ requestId }, "client disconnected during failover, stopping retries"); throw attachFailoverContext( - new ClientDisconnectedError("Client disconnected during failover"), + new ClientDisconnectedError("Client disconnected during failover", "candidate_selection"), { failoverHistory, didSendUpstream, + lastDispatchedFailoverAttempt, concurrencyExcludedCandidates, + queue: queueLifecycle, } ); } @@ -875,6 +900,7 @@ export async function forwardWithFailover( throw attachFailoverContext(error instanceof Error ? error : new Error(String(error)), { failoverHistory, didSendUpstream, + lastDispatchedFailoverAttempt, concurrencyExcludedCandidates, }); } @@ -895,7 +921,12 @@ export async function forwardWithFailover( selectedUpstream = resumedSelection.selectedUpstream; finalSelectionReason = resumedSelection.selectionReason; queueLifecycle = resumedSelection.queue; + queueReservationPending = queueLifecycle?.status === "resumed"; } catch (resumeError) { + const resumedQueue = (resumeError as FailoverErrorWithHistory | null)?.queue ?? null; + if (resumedQueue) { + queueLifecycle = resumedQueue; + } if (resumeError instanceof AllCandidatesConcurrencyFullError) { appendConcurrencyExclusions(resumeError.excludedCandidates); lastError = resumeError; @@ -908,6 +939,7 @@ export async function forwardWithFailover( throw attachFailoverContext(resumeError, { failoverHistory, didSendUpstream, + lastDispatchedFailoverAttempt, concurrencyExcludedCandidates, queue: (resumeError as FailoverErrorWithHistory).queue ?? null, }); @@ -917,6 +949,7 @@ export async function forwardWithFailover( { failoverHistory, didSendUpstream, + lastDispatchedFailoverAttempt, concurrencyExcludedCandidates, queue: (resumeError as FailoverErrorWithHistory).queue ?? null, } @@ -934,14 +967,38 @@ export async function forwardWithFailover( throw attachFailoverContext(error instanceof Error ? error : new Error(String(error)), { failoverHistory, didSendUpstream, + lastDispatchedFailoverAttempt, concurrencyExcludedCandidates, queue: queueLifecycle, }); } } + if (request.signal.aborted) { + markQueueAborted(); + queueReservationPending = false; + if (selectedUpstream) { + releaseConnection(selectedUpstream.id); + } + throw attachFailoverContext( + new ClientDisconnectedError( + "Client disconnected before upstream dispatch", + "candidate_selection" + ), + { + failoverHistory, + didSendUpstream, + lastDispatchedFailoverAttempt, + concurrencyExcludedCandidates, + queue: queueLifecycle, + } + ); + } - // Check if we should continue trying if (!shouldContinueFailover(attemptCount, hasMoreUpstreams, config, request.signal.aborted)) { + queueReservationPending = false; + if (selectedUpstream) { + releaseConnection(selectedUpstream.id); + } // No more upstreams or hit max attempts - throw NoHealthyUpstreamsError // to indicate all failover attempts have been exhausted const exhaustedError = new NoHealthyUpstreamsError( @@ -953,7 +1010,9 @@ export async function forwardWithFailover( throw attachFailoverContext(exhaustedError, { failoverHistory, didSendUpstream, + lastDispatchedFailoverAttempt, concurrencyExcludedCandidates, + queue: queueLifecycle, }); } @@ -961,7 +1020,9 @@ export async function forwardWithFailover( throw attachFailoverContext(new NoHealthyUpstreamsError("No upstream available"), { failoverHistory, didSendUpstream, + lastDispatchedFailoverAttempt, concurrencyExcludedCandidates, + queue: queueLifecycle, }); } @@ -969,23 +1030,45 @@ export async function forwardWithFailover( let attemptUpstreamBaseUrl = selectedUpstream.baseUrl; const releaseSelectedConnectionOnce = createReleaseConnectionOnce(selectedUpstream.id); + const dispatchUpstream = selectedUpstream; + const didSendBeforeAttempt: boolean = didSendUpstream; + let fetchStartedForAttempt = false; + let forwardRequestCalled = false; + const markDispatchStarted = () => { + fetchStartedForAttempt = true; + didDispatchCurrentAttempt = true; + didSendUpstream = true; + queueReservationPending = false; + onDispatchStart?.(dispatchUpstream); + }; try { - // Create a new request with the buffered body + // Preserve the downstream signal so the outbound fetch is cancelled with the client. const proxyRequest = new Request(request.url, { method: request.method, headers: request.headers, body: requestBodyBuffer.byteLength > 0 ? requestBodyBuffer : undefined, + signal: request.signal, }); - const circuitBreakerConfig = await getEffectiveCircuitBreakerConfig(selectedUpstream.id); + const circuitBreakerConfig = await awaitBeforeDispatch( + request.signal, + () => getEffectiveCircuitBreakerConfig(selectedUpstream.id), + releaseSelectedConnectionOnce + ); const upstreamForProxy = prepareUpstreamForProxy(selectedUpstream, { firstByteTimeout: circuitBreakerConfig.firstByteTimeout, streamIdleTimeout: circuitBreakerConfig.streamIdleTimeout, }); attemptUpstreamBaseUrl = upstreamForProxy.baseUrl; - didSendUpstream = true; + // A cancellation observed before dispatch must never be reported as an upstream attempt. + if (request.signal.aborted) { + throw new ClientDisconnectedError( + "Client disconnected before upstream dispatch", + "candidate_selection" + ); + } // CLIProxyAPI 单账号映射上游：按绑定账号的前缀拼出携带前缀的模型名注入转发， // 使 CLIProxyAPI 把请求固定路由到该账号。普通上游与池上游不带账号文件名，跳过注入。 let cliproxyModelOverride: string | undefined; @@ -994,25 +1077,40 @@ export async function forwardWithFailover( selectedUpstream.cliproxyInstanceId && requestModel ) { - const accountPrefix = await resolveCliproxyAccountPrefix( - selectedUpstream.cliproxyInstanceId, - selectedUpstream.cliproxyAuthFileName + const accountPrefix = await awaitBeforeDispatch( + request.signal, + () => + resolveCliproxyAccountPrefix( + selectedUpstream.cliproxyInstanceId!, + selectedUpstream.cliproxyAuthFileName! + ), + releaseSelectedConnectionOnce ); if (accountPrefix) { cliproxyModelOverride = buildCliproxyPrefixedModel(accountPrefix, requestModel); } } - onDispatchStart?.(selectedUpstream); + if (request.signal.aborted) { + throw new ClientDisconnectedError( + "Client disconnected before upstream dispatch", + "candidate_selection" + ); + } + forwardRequestCalled = true; const result = await forwardRequest( proxyRequest, upstreamForProxy, path, requestId, compensationHeaders, - cliproxyModelOverride + cliproxyModelOverride, + markDispatchStarted ); - + // A response proves dispatch for legacy/test forwarders that do not invoke the seam callback. + if (!fetchStartedForAttempt) { + markDispatchStarted(); + } // Check if response indicates we should failover if (shouldTriggerFailover(result.statusCode, config)) { const failedResponse = await captureFailedResponse(result); @@ -1035,25 +1133,25 @@ export async function forwardWithFailover( void recordFailure(selectedUpstream.id, `http_${result.statusCode}`); } // Record failover attempt - failoverHistory.push( - buildFailoverAttempt( - selectedUpstream, - routeCapability, - attemptUpstreamBaseUrl, - finalSelectionReason, - { - errorType, - errorMessage: resolveFailedResponseErrorMessage(result.statusCode, failedResponse), - statusCode: result.statusCode, - responseHeaders: failedResponse.headers, - responseBodyText: failedResponse.bodyText, - responseBodyJson: failedResponse.bodyJson, - headerDiff: result.headerDiff, - circuitBreakerRecorded, - matchedFailureRule, - } - ) + const failoverAttempt = buildFailoverAttempt( + selectedUpstream, + routeCapability, + attemptUpstreamBaseUrl, + finalSelectionReason, + { + errorType, + errorMessage: resolveFailedResponseErrorMessage(result.statusCode, failedResponse), + statusCode: result.statusCode, + responseHeaders: failedResponse.headers, + responseBodyText: failedResponse.bodyText, + responseBodyJson: failedResponse.bodyJson, + headerDiff: result.headerDiff, + circuitBreakerRecorded, + matchedFailureRule, + } ); + failoverHistory.push(failoverAttempt); + lastDispatchedFailoverAttempt = failoverAttempt; failedUpstreamIds.push(selectedUpstream.id); lastError = new Error(`Upstream returned ${result.statusCode}`); continue; @@ -1156,24 +1254,46 @@ export async function forwardWithFailover( } catch (error) { // Release connection on error releaseSelectedConnectionOnce(); - const errorHeaderDiff = extractHeaderDiffFromError(error); + const errorMetadata = getProxyRequestErrorMetadata(error); + const errorHeaderDiff = errorMetadata?.headerDiff ?? null; + const fetchStartedFromError = errorMetadata?.fetchStarted; + if (forwardRequestCalled && !fetchStartedForAttempt && fetchStartedFromError !== false) { + markDispatchStarted(); + } + if (fetchStartedFromError === false) { + didDispatchCurrentAttempt = false; + didSendUpstream = didSendBeforeAttempt; + } // Check if client disconnected if (request.signal.aborted) { + if (!didDispatchCurrentAttempt) { + markQueueAborted(); + } + queueReservationPending = false; log.warn({ requestId }, "client disconnected during request, stopping"); - throw attachFailoverContext( - new ClientDisconnectedError("Client disconnected during request"), - { - failoverHistory, - didSendUpstream, - concurrencyExcludedCandidates, - queue: queueLifecycle, - } - ); + const disconnectError = + error instanceof ClientDisconnectedError || isQueueWaitAbortedError(error) + ? error + : new ClientDisconnectedError( + "Client disconnected during request", + didDispatchCurrentAttempt ? "downstream_streaming" : "candidate_selection" + ); + throw attachFailoverContext(disconnectError, { + failoverHistory, + didSendUpstream, + lastDispatchedFailoverAttempt, + concurrencyExcludedCandidates, + queue: queueLifecycle, + }); } + queueReservationPending = false; - // Record failure in circuit breaker for failoverable errors - if (isFailoverableError(error) || error instanceof CircuitBreakerOpenError) { + // Setup/body failures before the fetch starts are gateway rejections, not upstream failures. + if ( + didDispatchCurrentAttempt && + (isFailoverableError(error) || error instanceof CircuitBreakerOpenError) + ) { const errorEvidence = extractFailoverErrorEvidence(error); const errorType = getErrorType( error instanceof Error ? error : null, @@ -1197,25 +1317,27 @@ export async function forwardWithFailover( const errorMessage = errorEvidence.errorMessage; void markUnhealthy(selectedUpstream.id, errorMessage); // Record failover attempt - failoverHistory.push( - buildFailoverAttempt( - selectedUpstream, - routeCapability, - attemptUpstreamBaseUrl, - finalSelectionReason, - { - errorType, - errorMessage, - statusCode: errorEvidence.statusCode, - responseHeaders: errorEvidence.responseHeaders, - responseBodyText: errorEvidence.responseBodyText, - responseBodyJson: errorEvidence.responseBodyJson, - headerDiff: errorHeaderDiff, - circuitBreakerRecorded, - matchedFailureRule, - } - ) + const failoverAttempt = buildFailoverAttempt( + selectedUpstream, + routeCapability, + attemptUpstreamBaseUrl, + finalSelectionReason, + { + errorType, + errorMessage, + statusCode: errorEvidence.statusCode, + responseHeaders: errorEvidence.responseHeaders, + responseBodyText: errorEvidence.responseBodyText, + responseBodyJson: errorEvidence.responseBodyJson, + headerDiff: errorHeaderDiff, + circuitBreakerRecorded, + matchedFailureRule, + } ); + failoverHistory.push(failoverAttempt); + if (didDispatchCurrentAttempt) { + lastDispatchedFailoverAttempt = failoverAttempt; + } failedUpstreamIds.push(selectedUpstream.id); lastError = error instanceof Error ? error : new Error(String(error)); continue; @@ -1228,6 +1350,7 @@ export async function forwardWithFailover( throw attachFailoverContext(nonFailoverError, { failoverHistory, didSendUpstream, + lastDispatchedFailoverAttempt, concurrencyExcludedCandidates, headerDiff: errorHeaderDiff, queue: queueLifecycle, @@ -1235,6 +1358,69 @@ export async function forwardWithFailover( } } } +async function awaitBeforeDispatch( + signal: AbortSignal, + operation: () => Promise, + onAbort: () => void +): Promise { + return new Promise((resolve, reject) => { + let settled = false; + + const cleanup = () => { + signal.removeEventListener("abort", abortHandler); + }; + const abortHandler = () => { + if (settled) { + return; + } + settled = true; + cleanup(); + onAbort(); + reject( + new ClientDisconnectedError( + "Client disconnected before upstream dispatch", + "candidate_selection" + ) + ); + }; + + if (signal.aborted) { + abortHandler(); + return; + } + + signal.addEventListener("abort", abortHandler, { once: true }); + + let operationPromise: Promise; + try { + operationPromise = Promise.resolve(operation()); + } catch (error) { + settled = true; + cleanup(); + reject(error); + return; + } + + operationPromise.then( + (value) => { + if (settled) { + return; + } + settled = true; + cleanup(); + resolve(value); + }, + (error) => { + if (settled) { + return; + } + settled = true; + cleanup(); + reject(error); + } + ); + }); +} function createReleaseConnectionOnce(upstreamId: string): () => void { let released = false; @@ -1313,19 +1499,62 @@ async function resumeQueuedUpstreamSelection(options: { throw new AllCandidatesConcurrencyFullError([], waitableCandidate); } + let handoffReservationGranted = false; + let ownsHandoffReservation = true; + let handoffAbortListener: (() => void) | null = null; + const cleanupHandoffAbortListener = () => { + if (handoffAbortListener) { + request.signal.removeEventListener("abort", handoffAbortListener); + handoffAbortListener = null; + } + }; + const releaseHandoffReservation = () => { + if (!handoffReservationGranted) { + cleanupHandoffAbortListener(); + return; + } + if (!ownsHandoffReservation) { + cleanupHandoffAbortListener(); + return; + } + ownsHandoffReservation = false; + cleanupHandoffAbortListener(); + releaseConnection(waitableCandidate.upstream.id); + }; + handoffAbortListener = () => { + releaseHandoffReservation(); + }; + request.signal.addEventListener("abort", handoffAbortListener, { once: true }); + const waitGrantPromise = queued.waitPromise.then((grant) => { + handoffReservationGranted = true; + if (request.signal.aborted) { + releaseHandoffReservation(); + } + return grant; + }); + void waitGrantPromise.catch(() => {}); if (onQueueStateChange) { - void Promise.resolve(onQueueStateChange(waitingQueue)).catch((error) => + try { + const queueStateChange = onQueueStateChange(waitingQueue); + void Promise.resolve(queueStateChange).catch((error) => + log.error( + { err: error, requestId, upstreamId: waitableCandidate.upstream.id }, + "failed to persist queue waiting state" + ) + ); + } catch (error) { log.error( { err: error, requestId, upstreamId: waitableCandidate.upstream.id }, "failed to persist queue waiting state" - ) - ); + ); + } } let waitGrant: Awaited; try { - waitGrant = await queued.waitPromise; + waitGrant = await waitGrantPromise; } catch (error) { + cleanupHandoffAbortListener(); if (isQueueWaitTimeoutError(error)) { (error as FailoverErrorWithHistory).queue = { ...waitingQueue, @@ -1344,59 +1573,116 @@ async function resumeQueuedUpstreamSelection(options: { } throw error; } + const resumedQueue = (): RoutingQueueLog => ({ + ...waitingQueue, + status: "resumed", + resumed_at: new Date().toISOString(), + wait_duration_ms: waitGrant.waitDurationMs, + }); + const abortedQueue = (): RoutingQueueLog => ({ + ...waitingQueue, + status: "aborted", + wait_duration_ms: waitGrant.waitDurationMs, + }); + const createAbortedQueueError = (message: string): ClientDisconnectedError => { + const abortError = new ClientDisconnectedError(message, "candidate_selection"); + abortError.queue = abortedQueue(); + return abortError; + }; + + try { + if (request.signal.aborted) { + throw createAbortedQueueError("Client disconnected after queue admission"); + } - const refreshedCandidateSnapshot = await loadActiveUpstreamSnapshot(); + const refreshedCandidateSnapshot = await awaitBeforeDispatch( + request.signal, + () => loadActiveUpstreamSnapshot(), + releaseHandoffReservation + ); + if (request.signal.aborted) { + throw createAbortedQueueError("Client disconnected while refreshing queued upstream"); + } - const excludeIds = failedUpstreamIds.length > 0 ? failedUpstreamIds : undefined; - const resumeDecision = await decideQueuedUpstreamResume( - waitableCandidate.upstream.id, - candidateUpstreamIds, - excludeIds, - { candidateSnapshot: refreshedCandidateSnapshot } - ); + const excludeIds = failedUpstreamIds.length > 0 ? failedUpstreamIds : undefined; + const resumeDecision = await awaitBeforeDispatch( + request.signal, + () => + decideQueuedUpstreamResume( + waitableCandidate.upstream.id, + candidateUpstreamIds, + excludeIds, + { candidateSnapshot: refreshedCandidateSnapshot } + ), + releaseHandoffReservation + ); + + if (resumeDecision.action === "resume" && resumeDecision.upstream) { + ownsHandoffReservation = false; + cleanupHandoffAbortListener(); + return { + selectedUpstream: resumeDecision.upstream, + selectionReason: attachRetryReason(null, failoverHistory), + concurrencyExcludedCandidates: [], + queue: resumedQueue(), + }; + } + + releaseHandoffReservation(); + if (request.signal.aborted) { + throw createAbortedQueueError("Client disconnected before queued reselection"); + } + + const reselection = await reselectQueuedUpstreamOnce( + waitableCandidate.upstream.id, + candidateUpstreamIds, + resumeDecision.excludeIds, + { candidateSnapshot: refreshedCandidateSnapshot } + ); + + if (request.signal.aborted) { + releaseConnection(reselection.upstream.id); + throw createAbortedQueueError("Client disconnected after queued reselection"); + } - if (resumeDecision.action === "resume" && resumeDecision.upstream) { return { - selectedUpstream: resumeDecision.upstream, - selectionReason: attachRetryReason(null, failoverHistory), - concurrencyExcludedCandidates: [], - queue: { - ...waitingQueue, - status: "resumed", - resumed_at: new Date().toISOString(), - wait_duration_ms: waitGrant.waitDurationMs, - }, + selectedUpstream: reselection.upstream, + selectionReason: attachRetryReason(reselection.selectionReason ?? null, failoverHistory), + concurrencyExcludedCandidates: reselection.concurrencyExcluded ?? [], + queue: resumedQueue(), }; + } catch (error) { + releaseHandoffReservation(); + if (request.signal.aborted && !(error instanceof ClientDisconnectedError)) { + const abortError = new ClientDisconnectedError( + "Client disconnected during queued reselection", + "candidate_selection" + ); + abortError.queue = abortedQueue(); + throw abortError; + } + if (error instanceof ClientDisconnectedError) { + error.queue ??= abortedQueue(); + throw error; + } + if (error instanceof Error) { + (error as FailoverErrorWithHistory).queue ??= resumedQueue(); + } + throw error; } - - releaseConnection(waitableCandidate.upstream.id); - const reselection = await reselectQueuedUpstreamOnce( - waitableCandidate.upstream.id, - candidateUpstreamIds, - resumeDecision.excludeIds, - { candidateSnapshot: refreshedCandidateSnapshot } - ); - - return { - selectedUpstream: reselection.upstream, - selectionReason: attachRetryReason(reselection.selectionReason ?? null, failoverHistory), - concurrencyExcludedCandidates: reselection.concurrencyExcluded ?? [], - queue: { - ...waitingQueue, - status: "resumed", - resumed_at: new Date().toISOString(), - wait_duration_ms: waitGrant.waitDurationMs, - }, - }; } /** * Error thrown when downstream client disconnects. */ export class ClientDisconnectedError extends Error { - constructor(message: string) { + queue?: RoutingQueueLog | null; + failureStage?: RoutingFailureStage; + + constructor(message: string, failureStage?: RoutingFailureStage) { super(message); this.name = "ClientDisconnectedError"; + this.failureStage = failureStage; } } 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 fd0831e5..8c07ce7f 100644 --- a/src/app/api/proxy/v1/[...path]/proxy-request-lifecycle.ts +++ b/src/app/api/proxy/v1/[...path]/proxy-request-lifecycle.ts @@ -3,6 +3,7 @@ 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"; import { + isStreamRequest, prepareUpstreamForProxy, filterHeaders, injectAuthHeader, @@ -50,6 +51,7 @@ import { type UnifiedErrorReason, } from "@/lib/services/unified-error"; import type { + RequestThinkingConfig, ReasoningEffort, RequestedServiceTier, EffectiveServiceTier, @@ -80,7 +82,6 @@ import { forwardWithFailover, ClientDisconnectedError, getCircuitBlockedCandidates, - getLastSentFailoverAttempt, getUserHint, isNoAuthorizedUpstreamsError, isQueueWaitAbortedError, @@ -175,6 +176,98 @@ async function shouldRejectExceededApiKeyQuotaBeforeProxy(input: { return false; } } +function buildRoutingDecisionLog(input: { + model: string | null; + matchedRouteCapability: RouteCapability | null; + routeMatchSource: RouteMatchSource | null; + failureStage: RoutingFailureStage | null; + providerType?: string | null; +}): RoutingDecisionLog { + return { + original_model: input.model ?? "(path-based)", + resolved_model: input.model ?? "(path-based)", + model_redirect_applied: false, + provider_type: + input.providerType !== undefined + ? input.providerType + : input.matchedRouteCapability + ? getProviderByRouteCapability(input.matchedRouteCapability) + : null, + routing_type: "none", + matched_route_capability: input.matchedRouteCapability, + route_match_source: input.routeMatchSource, + capability_candidates_count: 0, + candidates: [], + excluded: [], + candidate_count: 0, + final_candidate_count: 0, + selected_upstream_id: null, + candidate_upstream_id: null, + actual_upstream_id: null, + did_send_upstream: false, + failure_stage: input.failureStage, + final_selection_reason: null, + selection_strategy: "weighted", + }; +} + +async function logRejectedRequest(input: { + apiKeyId: string | null; + apiKeyName?: string | null; + apiKeyPrefix?: string | null; + userId?: string | null; + request: NextRequest; + path: string; + model: string | null; + reasoningEffort?: ReasoningEffort | null; + requestedServiceTier?: RequestedServiceTier | null; + thinkingConfig?: RequestThinkingConfig | null; + requestId: string; + startTime: number; + statusCode: number; + errorMessage: string; + routingDecision: RoutingDecisionLog; + routingType?: "tiered" | "direct" | "provider_type" | null; + priorityTier?: number | null; + routingDurationMs?: number | null; + sessionId?: string | null; + isStream?: boolean; +}): Promise { + try { + await logRequest({ + apiKeyId: input.apiKeyId, + apiKeyName: input.apiKeyName ?? null, + apiKeyPrefix: input.apiKeyPrefix ?? null, + userId: input.userId ?? null, + upstreamId: null, + method: input.request.method, + path: input.path, + model: input.model, + reasoningEffort: input.reasoningEffort ?? null, + requestedServiceTier: input.requestedServiceTier ?? null, + effectiveServiceTier: null, + promptTokens: 0, + completionTokens: 0, + totalTokens: 0, + statusCode: input.statusCode, + durationMs: Date.now() - input.startTime, + routingDurationMs: input.routingDurationMs ?? null, + errorMessage: input.errorMessage, + routingType: input.routingType ?? null, + priorityTier: input.priorityTier ?? null, + failoverAttempts: 0, + failoverHistory: null, + routingDecision: input.routingDecision, + thinkingConfig: input.thinkingConfig ?? null, + sessionId: input.sessionId ?? null, + affinityHit: false, + affinityMigrated: false, + isStream: input.isStream ?? false, + }); + } catch (error) { + log.error({ err: error, requestId: input.requestId }, "failed to log rejected proxy request"); + } +} function buildApiKeyQuotaExceededErrorMessage( apiKeyId: string, @@ -214,57 +307,32 @@ async function logApiKeyQuotaRejectedRequest(input: { matchedRouteCapability: RouteCapability | null; routeMatchSource: RouteMatchSource | null; errorMessage: string; + isStream: boolean; }): Promise { - const routingDecision: RoutingDecisionLog = { - original_model: input.model ?? "(path-based)", - resolved_model: input.model ?? "(path-based)", - model_redirect_applied: false, - provider_type: null, - routing_type: "none", - matched_route_capability: input.matchedRouteCapability, - route_match_source: input.routeMatchSource, - capability_candidates_count: 0, - candidates: [], - excluded: [], - candidate_count: 0, - final_candidate_count: 0, - selected_upstream_id: null, - candidate_upstream_id: null, - actual_upstream_id: null, - did_send_upstream: false, - failure_stage: "candidate_selection", - selection_strategy: "weighted", - }; - - await logRequest({ + await logRejectedRequest({ apiKeyId: input.apiKeyId, apiKeyName: input.apiKeyName, apiKeyPrefix: input.apiKeyPrefix, userId: input.userId, - upstreamId: null, - method: input.request.method, + request: input.request, path: input.path, model: input.model, reasoningEffort: input.reasoningEffort, requestedServiceTier: input.requestedServiceTier, - effectiveServiceTier: null, - promptTokens: 0, - completionTokens: 0, - totalTokens: 0, - statusCode: getHttpStatusForError("API_KEY_QUOTA_EXCEEDED"), - durationMs: Date.now() - input.startTime, - routingDurationMs: null, - errorMessage: input.errorMessage, - routingType: null, - priorityTier: null, - failoverAttempts: 0, - failoverHistory: null, - routingDecision, thinkingConfig: input.thinkingConfig, + requestId: input.requestId, + startTime: input.startTime, sessionId: input.sessionId, - affinityHit: false, - affinityMigrated: false, - isStream: false, + statusCode: getHttpStatusForError("API_KEY_QUOTA_EXCEEDED"), + errorMessage: input.errorMessage, + isStream: input.isStream, + routingDecision: buildRoutingDecisionLog({ + model: input.model, + matchedRouteCapability: input.matchedRouteCapability, + routeMatchSource: input.routeMatchSource, + failureStage: "auth_filter", + providerType: null, + }), }); } @@ -293,59 +361,31 @@ async function logApiKeyAdmissionRejectedRequest(input: { routeMatchSource: RouteMatchSource | null; errorCode: "API_KEY_MODEL_NOT_ALLOWED" | "API_KEY_RATE_LIMITED"; errorMessage: string; + isStream: boolean; }): Promise { - const routingDecision: RoutingDecisionLog = { - original_model: input.model ?? "(path-based)", - resolved_model: input.model ?? "(path-based)", - model_redirect_applied: false, - provider_type: input.matchedRouteCapability - ? getProviderByRouteCapability(input.matchedRouteCapability) - : null, - routing_type: "none", - matched_route_capability: input.matchedRouteCapability, - route_match_source: input.routeMatchSource, - capability_candidates_count: 0, - candidates: [], - excluded: [], - candidate_count: 0, - final_candidate_count: 0, - selected_upstream_id: null, - candidate_upstream_id: null, - actual_upstream_id: null, - did_send_upstream: false, - failure_stage: "auth_filter", - selection_strategy: "weighted", - }; - - await logRequest({ + await logRejectedRequest({ apiKeyId: input.apiKeyId, apiKeyName: input.apiKeyName, apiKeyPrefix: input.apiKeyPrefix, userId: input.userId, - upstreamId: null, - method: input.request.method, + request: input.request, path: input.path, model: input.model, reasoningEffort: input.reasoningEffort, requestedServiceTier: input.requestedServiceTier, - effectiveServiceTier: null, - promptTokens: 0, - completionTokens: 0, - totalTokens: 0, - statusCode: getHttpStatusForError(input.errorCode), - durationMs: Date.now() - input.startTime, - routingDurationMs: null, - errorMessage: input.errorMessage, - routingType: null, - priorityTier: null, - failoverAttempts: 0, - failoverHistory: null, - routingDecision, thinkingConfig: input.thinkingConfig, + requestId: input.requestId, + startTime: input.startTime, sessionId: input.sessionId, - affinityHit: false, - affinityMigrated: false, - isStream: false, + statusCode: getHttpStatusForError(input.errorCode), + errorMessage: input.errorMessage, + isStream: input.isStream, + routingDecision: buildRoutingDecisionLog({ + model: input.model, + matchedRouteCapability: input.matchedRouteCapability, + routeMatchSource: input.routeMatchSource, + failureStage: "auth_filter", + }), }); } @@ -361,28 +401,12 @@ async function logLocalApiKeyModelListRequest(input: { matchedRouteCapability: RouteCapability | null; routeMatchSource: RouteMatchSource | null; }): Promise { - const routingDecision: RoutingDecisionLog = { - original_model: "(model-list)", - resolved_model: "(model-list)", - model_redirect_applied: false, - provider_type: input.matchedRouteCapability - ? getProviderByRouteCapability(input.matchedRouteCapability) - : null, - routing_type: "none", - matched_route_capability: input.matchedRouteCapability, - route_match_source: input.routeMatchSource, - capability_candidates_count: 0, - candidates: [], - excluded: [], - candidate_count: 0, - final_candidate_count: 0, - selected_upstream_id: null, - candidate_upstream_id: null, - actual_upstream_id: null, - did_send_upstream: false, - failure_stage: null, - selection_strategy: "weighted", - }; + const routingDecision = buildRoutingDecisionLog({ + model: "(model-list)", + matchedRouteCapability: input.matchedRouteCapability, + routeMatchSource: input.routeMatchSource, + failureStage: null, + }); await logRequest({ apiKeyId: input.apiKeyId, @@ -993,6 +1017,7 @@ function extractReasoningEffortFromBody( async function extractRequestContext(request: NextRequest, path: string): Promise { const modelFromPath = extractGeminiModelFromPath(path); + const requestUrl = new URL(request.url); try { const clonedRequest = request.clone(); @@ -1003,7 +1028,7 @@ async function extractRequestContext(request: NextRequest, path: string): Promis model: modelFromPath, sessionId: null, bodyJson: null, - isStream: false, + isStream: isStreamRequest({}, path, requestUrl), reasoningEffort: null, requestedServiceTier: null, }; @@ -1011,7 +1036,7 @@ async function extractRequestContext(request: NextRequest, path: string): Promis const bodyJson = JSON.parse(bodyText) as Record; const modelFromBody = typeof bodyJson.model === "string" ? bodyJson.model || null : null; - const isStream = bodyJson.stream === true; + const isStream = isStreamRequest(bodyJson, path, requestUrl); const reasoningEffort = extractReasoningEffortFromBody(bodyJson); const requestedServiceTier = normalizeRequestedServiceTier(bodyJson.service_tier); @@ -1029,7 +1054,7 @@ async function extractRequestContext(request: NextRequest, path: string): Promis model: modelFromPath, sessionId: null, bodyJson: null, - isStream: false, + isStream: isStreamRequest({}, path, requestUrl), reasoningEffort: null, requestedServiceTier: null, }; @@ -1047,12 +1072,43 @@ export async function handleProxy(request: NextRequest, context: RouteContext): // 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: { + errorMessage: string; + apiKeyId?: string | null; + apiKeyName?: string | null; + apiKeyPrefix?: string | null; + userId?: string | null; + }): Promise => { + await logRejectedRequest({ + apiKeyId: input.apiKeyId ?? null, + apiKeyName: input.apiKeyName, + apiKeyPrefix: input.apiKeyPrefix, + userId: input.userId, + request, + path, + model: null, + requestId, + startTime, + statusCode: 401, + errorMessage: input.errorMessage, + isStream: authRequestIsStream, + routingDecision: buildRoutingDecisionLog({ + model: null, + matchedRouteCapability: null, + routeMatchSource: null, + failureStage: "auth_filter", + }), + }); + }; // Extract and validate API key const { keyValue, authSource } = extractProxyApiKey(request); 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 }); } log.debug({ requestId, authSource }, "proxy auth: extracted API key"); @@ -1069,6 +1125,13 @@ export async function handleProxy(request: NextRequest, context: RouteContext): if (isValid) { // Check expiration if (candidate.expiresAt && candidate.expiresAt < new Date()) { + await logAuthRejected({ + errorMessage: "API key has expired", + apiKeyId: candidate.id, + apiKeyName: candidate.name, + apiKeyPrefix: candidate.keyPrefix, + userId: candidate.userId, + }); return NextResponse.json({ error: "API key has expired" }, { status: 401 }); } validApiKey = candidate; @@ -1077,6 +1140,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 }); } @@ -1093,6 +1157,13 @@ export async function handleProxy(request: NextRequest, context: RouteContext): { requestId, keyPrefix: validApiKey.keyPrefix, ownerId: validApiKey.userId }, "proxy auth: rejected API key owned by an inactive user" ); + await logAuthRejected({ + errorMessage: "API key is disabled", + apiKeyId: validApiKey.id, + apiKeyName: validApiKey.name, + apiKeyPrefix: validApiKey.keyPrefix, + userId: validApiKey.userId, + }); return NextResponse.json({ error: "API key is disabled" }, { status: 401 }); } } @@ -1154,6 +1225,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): requestId, startTime, sessionId: null, + isStream: requestedStream, matchedRouteCapability, routeMatchSource: matchedRouteMatchSource, errorCode, @@ -1208,8 +1280,9 @@ export async function handleProxy(request: NextRequest, context: RouteContext): reasoningEffort, requestedServiceTier, thinkingConfig, - requestId, startTime, + requestId, + isStream: requestedStream, sessionId: null, matchedRouteCapability, routeMatchSource: matchedRouteMatchSource, @@ -1235,77 +1308,36 @@ export async function handleProxy(request: NextRequest, context: RouteContext): user_hint: "当前请求路径未匹配到受支持的能力类型，请检查请求方法和路径是否在支持列表中", }); - const unsupportedRoutingDecision: RoutingDecisionLog = { - original_model: model ?? "(path-based)", - resolved_model: model ?? "(path-based)", - model_redirect_applied: false, - provider_type: null, - routing_type: "none", - matched_route_capability: null, - route_match_source: null, - capability_candidates_count: 0, - candidates: [], - excluded: [], - candidate_count: 0, - final_candidate_count: 0, - selected_upstream_id: null, - candidate_upstream_id: null, - actual_upstream_id: null, - did_send_upstream: false, - failure_stage: "candidate_selection", - selection_strategy: "weighted", - }; + const unsupportedRoutingDecision = buildRoutingDecisionLog({ + model, + matchedRouteCapability: null, + routeMatchSource: null, + failureStage: "candidate_selection", + providerType: null, + }); log.warn( { requestId, method: request.method, path, matchedRouteCapability: null }, "path capability not matched, skipping upstream routing" ); - try { - const createdLog = await logRequest({ - apiKeyId: validApiKey.id, - ...apiKeySnapshot, - upstreamId: null, - method: request.method, - path, - model, - reasoningEffort, - requestedServiceTier, - effectiveServiceTier: null, - promptTokens: 0, - completionTokens: 0, - totalTokens: 0, - statusCode: unsupportedResponse.status, - durationMs: unsupportedDurationMs, - routingDurationMs: unsupportedDurationMs, - errorMessage: "path capability not matched, skipping upstream routing", - routingType, - priorityTier: null, - failoverAttempts: 0, - failoverHistory: null, - routingDecision: unsupportedRoutingDecision, - }); - - if (createdLog?.id) { - await persistBillingSnapshotSafely({ - requestLogId: createdLog.id, - apiKeyId: validApiKey.id, - upstreamId: null, - model, - requestedServiceTier, - effectiveServiceTier: null, - usage: { - promptTokens: 0, - completionTokens: 0, - totalTokens: 0, - cacheReadTokens: 0, - cacheWriteTokens: 0, - }, - requestId, - }); - } - } catch (error) { - log.error({ err: error, requestId, path }, "failed to log unsupported path request"); - } + await logRejectedRequest({ + apiKeyId: validApiKey.id, + ...apiKeySnapshot, + request, + path, + model, + reasoningEffort, + requestedServiceTier, + thinkingConfig, + requestId, + startTime, + statusCode: unsupportedResponse.status, + errorMessage: "path capability not matched, skipping upstream routing", + routingType, + routingDurationMs: unsupportedDurationMs, + isStream: requestedStream, + routingDecision: unsupportedRoutingDecision, + }); return unsupportedResponse; } @@ -1336,11 +1368,35 @@ export async function handleProxy(request: NextRequest, context: RouteContext): activeUpstreamSnapshot = await loadActiveUpstreamSnapshot(); } catch (error) { log.error({ err: error, requestId }, "failed to load active upstream snapshot"); - return createUnifiedErrorResponse("SERVICE_UNAVAILABLE", { + const snapshotResponse = createUnifiedErrorResponse("SERVICE_UNAVAILABLE", { did_send_upstream: false, request_id: requestId, user_hint: "上游状态暂时不可用，请稍后重试", }); + await logRejectedRequest({ + apiKeyId: validApiKey.id, + ...apiKeySnapshot, + request, + path, + model, + reasoningEffort, + requestedServiceTier, + thinkingConfig, + requestId, + startTime, + statusCode: snapshotResponse.status, + errorMessage: "failed to load active upstream snapshot", + routingType, + routingDurationMs: Date.now() - startTime, + isStream: requestedStream, + routingDecision: buildRoutingDecisionLog({ + model, + matchedRouteCapability, + routeMatchSource, + failureStage: "candidate_selection", + }), + }); + return snapshotResponse; } const activeUpstreams = activeUpstreamSnapshot.map((entry) => entry.upstream); const allowedUpstreamIds = @@ -1449,7 +1505,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): }, "no upstream supports matched route capability" ); - return createUnifiedErrorResponse("NO_UPSTREAMS_CONFIGURED", { + const noCapabilityResponse = createUnifiedErrorResponse("NO_UPSTREAMS_CONFIGURED", { reason: "NO_HEALTHY_CANDIDATES", did_send_upstream: false, request_id: requestId, @@ -1458,6 +1514,42 @@ export async function handleProxy(request: NextRequest, context: RouteContext): ? `未找到支持路径能力 ${matchedRouteCapability} 或回退能力 ${fallbackCapability} 的上游，请先检查上游能力配置` : `未找到支持路径能力 ${matchedRouteCapability} 的上游，请先检查上游能力配置`, }); + await logRejectedRequest({ + apiKeyId: validApiKey.id, + ...apiKeySnapshot, + request, + path, + model, + reasoningEffort, + requestedServiceTier, + thinkingConfig, + requestId, + startTime, + statusCode: noCapabilityResponse.status, + errorMessage: "no upstream supports matched route capability", + routingType, + routingDurationMs: Date.now() - startTime, + isStream: requestedStream, + routingDecision: transformPathRoutingDecisionLog( + { + matchedRouteCapability, + routeMatchSource, + originalModel: model, + resolvedModel: model, + modelRedirectApplied: false, + capabilityCandidates, + finalCandidates: [], + excludedCandidates: [], + candidateCircuitStates: buildCandidateCircuitStateMap( + capabilityCandidates, + activeUpstreamSnapshot + ), + }, + null, + { didSendUpstream: false, failureStage: "candidate_selection" } + ), + }); + return noCapabilityResponse; } if (finalCapabilityCandidates.length === 0) { @@ -1473,14 +1565,49 @@ export async function handleProxy(request: NextRequest, context: RouteContext): }, "no authorized upstream for matched route capability" ); - return createUnifiedErrorResponse("NO_AUTHORIZED_UPSTREAMS", { + const noAuthorizedResponse = createUnifiedErrorResponse("NO_AUTHORIZED_UPSTREAMS", { reason: "NO_AUTHORIZED_UPSTREAMS", did_send_upstream: false, request_id: requestId, user_hint: "当前密钥没有可用的路径能力授权，请在密钥配置中绑定对应上游", }); + await logRejectedRequest({ + apiKeyId: validApiKey.id, + ...apiKeySnapshot, + request, + path, + model, + reasoningEffort, + requestedServiceTier, + thinkingConfig, + requestId, + startTime, + statusCode: noAuthorizedResponse.status, + errorMessage: "no authorized upstream for matched route capability", + routingType, + routingDurationMs: Date.now() - startTime, + isStream: requestedStream, + routingDecision: transformPathRoutingDecisionLog( + { + matchedRouteCapability, + routeMatchSource, + originalModel: model, + resolvedModel: model, + modelRedirectApplied: false, + capabilityCandidates, + finalCandidates: [], + excludedCandidates: [], + candidateCircuitStates: buildCandidateCircuitStateMap( + capabilityCandidates, + activeUpstreamSnapshot + ), + }, + null, + { didSendUpstream: false, failureStage: "candidate_selection" } + ), + }); + return noAuthorizedResponse; } - excludedCapabilityCandidates = []; candidateCircuitStates = buildCandidateCircuitStateMap( capabilityCandidates, @@ -1522,37 +1649,24 @@ export async function handleProxy(request: NextRequest, context: RouteContext): } ); - try { - await logRequest({ - apiKeyId: validApiKey.id, - ...apiKeySnapshot, - upstreamId: null, - method: request.method, - path, - model, - reasoningEffort, - requestedServiceTier, - effectiveServiceTier: null, - promptTokens: 0, - completionTokens: 0, - totalTokens: 0, - statusCode: rejectedResponse.status, - durationMs: rejectedDurationMs, - routingDurationMs: rejectedDurationMs, - errorMessage: "all authorized upstreams were excluded by model rules", - routingType, - priorityTier: null, - failoverAttempts: 0, - failoverHistory: null, - routingDecision: rejectedRoutingDecision, - thinkingConfig, - }); - } catch (error) { - log.error( - { err: error, requestId, matchedRouteCapability, model }, - "failed to log model rule exclusion response" - ); - } + await logRejectedRequest({ + apiKeyId: validApiKey.id, + ...apiKeySnapshot, + request, + path, + model, + reasoningEffort, + requestedServiceTier, + thinkingConfig, + requestId, + startTime, + statusCode: rejectedResponse.status, + errorMessage: "all authorized upstreams were excluded by model rules", + routingType, + routingDurationMs: rejectedDurationMs, + isStream: requestedStream, + routingDecision: rejectedRoutingDecision, + }); return rejectedResponse; } @@ -1598,6 +1712,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): sessionId: null, matchedRouteCapability, routeMatchSource: matchedRouteMatchSource, + isStream: requestedStream, errorMessage, }); } catch (error) { @@ -1645,6 +1760,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): let failoverHistory: FailoverAttempt[] = []; let requestLogId: string | null = null; let requestLogReady: Promise = Promise.resolve(null); + let queueStatePersistence: Promise = Promise.resolve(); let isAffinityHit = false; let isAffinityMigrated = false; @@ -1704,9 +1820,13 @@ export async function handleProxy(request: NextRequest, context: RouteContext): return requestLogId; }; - const persistQueueWaitingState = (queue: RoutingQueueLog) => - requestLogReady + const persistQueueWaitingState = (queue: RoutingQueueLog) => { + const persistence = queueStatePersistence + .then(() => requestLogReady) .then(async (readyLogId) => { + if (request.signal.aborted) { + return; + } const currentLogId = requestLogId ?? readyLogId; if (!currentLogId) { return; @@ -1744,6 +1864,9 @@ export async function handleProxy(request: NextRequest, context: RouteContext): .catch((error) => { log.error({ err: error, requestId }, "failed to update request log queue state"); }); + queueStatePersistence = persistence; + return persistence; + }; // Forward request to upstream let compensationHeaders: import("@/lib/services/proxy-client").CompensationHeader[] = []; @@ -1936,8 +2059,8 @@ export async function handleProxy(request: NextRequest, context: RouteContext): ); // Complete the start log asynchronously without delaying response handling. - void requestLogReady - .then((readyLogId) => { + void Promise.all([requestLogReady, queueStatePersistence]) + .then(([readyLogId]) => { const currentLogId = requestLogId ?? readyLogId; if (!currentLogId) { return; @@ -2489,6 +2612,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): recordApiKeyTokenUsage(validApiKey.id, usageForBilling.totalTokens, validApiKey.tpmLimit); // Log request + await queueStatePersistence; await awaitRequestLogReady(); let persistedLogId: string | null = requestLogId; if (requestLogId) { @@ -2676,16 +2800,14 @@ export async function handleProxy(request: NextRequest, context: RouteContext): const durationMs = Date.now() - startTime; const lastFailoverAttempt = failoverHistory[failoverHistory.length - 1]; - const lastSentFailoverAttempt = getLastSentFailoverAttempt(failoverHistory); - const didSendUpstream = resolveDidSendUpstream( - error as FailoverErrorWithHistory | null, - lastSentFailoverAttempt - ); + const failoverError = error as FailoverErrorWithHistory | null; + const lastDispatchedFailoverAttempt = failoverError?.lastDispatchedFailoverAttempt; + const didSendUpstream = resolveDidSendUpstream(failoverError); if (!didSendUpstream) { routingDurationMs ??= durationMs; } const attributionFailoverAttempt = didSendUpstream - ? (lastSentFailoverAttempt ?? lastFailoverAttempt) + ? (lastDispatchedFailoverAttempt ?? lastFailoverAttempt) : lastFailoverAttempt; const queueLifecycle = withQueueStreamFlag( (error as FailoverErrorWithHistory | null)?.queue ?? null, @@ -2712,7 +2834,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): const failureStage = resolveFailureStage(error, didSendUpstream, attributionFailoverAttempt); const failureReason = resolveFailureReason(error, didSendUpstream, attributionFailoverAttempt); const actualUpstreamId = - lastSentFailoverAttempt?.upstream_id ?? + lastDispatchedFailoverAttempt?.upstream_id ?? (didSendUpstream ? (selectedCandidate?.id ?? null) : null); const candidateUpstreamId = didSendUpstream ? (attributionFailoverAttempt?.upstream_id ?? selectedCandidate?.id ?? null) @@ -2818,6 +2940,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): statusCode: 200, errorMessage: null, }; + await queueStatePersistence; await awaitRequestLogReady(); if (requestLogId) { await updateRequestLog(requestLogId, modelListLogFields); @@ -2838,7 +2961,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): } } - if (shouldRecordFailure && inboundBody) { + if (shouldRecordFailure && inboundBody && didSendUpstream) { const fallbackOutboundHeaders = filterHeaders(new Headers(request.headers)).filtered; applyCompensationHeaders(fallbackOutboundHeaders, compensationHeaders); const fallbackProviderType = @@ -2962,6 +3085,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): } // Log failed request (internal logging with full details) + await queueStatePersistence; const failureLogFields = { ...failureLogBaseFields, upstreamId: actualUpstreamId, @@ -2984,7 +3108,7 @@ export async function handleProxy(request: NextRequest, context: RouteContext): persistedLogId = createdLog.id; } - if (persistedLogId) { + if (persistedLogId && didSendUpstream) { await persistBillingSnapshotSafely({ requestLogId: persistedLogId, apiKeyId: validApiKey.id, diff --git a/src/lib/services/load-balancer.ts b/src/lib/services/load-balancer.ts index fad267fd..24119054 100644 --- a/src/lib/services/load-balancer.ts +++ b/src/lib/services/load-balancer.ts @@ -470,26 +470,19 @@ function tryReserveConnectionSlot(upstream: Upstream): { current: number; max: number | null; } { - const current = getConnectionCount(upstream.id); const max = upstream.maxConcurrency; - - if (max === null || max === undefined || max <= 0) { - upstreamQueueAdmission.tryReserveImmediate({ - upstreamId: upstream.id, - maxConcurrency: null, - }); - return { reserved: true, current, max: null }; - } - - if (current >= max) { - return { reserved: false, current, max }; - } - - upstreamQueueAdmission.tryReserveImmediate({ + const reservation = upstreamQueueAdmission.tryReserveImmediate({ upstreamId: upstream.id, - maxConcurrency: max, + maxConcurrency: max == null || max <= 0 ? null : max, }); - return { reserved: true, current, max }; + + return { + reserved: reservation.reserved, + current: reservation.reserved + ? Math.max(0, reservation.activeCount - 1) + : reservation.activeCount, + max: max == null || max <= 0 ? null : max, + }; } /** diff --git a/src/lib/services/proxy-client.ts b/src/lib/services/proxy-client.ts index b5676285..784b7b68 100644 --- a/src/lib/services/proxy-client.ts +++ b/src/lib/services/proxy-client.ts @@ -67,7 +67,40 @@ export interface HeaderDiff { unchanged: Array<{ header: string; value: string }>; } -type ProxyRequestErrorWithHeaderDiff = Error & { headerDiff?: HeaderDiff }; +/** Metadata attached to errors raised while preparing or dispatching an upstream request. */ +export interface ProxyRequestErrorMetadata { + headerDiff?: HeaderDiff | null; + fetchStarted: boolean; +} + +type ProxyRequestErrorWithMetadata = Error & { + proxyRequestMetadata?: ProxyRequestErrorMetadata; +}; + +/** Attach typed lifecycle evidence to a proxy request error. */ +export function attachProxyRequestErrorMetadata( + error: T, + metadata: ProxyRequestErrorMetadata +): T { + const existing = getProxyRequestErrorMetadata(error); + (error as ProxyRequestErrorWithMetadata).proxyRequestMetadata = { + ...(existing ?? {}), + ...metadata, + }; + return error; +} + +/** Read lifecycle evidence attached to a proxy request error. */ +export function getProxyRequestErrorMetadata(error: unknown): ProxyRequestErrorMetadata | null { + if (!error || typeof error !== "object") { + return null; + } + const metadata = (error as ProxyRequestErrorWithMetadata).proxyRequestMetadata; + if (!metadata || typeof metadata !== "object" || typeof metadata.fetchStarted !== "boolean") { + return null; + } + return metadata; +} /** * Error raised when a streaming upstream does not emit usable content in time. @@ -990,7 +1023,8 @@ function applyModelOverride( return { body: nextBody, path: nextPath }; } -function isStreamRequest( +/** Determine whether the request expects a streaming response. */ +export function isStreamRequest( bodyJson: Record, path: string, requestUrl: URL @@ -1058,7 +1092,8 @@ export async function forwardRequest( path: string, requestId: string, compensationHeaders?: CompensationHeader[], - modelOverride?: string + modelOverride?: string, + onDispatchStart?: () => void ): Promise { // Prepare headers const originalHeaders = new Headers(request.headers); @@ -1158,7 +1193,17 @@ export async function forwardRequest( const requestSearch = requestUrl.search; // Read request body - let body = await request.arrayBuffer(); + let body: ArrayBuffer; + try { + body = await request.arrayBuffer(); + } catch (error) { + const requestError = + error instanceof Error ? error : new Error(typeof error === "string" ? error : String(error)); + throw attachProxyRequestErrorMetadata(requestError, { + headerDiff: requestHeaderDiff, + fetchStarted: false, + }); + } let effectivePath = path; // CLIProxyAPI 单账号映射上游：将模型名改写为携带账号前缀的形式， @@ -1189,20 +1234,39 @@ export async function forwardRequest( } } - // Create abort controller for timeout + // Abort the upstream fetch when either its timeout or the downstream request ends. const controller = new AbortController(); - const timeoutId = setTimeout(() => controller.abort(), upstream.timeout * 1000); + const fetchSignal = AbortSignal.any([request.signal, controller.signal]); + let timeoutTriggered = false; + const timeoutId = setTimeout(() => { + timeoutTriggered = true; + controller.abort(); + }, upstream.timeout * 1000); const upstreamSendTime = Date.now(); + let fetchStarted = false; try { - // Make upstream request - const upstreamResponse = await fetch(url, { + if (request.signal.aborted) { + const abortError = new Error("The downstream request was aborted before upstream dispatch"); + abortError.name = "AbortError"; + throw abortError; + } + const upstreamResponsePromise = fetch(url, { method: request.method, headers, body: body.byteLength > 0 ? body : undefined, - signal: controller.signal, + signal: fetchSignal, }); + fetchStarted = true; + try { + onDispatchStart?.(); + } catch (error) { + controller.abort(); + await upstreamResponsePromise.catch(() => undefined); + throw error; + } + const upstreamResponse = await upstreamResponsePromise; clearTimeout(timeoutId); @@ -1352,17 +1416,30 @@ export async function forwardRequest( clearTimeout(timeoutId); if (error instanceof Error && error.name === "AbortError") { + if (request.signal.aborted && !timeoutTriggered) { + reqLog.warn({ upstream: upstream.name }, "upstream request cancelled by downstream client"); + const cancellationError = new Error("Upstream request cancelled by downstream client"); + throw attachProxyRequestErrorMetadata(cancellationError, { + headerDiff: requestHeaderDiff, + fetchStarted, + }); + } reqLog.error({ timeout: upstream.timeout }, "upstream request timed out"); const timeoutError = new Error(`Upstream request timed out after ${upstream.timeout}s`); - (timeoutError as ProxyRequestErrorWithHeaderDiff).headerDiff = requestHeaderDiff; - throw timeoutError; + throw attachProxyRequestErrorMetadata(timeoutError, { + headerDiff: requestHeaderDiff, + fetchStarted, + }); } const requestError = error instanceof Error ? error : new Error(typeof error === "string" ? error : String(error)); - (requestError as ProxyRequestErrorWithHeaderDiff).headerDiff = requestHeaderDiff; - reqLog.error({ err: requestError }, "upstream request failed"); - throw requestError; + const requestErrorWithMetadata = attachProxyRequestErrorMetadata(requestError, { + headerDiff: requestHeaderDiff, + fetchStarted, + }); + reqLog.error({ err: requestErrorWithMetadata }, "upstream request failed"); + throw requestErrorWithMetadata; } } diff --git a/tests/unit/api/proxy/route.test.ts b/tests/unit/api/proxy/route.test.ts index 1f642dd9..be055dc2 100644 --- a/tests/unit/api/proxy/route.test.ts +++ b/tests/unit/api/proxy/route.test.ts @@ -115,7 +115,8 @@ vi.mock("@/lib/services/route-capability-migration", () => ({ ensureRouteCapabilityMigration: vi.fn(async () => {}), })); -vi.mock("@/lib/services/proxy-client", () => { +vi.mock("@/lib/services/proxy-client", async (importOriginal) => { + const actual = await importOriginal(); class FirstByteTimeoutError extends Error { constructor(public readonly timeoutMs: number) { super(`Upstream first byte timed out after ${Math.round(timeoutMs / 1000)}s`); @@ -150,6 +151,8 @@ vi.mock("@/lib/services/proxy-client", () => { } return { + isStreamRequest: actual.isStreamRequest, + getProxyRequestErrorMetadata: actual.getProxyRequestErrorMetadata, forwardRequest: vi.fn(), prepareUpstreamForProxy: vi.fn((upstream, timeoutConfig) => ({ id: upstream.id, @@ -554,6 +557,10 @@ describe("proxy route upstream selection", () => { request: NextRequest, context: { params: Promise<{ path: string[] }> } ) => Promise; + let handleProxy: ( + request: NextRequest, + context: { params: Promise<{ path: string[] }> } + ) => Promise; let GET: ( request: NextRequest, context: { params: Promise<{ path: string[] }> } @@ -593,6 +600,8 @@ describe("proxy route upstream selection", () => { const routeModule = await import("@/app/api/proxy/v1/[...path]/route"); const { db } = await import("@/lib/db"); POST = routeModule.POST; + const lifecycleModule = await import("@/app/api/proxy/v1/[...path]/proxy-request-lifecycle"); + handleProxy = lifecycleModule.handleProxy; GET = routeModule.GET; vi.mocked(db.query.upstreams.findMany).mockResolvedValue(DEFAULT_ACTIVE_UPSTREAMS); vi.mocked(db.query.upstreamHealth.findMany).mockResolvedValue([]); @@ -804,6 +813,88 @@ describe("proxy route upstream selection", () => { ); expect(lifecycleEvents.indexOf("billing")).toBeLessThan(lifecycleEvents.indexOf("recording")); }); + 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 { logRequest } = await import("@/lib/services/request-logger"); + const { calculateAndPersistRequestBillingSnapshot } = + await import("@/lib/services/billing-cost-service"); + const { buildFixture, recordTrafficFixture } = await import("@/lib/services/traffic-recorder"); + + process.env.RECORDER_ENABLED = "true"; + process.env.RECORDER_MODE = "all"; + vi.mocked(db.query.apiKeys.findMany).mockResolvedValueOnce([ + { + id: "key-rejected", + keyHash: "hash-rejected", + keyPrefix: "sk-test", + expiresAt: null, + isActive: true, + }, + ]); + + const response = await handleProxy( + new NextRequest("http://localhost/api/proxy/v1/custom/not-matched", { + method: "POST", + headers: { + authorization: "Bearer sk-rejected", + "content-type": "application/json", + }, + body: JSON.stringify({ model: "gpt-5.2", input: "hello" }), + }), + { params: Promise.resolve({ path: ["custom", "not-matched"] }) } + ); + + expect(response.status).toBe(503); + expect(logRequest).toHaveBeenCalledWith( + expect.objectContaining({ + statusCode: 503, + durationMs: expect.any(Number), + routingDecision: expect.objectContaining({ + failure_stage: "candidate_selection", + did_send_upstream: false, + }), + }) + ); + expect(calculateAndPersistRequestBillingSnapshot).not.toHaveBeenCalled(); + expect(buildFixture).not.toHaveBeenCalled(); + 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 { logRequest } = await import("@/lib/services/request-logger"); + const { calculateAndPersistRequestBillingSnapshot } = + await import("@/lib/services/billing-cost-service"); + const { recordTrafficFixture } = await import("@/lib/services/traffic-recorder"); + + process.env.RECORDER_ENABLED = "true"; + process.env.RECORDER_MODE = "all"; + + const response = await handleProxy( + 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"] }) } + ); + + expect(response.status).toBe(401); + expect(logRequest).toHaveBeenCalledWith( + expect.objectContaining({ + apiKeyId: null, + statusCode: 401, + errorMessage: "Missing API key", + isStream: true, + routingDecision: expect.objectContaining({ + failure_stage: "auth_filter", + did_send_upstream: false, + }), + }) + ); + expect(calculateAndPersistRequestBillingSnapshot).not.toHaveBeenCalled(); + expect(recordTrafficFixture).not.toHaveBeenCalled(); + }); describe("proxy auth header compatibility", () => { it("should authenticate using x-api-key when authorization is absent", async () => { @@ -3101,21 +3192,7 @@ describe("proxy route upstream selection", () => { }), }) ); - expect(calculateAndPersistRequestBillingSnapshot).toHaveBeenCalledWith( - expect.objectContaining({ - requestLogId: "unsupported-log", - apiKeyId: "key-1", - upstreamId: null, - model: "gpt-5.2", - usage: { - promptTokens: 0, - completionTokens: 0, - totalTokens: 0, - cacheReadTokens: 0, - cacheWriteTokens: 0, - }, - }) - ); + expect(calculateAndPersistRequestBillingSnapshot).not.toHaveBeenCalled(); }); it("should route messages requests to anthropic upstream when available", async () => { @@ -3204,7 +3281,8 @@ describe("proxy route upstream selection", () => { "v1/messages", expect.any(String), expect.any(Array), - undefined + undefined, + expect.any(Function) ); }); @@ -3725,7 +3803,9 @@ describe("proxy route upstream selection", () => { }), }); - const response = await POST(request, { params: Promise.resolve({ path: ["v1", "messages"] }) }); + const response = await handleProxy(request, { + params: Promise.resolve({ path: ["v1", "messages"] }), + }); expect(response.status).toBe(200); expect(markUnhealthy).not.toHaveBeenCalledWith("up-anthropic-1", expect.any(String)); @@ -3825,7 +3905,9 @@ describe("proxy route upstream selection", () => { }), }); - const response = await POST(request, { params: Promise.resolve({ path: ["v1", "messages"] }) }); + const response = await handleProxy(request, { + params: Promise.resolve({ path: ["v1", "messages"] }), + }); const payload = (await response.json()) as { error: { reason?: string; user_hint?: string } }; expect(response.status).toBe(503); @@ -3960,7 +4042,9 @@ describe("proxy route upstream selection", () => { }), }); - const response = await POST(request, { params: Promise.resolve({ path: ["v1", "messages"] }) }); + const response = await handleProxy(request, { + params: Promise.resolve({ path: ["v1", "messages"] }), + }); expect(response.status).toBe(200); expect(vi.mocked(upstreamQueueAdmission.enqueueWait)).toHaveBeenCalledWith( @@ -4004,15 +4088,16 @@ describe("proxy route upstream selection", () => { }) ); }); - - it("should release the resumed slot and reselect once when the queued upstream disappears", async () => { + it("releases a handed-off queue slot when the client aborts before dispatch", async () => { const { db } = await import("@/lib/db"); const { forwardRequest } = await import("@/lib/services/proxy-client"); - const { routeByModel } = await import("@/lib/services/model-router"); + const { updateRequestLog } = await import("@/lib/services/request-logger"); + const { calculateAndPersistRequestBillingSnapshot } = + await import("@/lib/services/billing-cost-service"); + const { recordTrafficFixture } = await import("@/lib/services/traffic-recorder"); const { selectFromProviderType, decideQueuedUpstreamResume, - reselectQueuedUpstreamOnce, AllCandidatesConcurrencyFullError, releaseConnection, } = await import("@/lib/services/load-balancer"); @@ -4031,47 +4116,17 @@ describe("proxy route upstream selection", () => { max_queue_length: 4, }, }; - const fallbackUpstream = { - ...DEFAULT_ACTIVE_UPSTREAMS[1], - id: "up-fallback", - name: "fallback-upstream", - providerType: "anthropic", - routeCapabilities: ["anthropic_messages"], - baseUrl: "https://api.anthropic.com", - }; + const controller = new AbortController(); vi.mocked(db.query.apiKeys.findMany).mockResolvedValueOnce([ { id: "key-1", keyHash: "hash-1", expiresAt: null, isActive: true }, ]); - vi.mocked(db.query.upstreams.findMany).mockResolvedValueOnce([ - waitableUpstream, - fallbackUpstream, - ]); + vi.mocked(db.query.upstreams.findMany) + .mockResolvedValueOnce([waitableUpstream]) + .mockResolvedValueOnce([waitableUpstream]); vi.mocked(db.query.apiKeyUpstreams.findMany).mockResolvedValueOnce([ { upstreamId: "up-queued" }, - { upstreamId: "up-fallback" }, ]); - - vi.mocked(routeByModel).mockResolvedValueOnce({ - upstream: waitableUpstream, - providerType: "anthropic", - resolvedModel: "claude-test", - candidateUpstreams: [], - excludedUpstreams: [], - routingDecision: { - originalModel: "claude-test", - resolvedModel: "claude-test", - providerType: "anthropic", - upstreamName: "queued-upstream", - allowedModelsFilter: false, - modelRedirectApplied: false, - circuitBreakerFilter: false, - routingType: "provider_type", - candidateCount: 2, - finalCandidateCount: 0, - }, - }); - vi.mocked(selectFromProviderType).mockRejectedValueOnce( new AllCandidatesConcurrencyFullError( [ @@ -4106,107 +4161,615 @@ describe("proxy route upstream selection", () => { queueLengthRemaining: 0, }), }); - vi.mocked(decideQueuedUpstreamResume).mockResolvedValueOnce({ - action: "reselect_once", - reason: "bound_missing", - upstream: null, - excludeIds: ["up-queued"], - }); - vi.mocked(reselectQueuedUpstreamOnce).mockResolvedValueOnce({ - upstream: fallbackUpstream, - selectedTier: 1, - circuitBreakerFiltered: 0, - quotaFiltered: 0, - concurrencyFiltered: 0, - concurrencyExcluded: [], - totalCandidates: 2, - affinityHit: false, - affinityMigrated: false, - selectionReason: null, - }); - vi.mocked(forwardRequest).mockResolvedValueOnce({ - statusCode: 200, - headers: new Headers(), - body: new Uint8Array(), - isStream: false, - usage: null, - headerDiff: null, + vi.mocked(decideQueuedUpstreamResume).mockImplementationOnce(async () => { + controller.abort(); + return { + action: "resume", + reason: "bound_available", + upstream: waitableUpstream, + excludeIds: [], + }; }); - const request = new NextRequest("http://localhost/api/proxy/v1/messages", { - method: "POST", - headers: { - authorization: "Bearer sk-test", - "content-type": "application/json", - }, - body: JSON.stringify({ - model: "claude-test", - messages: [{ role: "user", content: "hi" }], + const response = await handleProxy( + new NextRequest("http://localhost/api/proxy/v1/messages", { + method: "POST", + signal: controller.signal, + headers: { + authorization: "Bearer sk-test", + "content-type": "application/json", + }, + body: JSON.stringify({ + model: "claude-test", + messages: [{ role: "user", content: "hi" }], + }), }), - }); - - const response = await POST(request, { params: Promise.resolve({ path: ["v1", "messages"] }) }); + { params: Promise.resolve({ path: ["v1", "messages"] }) } + ); + const data = await response.json(); - expect(response.status).toBe(200); - expect(vi.mocked(reselectQueuedUpstreamOnce)).toHaveBeenCalledWith( - "up-queued", - ["up-queued", "up-fallback"], - ["up-queued"], - expect.objectContaining({ candidateSnapshot: expect.any(Array) }) + expect(response.status).toBe(499); + expect(data.error).toEqual( + expect.objectContaining({ code: "CLIENT_DISCONNECTED", did_send_upstream: false }) ); - expect(vi.mocked(releaseConnection).mock.calls.map(([upstreamId]) => upstreamId)).toEqual([ - "up-queued", - "up-fallback", - ]); + expect(forwardRequest).not.toHaveBeenCalled(); + expect(vi.mocked(releaseConnection)).toHaveBeenCalledWith("up-queued"); + expect(calculateAndPersistRequestBillingSnapshot).not.toHaveBeenCalled(); + expect(recordTrafficFixture).not.toHaveBeenCalled(); + expect( + vi + .mocked(updateRequestLog) + .mock.calls.some(([, payload]) => payload?.routingDecision?.queue?.status === "aborted") + ).toBe(true); + const queueAbortLog = vi + .mocked(updateRequestLog) + .mock.calls.find(([, payload]) => payload?.routingDecision?.queue?.status === "aborted")?.[1]; + expect(queueAbortLog?.routingDecision?.failure_stage).toBe("candidate_selection"); }); - - it("should classify queue wait timeout separately from upstream timeout", async () => { + it("releases a selected slot when the client aborts during pre-dispatch setup", async () => { const { db } = await import("@/lib/db"); const { forwardRequest } = await import("@/lib/services/proxy-client"); - const { routeByModel } = await import("@/lib/services/model-router"); const { updateRequestLog } = await import("@/lib/services/request-logger"); - const { markUnhealthy } = await import("@/lib/services/health-checker"); - const { recordFailure } = await import("@/lib/services/circuit-breaker"); - const { selectFromProviderType, AllCandidatesConcurrencyFullError } = + const { calculateAndPersistRequestBillingSnapshot } = + await import("@/lib/services/billing-cost-service"); + const { recordTrafficFixture } = await import("@/lib/services/traffic-recorder"); + const { selectFromProviderType, releaseConnection } = await import("@/lib/services/load-balancer"); - const { upstreamQueueAdmission, UpstreamQueueWaitTimeoutError } = - await import("@/lib/services/upstream-queue-admission"); - const waitableUpstream = { - ...DEFAULT_ACTIVE_UPSTREAMS[0], - id: "up-queued", - name: "queued-upstream", - providerType: "anthropic", - routeCapabilities: ["anthropic_messages"], - baseUrl: "https://api.anthropic.com", - queuePolicy: { - enabled: true, - timeout_ms: 30000, - max_queue_length: 4, - }, + const upstream = DEFAULT_ACTIVE_UPSTREAMS[0]; + const controller = new AbortController(); + const circuitConfig = { + failureThreshold: 5, + successThreshold: 2, + openDuration: 300000, + probeInterval: 30000, + firstByteTimeout: 30000, + streamIdleTimeout: 60000, }; + let resolveCircuitConfig!: (value: typeof circuitConfig) => void; + const circuitConfigPromise = new Promise((resolve) => { + resolveCircuitConfig = resolve; + }); vi.mocked(db.query.apiKeys.findMany).mockResolvedValueOnce([ - { id: "key-1", keyHash: "hash-1", expiresAt: null, isActive: true }, + { id: "key-pre-dispatch", keyHash: "hash-1", expiresAt: null, isActive: true }, ]); - vi.mocked(db.query.upstreams.findMany).mockResolvedValueOnce([waitableUpstream]); vi.mocked(db.query.apiKeyUpstreams.findMany).mockResolvedValueOnce([ - { upstreamId: "up-queued" }, + { upstreamId: upstream.id }, ]); + vi.mocked(selectFromProviderType).mockResolvedValueOnce({ + upstream, + providerType: "openai", + selectedTier: 0, + circuitBreakerFiltered: 0, + totalCandidates: 1, + }); + mockGetEffectiveCircuitBreakerConfig.mockImplementationOnce(() => circuitConfigPromise); - vi.mocked(routeByModel).mockResolvedValueOnce({ - upstream: waitableUpstream, - providerType: "anthropic", - resolvedModel: "claude-test", - candidateUpstreams: [], - excludedUpstreams: [], - routingDecision: { - originalModel: "claude-test", - resolvedModel: "claude-test", - providerType: "anthropic", - upstreamName: "queued-upstream", - allowedModelsFilter: false, - modelRedirectApplied: false, + const responsePromise = POST( + new NextRequest("http://localhost/api/proxy/v1/chat/completions", { + method: "POST", + signal: controller.signal, + headers: { + authorization: "Bearer sk-test", + "content-type": "application/json", + }, + body: JSON.stringify({ + model: "gpt-5.2", + messages: [{ role: "user", content: "hi" }], + }), + }), + { params: Promise.resolve({ path: ["v1", "chat", "completions"] }) } + ); + + await expect.poll(() => mockGetEffectiveCircuitBreakerConfig.mock.calls.length).toBe(1); + controller.abort(); + + const response = await responsePromise; + const data = await response.json(); + resolveCircuitConfig(circuitConfig); + + expect(response.status).toBe(499); + expect(data.error).toEqual( + expect.objectContaining({ code: "CLIENT_DISCONNECTED", did_send_upstream: false }) + ); + expect(forwardRequest).not.toHaveBeenCalled(); + expect(vi.mocked(releaseConnection)).toHaveBeenCalledWith(upstream.id); + expect(calculateAndPersistRequestBillingSnapshot).not.toHaveBeenCalled(); + expect(recordTrafficFixture).not.toHaveBeenCalled(); + const abortLog = vi + .mocked(updateRequestLog) + .mock.calls.find( + ([, payload]) => payload?.routingDecision?.failure_stage === "candidate_selection" + )?.[1]; + expect(abortLog?.routingDecision).toEqual( + expect.objectContaining({ failure_stage: "candidate_selection", did_send_upstream: false }) + ); + }); + + it("releases a handed-off slot while queue state persistence is pending", async () => { + const { forwardWithFailover, ClientDisconnectedError } = + await import("@/app/api/proxy/v1/[...path]/proxy-execution"); + const { AllCandidatesConcurrencyFullError, releaseConnection } = + await import("@/lib/services/load-balancer"); + const { upstreamQueueAdmission } = await import("@/lib/services/upstream-queue-admission"); + + const waitableUpstream = { + ...DEFAULT_ACTIVE_UPSTREAMS[0], + id: "up-queued-pending-log", + name: "queued-pending-log", + providerType: "anthropic", + routeCapabilities: ["anthropic_messages"], + queuePolicy: { + enabled: true, + timeout_ms: 30000, + max_queue_length: 4, + }, + }; + const controller = new AbortController(); + let resolveQueueState!: () => void; + const queueStatePromise = new Promise((resolve) => { + resolveQueueState = resolve; + }); + let resolveResumeDecision!: (decision: { + action: "resume"; + upstream: typeof waitableUpstream; + }) => void; + const resumeDecisionPromise = new Promise<{ + action: "resume"; + upstream: typeof waitableUpstream; + }>((resolve) => { + resolveResumeDecision = resolve; + }); + + vi.mocked(mockSelectFromUpstreamCandidates).mockRejectedValueOnce( + new AllCandidatesConcurrencyFullError( + [ + { + upstreamId: waitableUpstream.id, + upstreamName: waitableUpstream.name, + upstreamBaseUrl: waitableUpstream.baseUrl, + upstreamProviderType: "anthropic", + tier: 0, + currentConcurrency: 1, + maxConcurrency: 1, + }, + ], + { + upstream: waitableUpstream, + tier: 0, + currentConcurrency: 1, + maxConcurrency: 1, + } + ) + ); + vi.mocked(upstreamQueueAdmission.enqueueWait).mockReturnValueOnce({ + accepted: true, + reason: "queued", + position: 1, + queueLength: 1, + waitPromise: Promise.resolve({ + upstreamId: waitableUpstream.id, + requestId: "req-queued-pending-log", + waitDurationMs: 25, + activeCount: 1, + queueLengthRemaining: 0, + }), + }); + vi.mocked(mockDecideQueuedUpstreamResume).mockReturnValueOnce(resumeDecisionPromise); + + const forwardPromise = forwardWithFailover({ + request: new Request("http://localhost/api/proxy/v1/messages", { + method: "POST", + signal: controller.signal, + body: JSON.stringify({ model: "claude-test", messages: [] }), + }), + routeCapability: "anthropic_messages", + path: "messages", + requestId: "req-queued-pending-log", + candidateUpstreamIds: [waitableUpstream.id], + requestModel: "claude-test", + affinityContext: null, + compensationHeaders: [], + onQueueStateChange: () => queueStatePromise, + }); + + await expect + .poll(() => vi.mocked(upstreamQueueAdmission.enqueueWait).mock.calls.length) + .toBe(1); + await expect.poll(() => vi.mocked(mockDecideQueuedUpstreamResume).mock.calls.length).toBe(1); + controller.abort(); + await expect + .poll(() => + vi + .mocked(releaseConnection) + .mock.calls.some(([upstreamId]) => upstreamId === waitableUpstream.id) + ) + .toBe(true); + + resolveQueueState(); + await expect(forwardPromise).rejects.toBeInstanceOf(ClientDisconnectedError); + resolveResumeDecision({ action: "resume", upstream: waitableUpstream }); + expect(vi.mocked(releaseConnection)).toHaveBeenCalledTimes(1); + }); + it("stops inbound body buffering when the client aborts before routing", async () => { + const { forwardWithFailover } = await import("@/app/api/proxy/v1/[...path]/proxy-execution"); + const controller = new AbortController(); + let resolveBody!: (body: ArrayBuffer) => void; + const pendingBody = new Promise((resolve) => { + resolveBody = resolve; + }); + const request = { + url: "http://localhost/api/proxy/v1/messages", + method: "POST", + headers: new Headers({ "content-type": "application/json" }), + signal: controller.signal, + clone: () => ({ arrayBuffer: () => pendingBody }), + } as unknown as Request; + + const forwardPromise = forwardWithFailover({ + request, + routeCapability: "anthropic_messages", + path: "messages", + requestId: "req-body-buffer-abort", + candidateUpstreamIds: ["upstream-never-selected"], + requestModel: "claude-test", + affinityContext: null, + compensationHeaders: [], + }); + + controller.abort(); + await expect(forwardPromise).rejects.toMatchObject({ + name: "ClientDisconnectedError", + failureStage: "candidate_selection", + didSendUpstream: false, + }); + expect(mockSelectFromUpstreamCandidates).not.toHaveBeenCalled(); + resolveBody(new ArrayBuffer(0)); + }); + + it("marks a resumed queue as aborted when cancellation interrupts pre-dispatch setup", async () => { + const { forwardWithFailover } = await import("@/app/api/proxy/v1/[...path]/proxy-execution"); + const { AllCandidatesConcurrencyFullError, decideQueuedUpstreamResume, releaseConnection } = + await import("@/lib/services/load-balancer"); + const { upstreamQueueAdmission } = await import("@/lib/services/upstream-queue-admission"); + const { forwardRequest } = await import("@/lib/services/proxy-client"); + + const waitableUpstream = { + ...DEFAULT_ACTIVE_UPSTREAMS[0], + id: "up-queued-resumed-abort", + name: "queued-resumed-abort", + providerType: "anthropic", + routeCapabilities: ["anthropic_messages"], + queuePolicy: { + enabled: true, + timeout_ms: 30000, + max_queue_length: 4, + }, + }; + const controller = new AbortController(); + let resolveCircuitConfig!: (value: { + failureThreshold: number; + successThreshold: number; + openDuration: number; + probeInterval: number; + firstByteTimeout: number; + streamIdleTimeout: number; + }) => void; + const circuitConfigPromise = new Promise<{ + failureThreshold: number; + successThreshold: number; + openDuration: number; + probeInterval: number; + firstByteTimeout: number; + streamIdleTimeout: number; + }>((resolve) => { + resolveCircuitConfig = resolve; + }); + + vi.mocked(mockSelectFromUpstreamCandidates).mockRejectedValueOnce( + new AllCandidatesConcurrencyFullError( + [ + { + upstreamId: waitableUpstream.id, + upstreamName: waitableUpstream.name, + upstreamBaseUrl: waitableUpstream.baseUrl, + upstreamProviderType: "anthropic", + tier: 0, + currentConcurrency: 1, + maxConcurrency: 1, + }, + ], + { + upstream: waitableUpstream, + tier: 0, + currentConcurrency: 1, + maxConcurrency: 1, + } + ) + ); + vi.mocked(upstreamQueueAdmission.enqueueWait).mockReturnValueOnce({ + accepted: true, + reason: "queued", + position: 1, + queueLength: 1, + waitPromise: Promise.resolve({ + upstreamId: waitableUpstream.id, + requestId: "req-queued-resumed-abort", + waitDurationMs: 25, + activeCount: 1, + queueLengthRemaining: 0, + }), + }); + vi.mocked(decideQueuedUpstreamResume).mockResolvedValueOnce({ + action: "resume", + reason: "bound_available", + upstream: waitableUpstream, + excludeIds: [], + }); + vi.mocked(mockGetEffectiveCircuitBreakerConfig).mockReturnValueOnce(circuitConfigPromise); + + const forwardPromise = forwardWithFailover({ + request: new Request("http://localhost/api/proxy/v1/messages", { + method: "POST", + signal: controller.signal, + body: JSON.stringify({ model: "claude-test", messages: [] }), + }), + routeCapability: "anthropic_messages", + path: "messages", + requestId: "req-queued-resumed-abort", + candidateUpstreamIds: [waitableUpstream.id], + requestModel: "claude-test", + affinityContext: null, + compensationHeaders: [], + }); + + await expect + .poll(() => vi.mocked(mockGetEffectiveCircuitBreakerConfig).mock.calls.length) + .toBe(1); + controller.abort(); + await expect + .poll(() => + vi + .mocked(releaseConnection) + .mock.calls.some(([upstreamId]) => upstreamId === waitableUpstream.id) + ) + .toBe(true); + resolveCircuitConfig({ + failureThreshold: 5, + successThreshold: 2, + openDuration: 300000, + probeInterval: 30000, + firstByteTimeout: 30000, + streamIdleTimeout: 60000, + }); + + await expect(forwardPromise).rejects.toMatchObject({ + name: "ClientDisconnectedError", + queue: expect.objectContaining({ status: "aborted" }), + }); + expect(vi.mocked(forwardRequest)).not.toHaveBeenCalled(); + }); + + it("classifies queue cancellation as candidate selection after a prior upstream attempt", async () => { + const { resolveFailureStage } = await import("@/app/api/proxy/v1/[...path]/proxy-execution"); + const { UpstreamQueueWaitAbortedError } = + await import("@/lib/services/upstream-queue-admission"); + + const error = new UpstreamQueueWaitAbortedError("up-queued", "req-queued", 25); + + expect(resolveFailureStage(error, true, undefined)).toBe("candidate_selection"); + }); + + it("should release the resumed slot and reselect once when the queued upstream disappears", async () => { + const { db } = await import("@/lib/db"); + const { forwardRequest } = await import("@/lib/services/proxy-client"); + const { routeByModel } = await import("@/lib/services/model-router"); + const { + selectFromProviderType, + decideQueuedUpstreamResume, + reselectQueuedUpstreamOnce, + AllCandidatesConcurrencyFullError, + releaseConnection, + } = await import("@/lib/services/load-balancer"); + const { upstreamQueueAdmission } = await import("@/lib/services/upstream-queue-admission"); + + const waitableUpstream = { + ...DEFAULT_ACTIVE_UPSTREAMS[0], + id: "up-queued", + name: "queued-upstream", + providerType: "anthropic", + routeCapabilities: ["anthropic_messages"], + baseUrl: "https://api.anthropic.com", + queuePolicy: { + enabled: true, + timeout_ms: 30000, + max_queue_length: 4, + }, + }; + const fallbackUpstream = { + ...DEFAULT_ACTIVE_UPSTREAMS[1], + id: "up-fallback", + name: "fallback-upstream", + providerType: "anthropic", + routeCapabilities: ["anthropic_messages"], + baseUrl: "https://api.anthropic.com", + }; + + vi.mocked(db.query.apiKeys.findMany).mockResolvedValueOnce([ + { id: "key-1", keyHash: "hash-1", expiresAt: null, isActive: true }, + ]); + vi.mocked(db.query.upstreams.findMany).mockResolvedValueOnce([ + waitableUpstream, + fallbackUpstream, + ]); + vi.mocked(db.query.apiKeyUpstreams.findMany).mockResolvedValueOnce([ + { upstreamId: "up-queued" }, + { upstreamId: "up-fallback" }, + ]); + + vi.mocked(routeByModel).mockResolvedValueOnce({ + upstream: waitableUpstream, + providerType: "anthropic", + resolvedModel: "claude-test", + candidateUpstreams: [], + excludedUpstreams: [], + routingDecision: { + originalModel: "claude-test", + resolvedModel: "claude-test", + providerType: "anthropic", + upstreamName: "queued-upstream", + allowedModelsFilter: false, + modelRedirectApplied: false, + circuitBreakerFilter: false, + routingType: "provider_type", + candidateCount: 2, + finalCandidateCount: 0, + }, + }); + + vi.mocked(selectFromProviderType).mockRejectedValueOnce( + new AllCandidatesConcurrencyFullError( + [ + { + upstreamId: "up-queued", + upstreamName: "queued-upstream", + upstreamBaseUrl: "https://api.anthropic.com", + upstreamProviderType: "anthropic", + tier: 0, + currentConcurrency: 1, + maxConcurrency: 1, + }, + ], + { + upstream: waitableUpstream, + tier: 0, + currentConcurrency: 1, + maxConcurrency: 1, + } + ) + ); + vi.mocked(upstreamQueueAdmission.enqueueWait).mockReturnValueOnce({ + accepted: true, + reason: "queued", + position: 1, + queueLength: 1, + waitPromise: Promise.resolve({ + upstreamId: "up-queued", + requestId: "req-queued", + waitDurationMs: 25, + activeCount: 1, + queueLengthRemaining: 0, + }), + }); + vi.mocked(decideQueuedUpstreamResume).mockResolvedValueOnce({ + action: "reselect_once", + reason: "bound_missing", + upstream: null, + excludeIds: ["up-queued"], + }); + vi.mocked(reselectQueuedUpstreamOnce).mockResolvedValueOnce({ + upstream: fallbackUpstream, + selectedTier: 1, + circuitBreakerFiltered: 0, + quotaFiltered: 0, + concurrencyFiltered: 0, + concurrencyExcluded: [], + totalCandidates: 2, + affinityHit: false, + affinityMigrated: false, + selectionReason: null, + }); + vi.mocked(forwardRequest).mockResolvedValueOnce({ + statusCode: 200, + headers: new Headers(), + body: new Uint8Array(), + isStream: false, + usage: null, + headerDiff: null, + }); + + const request = new NextRequest("http://localhost/api/proxy/v1/messages", { + method: "POST", + headers: { + authorization: "Bearer sk-test", + "content-type": "application/json", + }, + body: JSON.stringify({ + model: "claude-test", + messages: [{ role: "user", content: "hi" }], + }), + }); + + const response = await handleProxy(request, { + params: Promise.resolve({ path: ["v1", "messages"] }), + }); + + expect(response.status).toBe(200); + expect(vi.mocked(reselectQueuedUpstreamOnce)).toHaveBeenCalledWith( + "up-queued", + ["up-queued", "up-fallback"], + ["up-queued"], + expect.objectContaining({ candidateSnapshot: expect.any(Array) }) + ); + expect(vi.mocked(releaseConnection).mock.calls.map(([upstreamId]) => upstreamId)).toEqual([ + "up-queued", + "up-fallback", + ]); + }); + + it("should classify queue wait timeout separately from upstream timeout", async () => { + const { db } = await import("@/lib/db"); + const { forwardRequest } = await import("@/lib/services/proxy-client"); + const { routeByModel } = await import("@/lib/services/model-router"); + const { updateRequestLog } = await import("@/lib/services/request-logger"); + const { markUnhealthy } = await import("@/lib/services/health-checker"); + const { recordFailure } = await import("@/lib/services/circuit-breaker"); + const { selectFromProviderType, AllCandidatesConcurrencyFullError } = + await import("@/lib/services/load-balancer"); + const { upstreamQueueAdmission, UpstreamQueueWaitTimeoutError } = + await import("@/lib/services/upstream-queue-admission"); + const { calculateAndPersistRequestBillingSnapshot } = + await import("@/lib/services/billing-cost-service"); + const { recordTrafficFixture } = await import("@/lib/services/traffic-recorder"); + process.env.RECORDER_ENABLED = "true"; + process.env.RECORDER_MODE = "all"; + + const waitableUpstream = { + ...DEFAULT_ACTIVE_UPSTREAMS[0], + id: "up-queued", + name: "queued-upstream", + providerType: "anthropic", + routeCapabilities: ["anthropic_messages"], + baseUrl: "https://api.anthropic.com", + queuePolicy: { + enabled: true, + timeout_ms: 30000, + max_queue_length: 4, + }, + }; + + vi.mocked(db.query.apiKeys.findMany).mockResolvedValueOnce([ + { id: "key-1", keyHash: "hash-1", expiresAt: null, isActive: true }, + ]); + vi.mocked(db.query.upstreams.findMany).mockResolvedValueOnce([waitableUpstream]); + vi.mocked(db.query.apiKeyUpstreams.findMany).mockResolvedValueOnce([ + { upstreamId: "up-queued" }, + ]); + + vi.mocked(routeByModel).mockResolvedValueOnce({ + upstream: waitableUpstream, + providerType: "anthropic", + resolvedModel: "claude-test", + candidateUpstreams: [], + excludedUpstreams: [], + routingDecision: { + originalModel: "claude-test", + resolvedModel: "claude-test", + providerType: "anthropic", + upstreamName: "queued-upstream", + allowedModelsFilter: false, + modelRedirectApplied: false, circuitBreakerFilter: false, routingType: "provider_type", candidateCount: 1, @@ -4256,7 +4819,9 @@ describe("proxy route upstream selection", () => { }), }); - const response = await POST(request, { params: Promise.resolve({ path: ["v1", "messages"] }) }); + const response = await handleProxy(request, { + params: Promise.resolve({ path: ["v1", "messages"] }), + }); const data = await response.json(); expect(response.status).toBe(504); @@ -4271,6 +4836,8 @@ describe("proxy route upstream selection", () => { expect(forwardRequest).not.toHaveBeenCalled(); expect(markUnhealthy).not.toHaveBeenCalled(); expect(recordFailure).not.toHaveBeenCalled(); + expect(calculateAndPersistRequestBillingSnapshot).not.toHaveBeenCalled(); + expect(recordTrafficFixture).not.toHaveBeenCalled(); expect( vi .mocked(updateRequestLog) @@ -4304,6 +4871,11 @@ describe("proxy route upstream selection", () => { await import("@/lib/services/load-balancer"); const { upstreamQueueAdmission, UpstreamQueueWaitAbortedError } = await import("@/lib/services/upstream-queue-admission"); + const { calculateAndPersistRequestBillingSnapshot } = + await import("@/lib/services/billing-cost-service"); + const { recordTrafficFixture } = await import("@/lib/services/traffic-recorder"); + process.env.RECORDER_ENABLED = "true"; + process.env.RECORDER_MODE = "all"; const waitableUpstream = { ...DEFAULT_ACTIVE_UPSTREAMS[0], @@ -4318,6 +4890,17 @@ describe("proxy route upstream selection", () => { max_queue_length: 4, }, }; + const controller = new AbortController(); + let rejectQueueWait!: (reason?: unknown) => void; + const queueWaitPromise = new Promise<{ + upstreamId: string; + requestId: string; + waitDurationMs: number; + activeCount: number; + queueLengthRemaining: number; + }>((_, reject) => { + rejectQueueWait = reject; + }); vi.mocked(db.query.apiKeys.findMany).mockResolvedValueOnce([ { id: "key-1", keyHash: "hash-1", expiresAt: null, isActive: true }, @@ -4372,24 +4955,32 @@ describe("proxy route upstream selection", () => { reason: "queued", position: 1, queueLength: 1, - waitPromise: Promise.reject( - new UpstreamQueueWaitAbortedError("up-queued", "req-queued", 1200) - ), + waitPromise: queueWaitPromise, }); - const request = new NextRequest("http://localhost/api/proxy/v1/messages", { - method: "POST", - headers: { - authorization: "Bearer sk-test", - "content-type": "application/json", - }, - body: JSON.stringify({ - model: "claude-test", - messages: [{ role: "user", content: "hi" }], + const responsePromise = handleProxy( + new NextRequest("http://localhost/api/proxy/v1/messages", { + method: "POST", + signal: controller.signal, + headers: { + authorization: "Bearer sk-test", + "content-type": "application/json", + }, + body: JSON.stringify({ + model: "claude-test", + messages: [{ role: "user", content: "hi" }], + }), }), - }); + { params: Promise.resolve({ path: ["v1", "messages"] }) } + ); + await expect + .poll(() => vi.mocked(upstreamQueueAdmission.enqueueWait).mock.calls.length) + .toBe(1); + controller.abort(); + rejectQueueWait(new UpstreamQueueWaitAbortedError("up-queued", "req-queued", 1200)); + + const response = await responsePromise; - const response = await POST(request, { params: Promise.resolve({ path: ["v1", "messages"] }) }); const data = await response.json(); expect(response.status).toBe(499); @@ -4404,6 +4995,8 @@ describe("proxy route upstream selection", () => { expect(forwardRequest).not.toHaveBeenCalled(); expect(markUnhealthy).not.toHaveBeenCalled(); expect(recordFailure).not.toHaveBeenCalled(); + expect(calculateAndPersistRequestBillingSnapshot).not.toHaveBeenCalled(); + expect(recordTrafficFixture).not.toHaveBeenCalled(); expect( vi .mocked(updateRequestLog) @@ -5651,7 +6244,7 @@ describe("proxy route upstream selection", () => { expect(recordTrafficFixture).toHaveBeenCalledTimes(1); }); - it("should not inject upstream auth headers when request was never sent upstream", async () => { + it("should not record or bill a request that was never sent upstream", async () => { process.env.RECORDER_ENABLED = "true"; const { db } = await import("@/lib/db"); @@ -5660,6 +6253,8 @@ describe("proxy route upstream selection", () => { const { selectFromProviderType, NoAuthorizedUpstreamsError } = await import("@/lib/services/load-balancer"); const { buildFixture, recordTrafficFixture } = await import("@/lib/services/traffic-recorder"); + const { calculateAndPersistRequestBillingSnapshot } = + await import("@/lib/services/billing-cost-service"); const routedUpstream = { id: "up-route", @@ -5729,17 +6324,138 @@ describe("proxy route upstream selection", () => { expect(injectAuthHeader).not.toHaveBeenCalled(); expect(db.query.upstreams.findFirst).not.toHaveBeenCalled(); - expect(buildFixture).toHaveBeenCalledTimes(1); - const fixtureParams = vi.mocked(buildFixture).mock.calls[0][0]; - expect(fixtureParams.upstream).toEqual( + expect(buildFixture).not.toHaveBeenCalled(); + expect(recordTrafficFixture).not.toHaveBeenCalled(); + expect(calculateAndPersistRequestBillingSnapshot).not.toHaveBeenCalled(); + }); + it("logs body preparation rejection without billing or recording", async () => { + process.env.RECORDER_ENABLED = "true"; + + const { db } = await import("@/lib/db"); + const { forwardRequest } = await import("@/lib/services/proxy-client"); + const { selectFromProviderType, releaseConnection } = + await import("@/lib/services/load-balancer"); + const { markUnhealthy } = await import("@/lib/services/health-checker"); + const { recordFailure } = await import("@/lib/services/circuit-breaker"); + const { updateRequestLog } = await import("@/lib/services/request-logger"); + const { buildFixture, recordTrafficFixture } = await import("@/lib/services/traffic-recorder"); + const { calculateAndPersistRequestBillingSnapshot } = + await import("@/lib/services/billing-cost-service"); + + const upstream = { + ...DEFAULT_ACTIVE_UPSTREAMS[0], + id: "up-openai", + name: "body-failure", + }; + vi.mocked(db.query.apiKeys.findMany).mockResolvedValueOnce([ + { id: "key-body-failure", keyHash: "hash-1", expiresAt: null, isActive: true }, + ]); + vi.mocked(db.query.apiKeyUpstreams.findMany).mockResolvedValueOnce([ + { upstreamId: upstream.id }, + ]); + vi.mocked(selectFromProviderType).mockResolvedValueOnce({ + upstream, + providerType: "openai", + selectedTier: 0, + circuitBreakerFiltered: 0, + totalCandidates: 1, + }); + vi.mocked(forwardRequest).mockRejectedValueOnce( + Object.assign(new Error("fetch failed while reading request body"), { + proxyRequestMetadata: { fetchStarted: false }, + }) + ); + + const response = await POST( + new NextRequest("http://localhost/api/proxy/v1/chat/completions", { + method: "POST", + headers: { + authorization: "Bearer sk-body-failure", + "content-type": "application/json", + }, + body: JSON.stringify({ + model: "gpt-5.2", + messages: [{ role: "user", content: "hi" }], + }), + }), + { params: Promise.resolve({ path: ["v1", "chat", "completions"] }) } + ); + const data = await response.json(); + + expect(response.status).toBe(503); + expect(data.error).toEqual( expect.objectContaining({ - id: "unknown", - name: "not-sent", + code: "SERVICE_UNAVAILABLE", + reason: "NO_HEALTHY_CANDIDATES", + did_send_upstream: false, + }) + ); + expect(forwardRequest).toHaveBeenCalledTimes(1); + expect(vi.mocked(releaseConnection)).toHaveBeenCalledWith(upstream.id); + expect(markUnhealthy).not.toHaveBeenCalled(); + expect(recordFailure).not.toHaveBeenCalled(); + expect(buildFixture).not.toHaveBeenCalled(); + expect(recordTrafficFixture).not.toHaveBeenCalled(); + expect(calculateAndPersistRequestBillingSnapshot).not.toHaveBeenCalled(); + + const failureLog = vi.mocked(updateRequestLog).mock.calls.at(-1)?.[1] as { + durationMs?: number; + failoverAttempts?: number; + routingDecision?: Record; + }; + expect(failureLog.failoverAttempts).toBe(0); + expect(failureLog.durationMs).toEqual(expect.any(Number)); + expect(failureLog.routingDecision).toEqual( + expect.objectContaining({ + failure_stage: "candidate_selection", + did_send_upstream: false, + actual_upstream_id: null, + }) + ); + }); + it("logs active upstream snapshot rejection without dispatch", async () => { + const { db } = await import("@/lib/db"); + const { forwardRequest } = await import("@/lib/services/proxy-client"); + const { logRequest } = await import("@/lib/services/request-logger"); + const { calculateAndPersistRequestBillingSnapshot } = + await import("@/lib/services/billing-cost-service"); + + vi.mocked(db.query.apiKeys.findMany).mockResolvedValueOnce([ + { id: "key-snapshot-failure", keyHash: "hash-1", expiresAt: null, isActive: true }, + ]); + vi.mocked(db.query.upstreams.findMany).mockRejectedValueOnce( + new Error("upstream snapshot unavailable") + ); + + const response = await POST( + new NextRequest("http://localhost/api/proxy/v1/chat/completions", { + method: "POST", + headers: { + authorization: "Bearer sk-test", + "content-type": "application/json", + }, + body: JSON.stringify({ + model: "gpt-5.2", + messages: [{ role: "user", content: "hi" }], + }), + }), + { params: Promise.resolve({ path: ["v1", "chat", "completions"] }) } + ); + + expect(response.status).toBe(503); + expect(forwardRequest).not.toHaveBeenCalled(); + expect(calculateAndPersistRequestBillingSnapshot).not.toHaveBeenCalled(); + expect(logRequest).toHaveBeenCalledWith( + expect.objectContaining({ + statusCode: 503, + errorMessage: "failed to load active upstream snapshot", + durationMs: expect.any(Number), + routingDecision: expect.objectContaining({ + failure_stage: "candidate_selection", + did_send_upstream: false, + }), }) ); - expect(fixtureParams.outboundHeaders).toEqual({}); - expect(fixtureParams.outboundRequestSent).toBe(false); - expect(recordTrafficFixture).toHaveBeenCalledTimes(1); }); it("should classify downstream disconnect as CLIENT_DISCONNECTED reason", async () => { @@ -6961,12 +7677,7 @@ describe("proxy route upstream selection", () => { }); expect(data.error.request_id).toEqual(expect.any(String)); expect(forwardRequest).not.toHaveBeenCalled(); - expect(calculateAndPersistRequestBillingSnapshot).toHaveBeenCalledWith( - expect.objectContaining({ - requestedServiceTier: "fast", - effectiveServiceTier: null, - }) - ); + expect(calculateAndPersistRequestBillingSnapshot).not.toHaveBeenCalled(); }); it("should authorize unrestricted keys across all active upstreams", async () => { diff --git a/tests/unit/services/proxy-client.test.ts b/tests/unit/services/proxy-client.test.ts index bab10796..2ca47f75 100644 --- a/tests/unit/services/proxy-client.test.ts +++ b/tests/unit/services/proxy-client.test.ts @@ -15,10 +15,11 @@ import type { Upstream } from "@/lib/db"; const loggerSpies = vi.hoisted(() => { const info = vi.fn(); const debug = vi.fn(); + const warn = vi.fn(); const error = vi.fn(); - const child = vi.fn(() => ({ info, debug, error })); + const child = vi.fn(() => ({ info, debug, warn, error })); const createLogger = vi.fn(() => ({ child })); - return { info, debug, error, child, createLogger }; + return { info, debug, warn, error, child, createLogger }; }); vi.mock("@/lib/utils/logger", () => ({ @@ -1556,6 +1557,153 @@ describe("proxy-client", () => { }) ); }); + it("reports dispatch after the upstream fetch is invoked", async () => { + const mockResponse = new Response(JSON.stringify({ id: "dispatch" }), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + const fetchMock = vi.fn().mockResolvedValue(mockResponse); + global.fetch = fetchMock; + const onDispatchStart = vi.fn(); + const request = new Request("http://localhost/api", { + method: "POST", + body: JSON.stringify({ model: "gpt-4" }), + }); + + await forwardRequest( + request, + mockUpstream, + "chat/completions", + "req-dispatch", + undefined, + undefined, + onDispatchStart + ); + + expect(onDispatchStart).toHaveBeenCalledOnce(); + expect(onDispatchStart.mock.invocationCallOrder[0]).toBeGreaterThan( + fetchMock.mock.invocationCallOrder[0]! + ); + }); + it("does not report dispatch when fetch invocation throws synchronously", async () => { + const fetchMock = vi.fn(() => { + throw new Error("fetch invocation failed"); + }); + global.fetch = fetchMock; + const onDispatchStart = vi.fn(); + const request = new Request("http://localhost/api", { + method: "POST", + body: JSON.stringify({ model: "gpt-4" }), + }); + + await expect( + forwardRequest( + request, + mockUpstream, + "chat/completions", + "req-sync-fetch-failure", + undefined, + undefined, + onDispatchStart + ) + ).rejects.toMatchObject({ + message: "fetch invocation failed", + proxyRequestMetadata: { fetchStarted: false }, + }); + expect(fetchMock).toHaveBeenCalledOnce(); + expect(onDispatchStart).not.toHaveBeenCalled(); + }); + + it("marks request body failures as not dispatched", async () => { + const fetchMock = vi.fn(); + global.fetch = fetchMock; + const bodyReadError = new Error("request body read failed"); + const request = new Request("http://localhost/api", { + method: "POST", + body: JSON.stringify({ model: "gpt-4" }), + }); + vi.spyOn(request, "arrayBuffer").mockRejectedValueOnce(bodyReadError); + + await expect( + forwardRequest(request, mockUpstream, "chat/completions", "req-body-read") + ).rejects.toMatchObject({ + message: "request body read failed", + proxyRequestMetadata: { fetchStarted: false }, + }); + expect(fetchMock).not.toHaveBeenCalled(); + }); + it("does not fetch after the downstream request is already aborted", async () => { + const fetchMock = vi.fn(); + global.fetch = fetchMock; + const controller = new AbortController(); + const request = new Request("http://localhost/api", { + method: "POST", + body: JSON.stringify({ model: "gpt-4" }), + signal: controller.signal, + }); + controller.abort(); + const onDispatchStart = vi.fn(); + + await expect( + forwardRequest( + request, + mockUpstream, + "chat/completions", + "req-already-aborted", + undefined, + undefined, + onDispatchStart + ) + ).rejects.toThrow("Upstream request cancelled by downstream client"); + expect(fetchMock).not.toHaveBeenCalled(); + expect(onDispatchStart).not.toHaveBeenCalled(); + }); + + it("should abort the upstream fetch when the downstream request is aborted", async () => { + let resolveFetchStarted!: () => void; + const fetchStarted = new Promise((resolve) => { + resolveFetchStarted = resolve; + }); + + global.fetch = vi.fn().mockImplementation((_url: string, options: RequestInit) => { + resolveFetchStarted(); + return new Promise((_resolve, reject) => { + options.signal?.addEventListener( + "abort", + () => { + const error = new Error("Aborted"); + error.name = "AbortError"; + reject(error); + }, + { once: true } + ); + }); + }); + + const controller = new AbortController(); + const request = new Request("http://localhost/api", { + method: "POST", + body: JSON.stringify({ model: "gpt-4" }), + signal: controller.signal, + }); + + const forwardPromise = forwardRequest(request, mockUpstream, "chat/completions", "req-123"); + await fetchStarted; + + controller.abort(); + + await expect(forwardPromise).rejects.toThrow( + "Upstream request cancelled by downstream client" + ); + expect(loggerSpies.warn).toHaveBeenCalledWith( + { upstream: mockUpstream.name }, + "upstream request cancelled by downstream client" + ); + expect(loggerSpies.error).not.toHaveBeenCalledWith( + { timeout: mockUpstream.timeout }, + "upstream request timed out" + ); + }); it("should preserve original request query string when forwarding", async () => { const mockResponse = new Response(JSON.stringify({ id: "123" }), {