From 99e669000e1b63a5f3eb0ea8fbfa52f88618ddca Mon Sep 17 00:00:00 2001 From: Lukin Date: Tue, 7 Jul 2026 00:12:07 +0800 Subject: [PATCH] Fix timeout telemetry truth Co-Authored-By: Codex --- apps/gateway/src/app.test.ts | 26 +++++++++++++++++++ apps/gateway/src/app.ts | 8 +++++- apps/gateway/src/middleware/limits.ts | 25 ++++++++++++++++-- apps/gateway/src/routes/chat.ts | 20 ++++++++++---- apps/gateway/src/routes/gemini.ts | 12 ++++++--- apps/gateway/src/routes/images.ts | 11 +++++--- apps/gateway/src/routes/interactions.ts | 9 ++++--- apps/gateway/src/routes/messages.ts | 12 ++++++--- .../src/routes/payload-capture.test.ts | 19 ++++++++++++++ apps/gateway/src/routes/payload-capture.ts | 24 ++++++++++++++--- apps/gateway/src/routes/responses.ts | 22 +++++++++------- implementation-notes.md | 19 +++++++------- package.json | 2 +- 13 files changed, 163 insertions(+), 46 deletions(-) diff --git a/apps/gateway/src/app.test.ts b/apps/gateway/src/app.test.ts index 8efbd4d5..d1483432 100644 --- a/apps/gateway/src/app.test.ts +++ b/apps/gateway/src/app.test.ts @@ -5,6 +5,7 @@ import type { AppEnv } from "./app.js"; import { createApp } from "./app.js"; import type { LogFields, Logger, LogLevel } from "./logging.js"; import { HelmHttpError, handleError, openAIErrorEnvelope } from "./middleware/error-handler.js"; +import { requestTimedOut } from "./middleware/limits.js"; interface Captured { level: LogLevel; @@ -126,6 +127,31 @@ describe("createApp: trace_id, logging, error handling", () => { expect(lines.some((l) => l.level === "error")).toBe(true); }); + it("marks the request context as timed out while late route work continues", async () => { + const { logger } = fakeLogger(); + const app = createApp({ + logger, + limits: { maxBodyBytes: 1024, requestTimeoutMs: 5 }, + genTraceId: () => "trace-timeout", + }); + let lateTimedOut: boolean | null = null; + let lateDone: (() => void) | null = null; + const late = new Promise((resolve) => { + lateDone = resolve; + }); + app.get("/slow", async (c) => { + await new Promise((resolve) => setTimeout(resolve, 25)); + lateTimedOut = requestTimedOut(c); + lateDone?.(); + return c.text("late"); + }); + + const res = await app.request("/slow"); + expect(res.status).toBe(504); + await late; + expect(lateTimedOut).toBe(true); + }); + it("does not leak Authorization header into logs or error bodies", async () => { const { logger, lines } = fakeLogger(); const app = createApp({ logger }); diff --git a/apps/gateway/src/app.ts b/apps/gateway/src/app.ts index a06f5df2..3c983a18 100644 --- a/apps/gateway/src/app.ts +++ b/apps/gateway/src/app.ts @@ -3,7 +3,12 @@ import { Hono } from "hono"; import { readBuildInfo } from "./build-info.js"; import type { Logger } from "./logging.js"; import { handleError } from "./middleware/error-handler.js"; -import { bodyLimit, type LimitsConfig, timeout } from "./middleware/limits.js"; +import { + bodyLimit, + type LimitsConfig, + type RequestTimeoutState, + timeout, +} from "./middleware/limits.js"; import { normalizeHeaders } from "./middleware/normalize-headers.js"; import { requestLoggerMiddleware } from "./middleware/request-logger.js"; import { traceIdMiddleware } from "./middleware/trace-id.js"; @@ -26,6 +31,7 @@ export type AppEnv = { Variables: { trace_id: string; logger: Logger; + request_timeout?: RequestTimeoutState; }; }; diff --git a/apps/gateway/src/middleware/limits.ts b/apps/gateway/src/middleware/limits.ts index 5f4bc8ac..75edf58b 100644 --- a/apps/gateway/src/middleware/limits.ts +++ b/apps/gateway/src/middleware/limits.ts @@ -1,5 +1,5 @@ import { makeHelmError } from "@helm/shared"; -import type { MiddlewareHandler } from "hono"; +import type { Context, MiddlewareHandler } from "hono"; import type { AppEnv } from "../app.js"; import { HelmHttpError } from "./error-handler.js"; @@ -8,6 +8,19 @@ export interface LimitsConfig { requestTimeoutMs: number; } +export interface RequestTimeoutState { + signal: AbortSignal; + timedOut: boolean; +} + +export function requestSignal(c: Context): AbortSignal { + return c.get("request_timeout")?.signal ?? c.req.raw.signal; +} + +export function requestTimedOut(c: Context): boolean { + return c.get("request_timeout")?.timedOut === true; +} + function tooLarge(traceId: string): HelmHttpError { return new HelmHttpError( makeHelmError({ @@ -51,7 +64,15 @@ export function timeout(cfg: LimitsConfig): MiddlewareHandler { return async (c, next) => { const traceId = c.get("trace_id"); const controller = new AbortController(); - const timer = setTimeout(() => controller.abort(), cfg.requestTimeoutMs); + const state: RequestTimeoutState = { + signal: AbortSignal.any([c.req.raw.signal, controller.signal]), + timedOut: false, + }; + c.set("request_timeout", state); + const timer = setTimeout(() => { + state.timedOut = true; + controller.abort(); + }, cfg.requestTimeoutMs); try { await Promise.race([ next(), diff --git a/apps/gateway/src/routes/chat.ts b/apps/gateway/src/routes/chat.ts index 0d761963..085d1214 100644 --- a/apps/gateway/src/routes/chat.ts +++ b/apps/gateway/src/routes/chat.ts @@ -34,6 +34,7 @@ import { streamSSE } from "hono/streaming"; import type { AppEnv } from "../app.js"; import { INTERNAL_API_KEY_ID } from "../internal-key.js"; import { HelmHttpError } from "../middleware/error-handler.js"; +import { requestSignal, requestTimedOut } from "../middleware/limits.js"; import { type ServingAccount, stampServingAccount } from "../runtime/serving-account.js"; import type { WriteQueue } from "../runtime/write-queue.js"; import { downgradeClientFastModeIfDisallowed } from "./fast-mode.js"; @@ -45,6 +46,7 @@ import { captureEnabled, createSseCapture, createStreamGenerationTimer, + decisionForTimedOutRequest, type PayloadCaptureDeps, persistPayload, tokensFromUsage, @@ -492,8 +494,9 @@ export function registerChatRoutes(app: Hono, deps: ChatRouteDeps): void // affected by anything that touches the decision after the response returns. // With a write queue wired, the insert is deferred + batched off the hot path. const persist = async (decision: DecisionRecord) => { + const decisionSnapshot = requestTimedOut(c) ? decisionForTimedOutRequest(decision) : decision; const input: InsertTelemetryInput = { - decision: deps.redact(decision) as DecisionRecord, + decision: deps.redact(decisionSnapshot) as DecisionRecord, apiKeyId: identity.keyId, createdAt: new Date(), }; @@ -517,12 +520,13 @@ export function registerChatRoutes(app: Hono, deps: ChatRouteDeps): void responseJson: string | null, upstreamRequestJson: string | null, ) => { + const capturedResponseJson = requestTimedOut(c) ? null : responseJson; if (deps.writes !== undefined) { if (!captureEnabled(deps)) return; deps.writes.enqueuePayload({ requestId: traceId, requestJson, - responseJson, + responseJson: capturedResponseJson, upstreamRequestJson, createdAt: new Date(deps.now()), }); @@ -530,7 +534,13 @@ export function registerChatRoutes(app: Hono, deps: ChatRouteDeps): void } await persistPayload( deps, - { requestId: traceId, requestJson, responseJson, upstreamRequestJson, now: deps.now() }, + { + requestId: traceId, + requestJson, + responseJson: capturedResponseJson, + upstreamRequestJson, + now: deps.now(), + }, (msg) => c.get("logger").log("warn", msg, { trace_id: traceId }), ); }; @@ -720,7 +730,7 @@ export function registerChatRoutes(app: Hono, deps: ChatRouteDeps): void blockedModels: identity.caps?.blockedModels ?? null, }, }, - c.req.raw.signal, + requestSignal(c), classifyOverrides, ); // The subscription the pool selected (null for a configured/non-OAuth provider), @@ -805,7 +815,7 @@ export function registerChatRoutes(app: Hono, deps: ChatRouteDeps): void try { for await (const item of withHeartbeat(stream, { heartbeatMs, - signal: c.req.raw.signal, + signal: requestSignal(c), })) { if (item.type === "beat") { if (atEventBoundary(lastWrite)) await sse.write(HEARTBEAT_COMMENT); diff --git a/apps/gateway/src/routes/gemini.ts b/apps/gateway/src/routes/gemini.ts index 3947871d..0683cde9 100644 --- a/apps/gateway/src/routes/gemini.ts +++ b/apps/gateway/src/routes/gemini.ts @@ -9,6 +9,7 @@ import { streamSSE } from "hono/streaming"; import type { AppEnv } from "../app.js"; import { type ConcurrencyGatePort, concurrencyReleaseGuard } from "../middleware/concurrency.js"; import { estimateRequestTokens } from "../middleware/estimate-tokens.js"; +import { requestSignal, requestTimedOut } from "../middleware/limits.js"; import { stampServingAccount } from "../runtime/serving-account.js"; import { atEventBoundary, HEARTBEAT_COMMENT, withHeartbeat } from "./heartbeat.js"; import { resolveMemoryScope } from "./memory-scope.js"; @@ -224,7 +225,7 @@ export function registerGeminiRoute(app: Hono, deps: GeminiRouteDeps): v const acquired = await deps.concurrencyGate.acquire({ keyId: identity.keyId, limit: identity.caps?.concurrencyLimit ?? null, - signal: c.req.raw.signal, + signal: requestSignal(c), }); if (!acquired.ok) { c.header("retry-after", String(acquired.retryAfterSeconds)); @@ -268,7 +269,7 @@ export function registerGeminiRoute(app: Hono, deps: GeminiRouteDeps): v const counted = await deps.countTokens( { ...(native as Record), model: route.model }, identity, - c.req.raw.signal, + requestSignal(c), ); return c.json(counted as Record); } catch { @@ -340,7 +341,7 @@ export function registerGeminiRoute(app: Hono, deps: GeminiRouteDeps): v // PipelineError(invalid_request) for an empty request. let result: PipelineRunResult; try { - result = await deps.pipeline.run(ir, identity, c.req.raw.signal); + result = await deps.pipeline.run(ir, identity, requestSignal(c)); } catch (err) { if (err instanceof PipelineError) { return sendError(c, { @@ -379,7 +380,7 @@ export function registerGeminiRoute(app: Hono, deps: GeminiRouteDeps): v try { for await (const item of withHeartbeat(result.streamIR(), { heartbeatMs, - signal: c.req.raw.signal, + signal: requestSignal(c), })) { if (item.type === "beat") { if (atEventBoundary(lastWrite)) await sse.write(HEARTBEAT_COMMENT); @@ -440,6 +441,7 @@ export function registerGeminiRoute(app: Hono, deps: GeminiRouteDeps): v decision: result.decision, requestJson, responseJson: captureBodies ? captured.join("") : null, + timedOut: requestTimedOut(c), upstreamRequestJson: result.upstreamRequest ?? null, }, (msg) => c.get("logger").log("warn", msg, { trace_id: traceId }), @@ -471,6 +473,7 @@ export function registerGeminiRoute(app: Hono, deps: GeminiRouteDeps): v decision: result.decision, requestJson, responseJson: null, + timedOut: requestTimedOut(c), upstreamRequestJson: result.upstreamRequest ?? null, }, (msg) => c.get("logger").log("warn", msg, { trace_id: traceId }), @@ -497,6 +500,7 @@ export function registerGeminiRoute(app: Hono, deps: GeminiRouteDeps): v decision: result.decision, requestJson, responseJson: captureBodies ? JSON.stringify(body) : null, + timedOut: requestTimedOut(c), upstreamRequestJson: result.upstreamRequest ?? null, }, (msg) => c.get("logger").log("warn", msg, { trace_id: traceId }), diff --git a/apps/gateway/src/routes/images.ts b/apps/gateway/src/routes/images.ts index f47e3b95..eb55946c 100644 --- a/apps/gateway/src/routes/images.ts +++ b/apps/gateway/src/routes/images.ts @@ -5,6 +5,7 @@ import type { ContentfulStatusCode } from "hono/utils/http-status"; import type { AppEnv } from "../app.js"; import { type ConcurrencyGatePort, concurrencyReleaseGuard } from "../middleware/concurrency.js"; import { estimateRequestTokens } from "../middleware/estimate-tokens.js"; +import { requestSignal, requestTimedOut } from "../middleware/limits.js"; import type { GeminiRateLimiterPort } from "./gemini.js"; import { type ImageAttempt, @@ -146,7 +147,7 @@ export function registerImagesRoute(app: Hono, deps: ImagesRouteDeps): v const acquired = await deps.concurrencyGate.acquire({ keyId: identity.keyId, limit: identity.caps?.concurrencyLimit ?? null, - signal: c.req.raw.signal, + signal: requestSignal(c), }); if (!acquired.ok) { c.header("retry-after", String(acquired.retryAfterSeconds)); @@ -255,13 +256,13 @@ export function registerImagesRoute(app: Hono, deps: ImagesRouteDeps): v contents: [{ role: "user", parts: [{ text: parsed.data.prompt }] }], generationConfig: { responseModalities: ["TEXT", "IMAGE"] }, }, - { signal: c.req.raw.signal, captureUpstream }, + { signal: requestSignal(c), captureUpstream }, ); upstream = mapGeminiToImages((native ?? {}) as Record); } else { upstream = (await target.client.imageGeneration?.( { ...parsed.data, model: target.providerModel }, - { signal: c.req.raw.signal, captureUpstream }, + { signal: requestSignal(c), captureUpstream }, )) as Record; } const usage = (upstream as { usage?: Record }).usage ?? null; @@ -277,7 +278,7 @@ export function registerImagesRoute(app: Hono, deps: ImagesRouteDeps): v permittedChain.targets, deps.breaker, attempt, - c.req.raw.signal, + requestSignal(c), ); // 6b) Terminal failure. A client abort is a NON-provider fault → no record, no 5xx. @@ -303,6 +304,7 @@ export function registerImagesRoute(app: Hono, deps: ImagesRouteDeps): v decision, requestJson, responseJson: null, + timedOut: requestTimedOut(c), upstreamRequestJson: null, }, log, @@ -343,6 +345,7 @@ export function registerImagesRoute(app: Hono, deps: ImagesRouteDeps): v decision, requestJson, responseJson, + timedOut: requestTimedOut(c), upstreamRequestJson: result.upstreamRequestJson, }, log, diff --git a/apps/gateway/src/routes/interactions.ts b/apps/gateway/src/routes/interactions.ts index 1ed12e7b..d30f85f6 100644 --- a/apps/gateway/src/routes/interactions.ts +++ b/apps/gateway/src/routes/interactions.ts @@ -5,6 +5,7 @@ import type { ContentfulStatusCode } from "hono/utils/http-status"; import type { AppEnv } from "../app.js"; import { type ConcurrencyGatePort, concurrencyReleaseGuard } from "../middleware/concurrency.js"; import { estimateRequestTokens } from "../middleware/estimate-tokens.js"; +import { requestSignal, requestTimedOut } from "../middleware/limits.js"; import type { GeminiRateLimiterPort } from "./gemini.js"; import { type ImageAttempt, @@ -220,7 +221,7 @@ export function registerInteractionsRoute(app: Hono, deps: InteractionsR const acquired = await deps.concurrencyGate.acquire({ keyId: identity.keyId, limit: identity.caps?.concurrencyLimit ?? null, - signal: c.req.raw.signal, + signal: requestSignal(c), }); if (!acquired.ok) { c.header("retry-after", String(acquired.retryAfterSeconds)); @@ -336,7 +337,7 @@ export function registerInteractionsRoute(app: Hono, deps: InteractionsR parsed.data.generation_config ?? null, ), }, - { signal: c.req.raw.signal, captureUpstream }, + { signal: requestSignal(c), captureUpstream }, )) as Record; const interactionsBody = nativeToInteractions(native, `int_${traceId}`); const usageBody = usageBodyFromNative(native); @@ -348,7 +349,7 @@ export function registerInteractionsRoute(app: Hono, deps: InteractionsR }; }; - const outcome = await runImageChain(geminiTargets, deps.breaker, attempt, c.req.raw.signal); + const outcome = await runImageChain(geminiTargets, deps.breaker, attempt, requestSignal(c)); // 6b) Terminal failure. A client abort is a NON-provider fault → no record. if (!outcome.ok) { @@ -373,6 +374,7 @@ export function registerInteractionsRoute(app: Hono, deps: InteractionsR decision, requestJson, responseJson: null, + timedOut: requestTimedOut(c), upstreamRequestJson: null, }, log, @@ -412,6 +414,7 @@ export function registerInteractionsRoute(app: Hono, deps: InteractionsR decision, requestJson, responseJson, + timedOut: requestTimedOut(c), upstreamRequestJson: result.upstreamRequestJson, }, log, diff --git a/apps/gateway/src/routes/messages.ts b/apps/gateway/src/routes/messages.ts index 9494e8f9..494cc1cf 100644 --- a/apps/gateway/src/routes/messages.ts +++ b/apps/gateway/src/routes/messages.ts @@ -13,6 +13,7 @@ import type { ContentfulStatusCode } from "hono/utils/http-status"; import type { AppEnv } from "../app.js"; import { type ConcurrencyGatePort, concurrencyReleaseGuard } from "../middleware/concurrency.js"; import { estimateRequestTokens } from "../middleware/estimate-tokens.js"; +import { requestSignal, requestTimedOut } from "../middleware/limits.js"; import { type ServingAccount, stampServingAccount } from "../runtime/serving-account.js"; import { atEventBoundary, HEARTBEAT_COMMENT, withHeartbeat } from "./heartbeat.js"; import { type MemoryKeyDefaults, resolveMemoryScope } from "./memory-scope.js"; @@ -294,7 +295,7 @@ export function registerMessagesRoute(app: Hono, deps: MessagesRouteDeps } if (deps.countTokens !== undefined) { try { - return c.json(await deps.countTokens(obj, identity, c.req.raw.signal)); + return c.json(await deps.countTokens(obj, identity, requestSignal(c))); } catch { // Token helpers are compatibility helpers, not generation. Fall back to a // deterministic estimate instead of making /v1/messages/count_tokens flaky. @@ -358,7 +359,7 @@ export function registerMessagesRoute(app: Hono, deps: MessagesRouteDeps const acquired = await deps.concurrencyGate.acquire({ keyId: identity.keyId, limit: identity.caps?.concurrencyLimit ?? null, - signal: c.req.raw.signal, + signal: requestSignal(c), }); if (!acquired.ok) { c.header("retry-after", String(acquired.retryAfterSeconds)); @@ -490,7 +491,7 @@ export function registerMessagesRoute(app: Hono, deps: MessagesRouteDeps // (no placeholder synthesis) — map it to a 400 in the Anthropic envelope. let result: PipelineRunResult; try { - result = await deps.pipeline.run(ir, identity, c.req.raw.signal); + result = await deps.pipeline.run(ir, identity, requestSignal(c)); } catch (err) { if (err instanceof PipelineError) { return sendError(c, { @@ -540,7 +541,7 @@ export function registerMessagesRoute(app: Hono, deps: MessagesRouteDeps try { for await (const item of withHeartbeat(result.streamIR(), { heartbeatMs, - signal: c.req.raw.signal, + signal: requestSignal(c), })) { if (item.type === "beat") { if (atEventBoundary(lastWrite)) await sse.write(HEARTBEAT_COMMENT); @@ -605,6 +606,7 @@ export function registerMessagesRoute(app: Hono, deps: MessagesRouteDeps decision: result.decision, requestJson, responseJson: captureBodies ? captured.join("") : null, + timedOut: requestTimedOut(c), upstreamRequestJson: result.upstreamRequest ?? null, }, (msg) => c.get("logger").log("warn", msg, { trace_id: traceId }), @@ -643,6 +645,7 @@ export function registerMessagesRoute(app: Hono, deps: MessagesRouteDeps decision: result.decision, requestJson, responseJson: null, + timedOut: requestTimedOut(c), upstreamRequestJson: result.upstreamRequest ?? null, }, (msg) => c.get("logger").log("warn", msg, { trace_id: traceId }), @@ -669,6 +672,7 @@ export function registerMessagesRoute(app: Hono, deps: MessagesRouteDeps decision: result.decision, requestJson, responseJson: captureBodies ? JSON.stringify(body) : null, + timedOut: requestTimedOut(c), upstreamRequestJson: result.upstreamRequest ?? null, }, (msg) => c.get("logger").log("warn", msg, { trace_id: traceId }), diff --git a/apps/gateway/src/routes/payload-capture.test.ts b/apps/gateway/src/routes/payload-capture.test.ts index 8e2b4c17..2d91fe83 100644 --- a/apps/gateway/src/routes/payload-capture.test.ts +++ b/apps/gateway/src/routes/payload-capture.test.ts @@ -738,4 +738,23 @@ describe("recordServed — deferred write queue (the three pipeline faces)", () expect(s.inserted).toHaveLength(1); expect(s.payloads).toHaveLength(1); }); + + it("records a timed-out request as an error even if the provider later completed", async () => { + const s = sink(); + const d: RecordServedDeps = { + telemetry: s.telemetry, + redact: (x) => x, + now: () => 5000, + capturePayloads: () => true, + }; + await recordServed(d, { ...args, requestId: "req_timeout", timedOut: true }, () => {}); + expect(s.inserted[0]?.final).toMatchObject({ + status: "error", + error_reason: "timeout", + model_alias: "openai/gpt", + }); + expect(s.inserted[0]?.serving_account).toBeNull(); + expect(s.payloads[0]?.responseJson).toBeNull(); + expect(args.decision.final.status).toBe("ok"); + }); }); diff --git a/apps/gateway/src/routes/payload-capture.ts b/apps/gateway/src/routes/payload-capture.ts index 43093d79..7b282436 100644 --- a/apps/gateway/src/routes/payload-capture.ts +++ b/apps/gateway/src/routes/payload-capture.ts @@ -107,6 +107,18 @@ export interface RecordServedDeps extends PayloadCaptureDeps { writes?: WriteQueue; } +export function decisionForTimedOutRequest(decision: DecisionRecord): DecisionRecord { + return { + ...decision, + final: { + ...decision.final, + status: "error", + error_reason: "timeout", + }, + serving_account: null, + }; +} + // Record ONE served request: the telemetry row (always — this is what makes the // request appear in /admin/requests) plus the verbatim request/response payload // (gated by capture_payloads). Shared by the three pipeline faces (/v1/responses, @@ -121,12 +133,16 @@ export async function recordServed( decision: DecisionRecord; requestJson: string; responseJson: string | null; + timedOut?: boolean; // The exact body forwarded upstream (post inject + translation); null when no // provider served or capture context lacks it. upstreamRequestJson?: string | null; }, log: (msg: string) => void, ): Promise { + const decision = + args.timedOut === true ? decisionForTimedOutRequest(args.decision) : args.decision; + const responseJson = args.timedOut === true ? null : args.responseJson; // Deferred + batched path: enqueue both writes to run AFTER the response, off the // hot path. The redaction is done HERE (synchronously) so the enqueued snapshot is // independent of anything that touches the decision later. @@ -136,7 +152,7 @@ export async function recordServed( w.enqueuePayload({ requestId: args.requestId, requestJson: args.requestJson, - responseJson: args.responseJson, + responseJson, upstreamRequestJson: args.upstreamRequestJson ?? null, createdAt: new Date(deps.now()), }); @@ -144,7 +160,7 @@ export async function recordServed( // owns payload retention (archive-first), governed by the cleanup settings. } w.enqueueTelemetry({ - decision: deps.redact(args.decision), + decision: deps.redact(decision), apiKeyId: args.apiKeyId, createdAt: new Date(deps.now()), }); @@ -157,7 +173,7 @@ export async function recordServed( { requestId: args.requestId, requestJson: args.requestJson, - responseJson: args.responseJson, + responseJson, upstreamRequestJson: args.upstreamRequestJson ?? null, now: deps.now(), }, @@ -165,7 +181,7 @@ export async function recordServed( ); try { await deps.telemetry.insert({ - decision: deps.redact(args.decision), + decision: deps.redact(decision), apiKeyId: args.apiKeyId, createdAt: new Date(deps.now()), }); diff --git a/apps/gateway/src/routes/responses.ts b/apps/gateway/src/routes/responses.ts index ea507a47..ac88d119 100644 --- a/apps/gateway/src/routes/responses.ts +++ b/apps/gateway/src/routes/responses.ts @@ -7,6 +7,7 @@ import type { AppEnv } from "../app.js"; import { type ConcurrencyGatePort, concurrencyReleaseGuard } from "../middleware/concurrency.js"; import { HelmHttpError } from "../middleware/error-handler.js"; import { estimateRequestTokens } from "../middleware/estimate-tokens.js"; +import { requestSignal, requestTimedOut } from "../middleware/limits.js"; import { stampServingAccount } from "../runtime/serving-account.js"; import { atEventBoundary, HEARTBEAT_COMMENT, withHeartbeat } from "./heartbeat.js"; import { resolveMemoryScope } from "./memory-scope.js"; @@ -364,8 +365,8 @@ export function registerResponsesRoute(app: Hono, deps: ResponsesRouteDe } const body = deps.registry !== undefined - ? await method(responseId, identity, c.req.raw.signal, registryRecord ?? undefined) - : await method(responseId, identity, c.req.raw.signal); + ? await method(responseId, identity, requestSignal(c), registryRecord ?? undefined) + : await method(responseId, identity, requestSignal(c)); return c.json(body as Record); }; @@ -385,7 +386,7 @@ export function registerResponsesRoute(app: Hono, deps: ResponsesRouteDe ir.metadata = { ...(ir.metadata ?? {}), trace_id: traceId }; let result: PipelineRunResult; try { - result = await deps.pipeline.run(ir, identity, c.req.raw.signal); + result = await deps.pipeline.run(ir, identity, requestSignal(c)); const collected = await result.collect(); return c.json(deps.transformer.transformResponseOut(collected) as Record); } catch (err) { @@ -393,7 +394,7 @@ export function registerResponsesRoute(app: Hono, deps: ResponsesRouteDe throw err; } } - const body = await method(native, identity, c.req.raw.signal); + const body = await method(native, identity, requestSignal(c)); return c.json(body as Record); }; @@ -403,7 +404,7 @@ export function registerResponsesRoute(app: Hono, deps: ResponsesRouteDe const native = await parseJsonBody(c, traceId); const providerMethod = deps.lifecycle?.inputTokens; if (providerMethod !== undefined) { - const body = await providerMethod(native, identity, c.req.raw.signal); + const body = await providerMethod(native, identity, requestSignal(c)); return c.json(body as Record); } return c.json({ input_tokens: estimateResponsesInputTokens(native), estimated: true }); @@ -455,7 +456,7 @@ export function registerResponsesRoute(app: Hono, deps: ResponsesRouteDe const acquired = await deps.concurrencyGate.acquire({ keyId: identity.keyId, limit: identity.caps?.concurrencyLimit ?? null, - signal: c.req.raw.signal, + signal: requestSignal(c), }); if (!acquired.ok) { c.header("retry-after", String(acquired.retryAfterSeconds)); @@ -581,10 +582,10 @@ export function registerResponsesRoute(app: Hono, deps: ResponsesRouteDe let streamResponseId: string | null = null; let streamStatus: string | null = null; try { - result = await deps.pipeline.run(ir, identity, c.req.raw.signal); + result = await deps.pipeline.run(ir, identity, requestSignal(c)); for await (const item of withHeartbeat(result.streamIR(), { heartbeatMs, - signal: c.req.raw.signal, + signal: requestSignal(c), })) { if (item.type === "beat") { if (atEventBoundary(lastWrite)) await sse.write(HEARTBEAT_COMMENT); @@ -658,6 +659,7 @@ export function registerResponsesRoute(app: Hono, deps: ResponsesRouteDe decision: result.decision, requestJson, responseJson: captureBodies ? captured.join("") : null, + timedOut: requestTimedOut(c), upstreamRequestJson: result.upstreamRequest ?? null, }, (msg) => c.get("logger").log("warn", msg, { trace_id: traceId }), @@ -697,7 +699,7 @@ export function registerResponsesRoute(app: Hono, deps: ResponsesRouteDe // after the Responses prelude has already reached the client. let result: PipelineRunResult; try { - result = await deps.pipeline.run(ir, identity, c.req.raw.signal); + result = await deps.pipeline.run(ir, identity, requestSignal(c)); } catch (err) { if (err instanceof PipelineError) throw pipelineToHelm(err, traceId); throw err; @@ -735,6 +737,7 @@ export function registerResponsesRoute(app: Hono, deps: ResponsesRouteDe decision: result.decision, requestJson, responseJson: null, + timedOut: requestTimedOut(c), upstreamRequestJson: result.upstreamRequest ?? null, }, (msg) => c.get("logger").log("warn", msg, { trace_id: traceId }), @@ -776,6 +779,7 @@ export function registerResponsesRoute(app: Hono, deps: ResponsesRouteDe decision: result.decision, requestJson, responseJson: captureBodies ? JSON.stringify(body) : null, + timedOut: requestTimedOut(c), upstreamRequestJson: result.upstreamRequest ?? null, }, (msg) => c.get("logger").log("warn", msg, { trace_id: traceId }), diff --git a/implementation-notes.md b/implementation-notes.md index 7ae28a08..4038a628 100644 --- a/implementation-notes.md +++ b/implementation-notes.md @@ -7,6 +7,15 @@ --- +## 2026-07-06 · 请求总超时必须驱动下游 abort 与失败 telemetry（Gateway runtime / telemetry，docs/02/07，原则 3/5/7） + +- **背景（Lukin）**：生产 `gpt-5.5` 长请求超过 Helm `request_timeout_ms` 后，客户端收到 504，但上游 provider 后续完成，Telemetry 仍把 `final_status` 记成 `ok`，导致日志和 Admin requests 不能反映客户端真实结果。 +- **语义决策**：Gateway 总超时是**客户端可见的终态**；一旦触发，后续即使 provider 晚完成，也不能把该 request 记录为成功。持久化前把 `DecisionRecord.final.status` 规范成 `error`、`error_reason: "timeout"`，同时 `serving_account` 清空。Provider attempt 仍保留原始上游事实（例如 late `ok` / cost），避免把“上游后来完成”伪造成 provider failure。 +- **执行决策**：timeout 中间件在 Hono context 上挂一个 request state：`timedOut` 与 `AbortSignal.any([client, timeout])`。各路由的 provider/pipeline/concurrency 调用使用这个统一 signal，尽量在 Helm 超时时停止下游；真实客户端断开判断仍只看原始 client signal，避免把 Helm timeout 当成用户主动取消。 +- **payload 决策**：如果请求已经 timeout，`request_payloads.response_json` 写 `null`，因为客户端实际收到的是 504，不是晚到的 provider response。`upstream_request_json` 可继续保留，便于追查发给上游的请求。 +- **覆盖面**：OpenAI Chat 的自有 persist 路径和 `recordServed` 共享路径都覆盖；Messages / Responses / Gemini / Images / Interactions 都传入 request timeout state，防止协议面漂移。 +- **验证路径**：新增 timeout context 与 late-success telemetry 测试；覆盖 app/limits/chat/messages/responses/gemini/images/interactions/payload-capture targeted tests，并跑 gateway typecheck。 + ## 2026-07-06 · API key 绝对模型黑名单（Key governance / routing / Admin keys，docs/04/06/11，原则 5/6/7） - **背景（Lukin）**：每个 API key 需要能禁止若干具体模型，语义是“这个用户无论通过 direct model、显式 lane、auto/classified lane、alias-to-lane 还是 execution fallback，都不能实际用到这些模型”。 @@ -90,21 +99,13 @@ - **恢复边界**：干净窗口会继续清理已存在 cooldown；scoped model 窗口（如 Anthropic `7d-*`）仍不扩大成全账号停车。Codex reset credit 消费成功后通过现有刷新路径恢复窗口并清理 cooldown。 - **测试路径**：覆盖 `/oauth/quota` 从 Codex saturated PULL 新建 cooldown，以及 near-full PULL 不误停车；再跑 admin OAuth route focused tests、typecheck/lint/build。 -## 2026-07-05 · OAuth 凭证失效持久化为 needs reconnect（OAuth provider pool / Admin providers，docs/04/11，原则 3/5/7） - -- **背景（Lukin）**：Providers 页连通性测试可能返回 `oauth refresh failed (... status 401)`，但账号仍显示 connected，且未自动从调度中禁用；原实现只在当前 pool 进程内把 401/403 账号摘掉，admin status 与持久配置没有同步。 -- **状态决策**：refresh 400/401/403 或持久 upstream 401/403 视为 durable credential failure，而不是 429 usage cooldown。Helm 会写入 encrypted account settings：`credentialFailedAt` / `credentialFailureReason`，并把账号 `schedulable:false`；status 显示 unhealthy，重新合成 pool 时跳过该账号。 -- **恢复边界**：成功 reconnect 会清除 credential failure。若账号是 Helm 因凭证失败自动禁用，则恢复默认 schedulable；若原本就是用户手动 parked，则 reconnect 后仍保持手动 parked。 -- **操作边界**：Providers 页会显式返回 `credentialFailed` 并禁用 schedulable 开关；后端同样拒绝把 credential-failed 账号直接设回 schedulable。状态页若首次发现永久凭证失败，会触发 live pool rebuild，避免“后台已禁用但当前 pool 仍在用”。 -- **测试路径**：覆盖 core pool credential-failure hook、admin test 401 标记、account-settings mark/clear、synthesizeOAuthProviders 跳过 failed account、providers UI 失败后刷新；再跑 targeted Vitest、typecheck、lint、diff check。 - ## 历史条目摘要（最近 5 条） +- **2026-07-05 · OAuth 凭证失效持久化为 needs reconnect（OAuth provider pool / Admin providers，docs/04/11，原则 3/5/7）**：refresh/持久 upstream 400/401/403 标记 credential failure、写入账号设置并摘出调度，reconnect 成功后按手动/自动停车边界恢复。 - **2026-07-05 · 避免浪费策略纳入周额度与 Codex reset credits（OAuth provider pool / quota，docs/04/11，原则 3/5/7）**：`use_expiring` 汇总短窗口、周额度与 reset credits 软评分，quota PULL 会刷新 live pool snapshot 但不自动消费 credit。 - **2026-07-05 · Codex reset-credit 消费改为硬门禁（OAuth quota / Admin providers，docs/04/11，原则 3/5/7）**：手动/自动 reset credit 必须命中 weekly secondary 阈值与持久 guard，缺少 snapshot 时 fail-closed，避免烧稀缺额度。 - **2026-07-05 · OAuth 账号池支持可选额度使用策略（OAuth provider pool / routing / Admin providers，docs/04/11，原则 3/5/7）**：新增 balanced/manual_priority/low_risk/use_expiring 全局账号池策略，保持 previous_response_id 强亲和和 scoped quota 边界。 - **2026-07-04 · internal LLM prompt 输入用 XML 数据边界隔离（Memory / classifier eval，docs/03/08/12，原则 3/4/7）**：classifier eval 与 memory LLM 把可信任务和不可信消息放进 XML/JSON 数据区并 escape，防止用户内容突破数据边界。 -- **2026-07-04 · 请求记录最终订阅账号并重排请求列表字段（Telemetry / Admin requests，docs/04/07/11，原则 1/5/7）**：`DecisionRecord.serving_account` 记录最终服务订阅账号，admin 请求列表/详情按排障优先级展示 provider、account、served/requested model。 ## 更早历史总览 diff --git a/package.json b/package.json index 879cf82c..72816be3 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "helm-api", - "version": "0.25.14", + "version": "0.25.15", "private": true, "type": "module", "engines": {