diff --git a/backend/internal/handler/gateway_handler.go b/backend/internal/handler/gateway_handler.go index 8f863739ac..116346b4a6 100644 --- a/backend/internal/handler/gateway_handler.go +++ b/backend/internal/handler/gateway_handler.go @@ -1628,6 +1628,11 @@ func (h *GatewayHandler) mapUpstreamError(statusCode int) (int, string, string) // handleStreamingAwareError handles errors that may occur after streaming has started func (h *GatewayHandler) handleStreamingAwareError(c *gin.Context, status int, errType, message string, streamStarted bool) { if streamStarted { + // 响应状态码已固化为 200(ping/部分数据已 flush),错误只能就地以 SSE 帧回传。 + // 标记本次流内错误,供 ops_error_logger 补记——否则该中间件按 status>=400 采集, + // 这类挂在 200 流上的失败(如并发限流回退)不会进错误看板。 + service.MarkOpsStreamError(c, errType, message, status) + // /v1/responses 的严格 SDK(Codex CLI)要求终止事件必须属于 // response.completed/failed/incomplete/cancelled 集合。 // Anthropic-backed Responses 路径同样会因为通用 error 帧被拒。 diff --git a/backend/internal/handler/ops_error_logger.go b/backend/internal/handler/ops_error_logger.go index 8cd9d02ba0..5a1e57ff7d 100644 --- a/backend/internal/handler/ops_error_logger.go +++ b/backend/internal/handler/ops_error_logger.go @@ -590,6 +590,10 @@ func OpsErrorLoggerMiddleware(ops *service.OpsService) gin.HandlerFunc { } } if !hasUpstreamContext { + // 没有上游错误上下文,但网关可能在已固化的 200 流上就地补发了 SSE 错误帧 + // (如 ping 等待后并发超限、Wait 后二次计费校验失败)。这类失败若不在此补记, + // 会因 wire 状态码为 200 而在错误看板里彻底隐形。 + logOpsStreamError(c, ops, status) return } @@ -999,6 +1003,138 @@ func OpsErrorLoggerMiddleware(ops *service.OpsService) gin.HandlerFunc { } } +// logOpsStreamError 记录一次挂在已固化 HTTP 200 SSE 流上的就地错误。 +// 由于 wire 状态码停留在 200,常规的 status>=400 捕获路径永远不会触发; +// handleStreamingAwareError 通过 service.MarkOpsStreamError 标记这类错误, +// 此函数据此补记一条错误日志,让并发限流/流内失败在错误看板里可见。 +// +// 仅在 status<400 且不存在上游错误上下文时调用:上游透传错误已由中间件的 +// upstream-context 分支落库,无需在此重复记录。 +func logOpsStreamError(c *gin.Context, ops *service.OpsService, wireStatus int) { + streamErr, ok := service.GetOpsStreamError(c) + if !ok { + return + } + + // 命中 skip_monitoring=true 透传规则的请求跳过落库,与其它分支一致。 + if v, ok := c.Get(service.OpsSkipPassthroughKey); ok { + if skip, _ := v.(bool); skip { + return + } + } + + // 复用与 status>=400 分支相同的设置过滤(context canceled / 无可用账号等)。 + if shouldSkipOpsErrorLog(c.Request.Context(), ops, streamErr.Message, streamErr.Message, c.Request.URL.Path) { + return + } + + // 分级用「本应返回的状态码」(如并发限流 429),wire 状态码缺省时回退。 + classifyStatus := streamErr.IntendedStatus + if classifyStatus <= 0 { + classifyStatus = wireStatus + } + normalizedType := normalizeOpsErrorType(streamErr.ErrType, "") + phase, isBusinessLimited, errorOwner, errorSource := classifyOpsErrorLog(c, normalizedType, streamErr.Message, "", classifyStatus) + + apiKey := getOpsAPIKey(c) + clientRequestID, _ := c.Request.Context().Value(ctxkey.ClientRequestID).(string) + + model, _ := c.Get(opsModelKey) + var modelName string + if s, ok := model.(string); ok { + modelName = s + } + accountIDV, _ := c.Get(opsAccountIDKey) + var accountID *int64 + if v, ok := accountIDV.(int64); ok && v > 0 { + accountID = &v + } + + fallbackPlatform := guessPlatformFromPath(c.Request.URL.Path) + platform := resolveOpsPlatform(apiKey, fallbackPlatform) + + requestID := c.Writer.Header().Get("X-Request-Id") + if requestID == "" { + requestID = c.Writer.Header().Get("x-request-id") + } + + entry := &service.OpsInsertErrorLogInput{ + RequestID: requestID, + ClientRequestID: clientRequestID, + + AccountID: accountID, + Platform: platform, + Model: modelName, + RequestPath: func() string { + if c.Request != nil && c.Request.URL != nil { + return c.Request.URL.Path + } + return "" + }(), + // 就地 SSE 错误只出现在流式请求上。 + Stream: true, + InboundEndpoint: GetInboundEndpoint(c), + UpstreamEndpoint: GetUpstreamEndpoint(c, platform), + RequestedModel: modelName, + UpstreamModel: func() string { + if v, ok := c.Get(opsUpstreamModelKey); ok { + if s, ok := v.(string); ok { + return strings.TrimSpace(s) + } + } + return "" + }(), + RequestType: func() *int16 { + if v, ok := c.Get(opsRequestTypeKey); ok { + switch t := v.(type) { + case int16: + return &t + case int: + v16 := int16(t) + return &v16 + } + } + return nil + }(), + UserAgent: c.GetHeader("User-Agent"), + + ErrorPhase: phase, + ErrorType: normalizedType, + Severity: classifyOpsSeverity(normalizedType, classifyStatus), + StatusCode: wireStatus, + IsBusinessLimited: isBusinessLimited, + IsCountTokens: isCountTokensRequest(c), + + ErrorMessage: streamErr.Message, + ErrorBody: "", + ErrorSource: errorSource, + ErrorOwner: errorOwner, + + CreatedAt: time.Now(), + } + applyOpsLatencyFieldsFromContext(c, entry) + + if apiKey != nil { + entry.APIKeyID = &apiKey.ID + entry.APIKeyPrefix = keyPrefix(apiKey.Key, 8) + if apiKey.User != nil { + entry.UserID = &apiKey.User.ID + } + if apiKey.GroupID != nil { + entry.GroupID = apiKey.GroupID + } + if apiKey.Group != nil && apiKey.Group.Platform != "" { + entry.Platform = apiKey.Group.Platform + } + } + + if clientIP := strings.TrimSpace(ip.GetClientIP(c)); clientIP != "" { + entry.ClientIP = &clientIP + } + + enqueueOpsErrorLog(ops, entry) +} + // isCountTokensRequest checks if the request is a count_tokens request func isCountTokensRequest(c *gin.Context) bool { if c == nil || c.Request == nil || c.Request.URL == nil { diff --git a/backend/internal/handler/ops_error_logger_test.go b/backend/internal/handler/ops_error_logger_test.go index cf1685f2a4..89bdd938c1 100644 --- a/backend/internal/handler/ops_error_logger_test.go +++ b/backend/internal/handler/ops_error_logger_test.go @@ -139,6 +139,97 @@ func TestOpsErrorLoggerMiddleware_DoesNotBreakOuterMiddlewares(t *testing.T) { require.Equal(t, http.StatusNoContent, rec.Code) } +// setupOpsErrorLogTestQueue 阻止 enqueueOpsErrorLog 启动真实 worker,改用可检查的测试队列。 +func setupOpsErrorLogTestQueue(t *testing.T, size int) { + t.Helper() + resetOpsErrorLoggerStateForTest(t) + opsErrorLogOnce.Do(func() {}) + opsErrorLogMu.Lock() + opsErrorLogQueue = make(chan opsErrorLogJob, size) + opsErrorLogMu.Unlock() +} + +// 就地(in-band) SSE 错误挂在已固化的 HTTP 200 流上:wire 状态码为 200, +// 常规 status>=400 采集路径不会触发。logOpsStreamError 必须据 MarkOpsStreamError +// 补记一条错误日志,且用 IntendedStatus(429) 分级、StatusCode 仍记 wire 的 200。 +func TestLogOpsStreamError_RecordsInBandConcurrencyLimit(t *testing.T) { + setupOpsErrorLogTestQueue(t, 4) + + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + c.Set(opsModelKey, "test-model") + + service.MarkOpsStreamError(c, "rate_limit_error", + "Concurrency limit exceeded for account, please retry later", http.StatusTooManyRequests) + + ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil) + logOpsStreamError(c, ops, http.StatusOK) + + require.Equal(t, int64(1), OpsErrorLogEnqueuedTotal()) + require.Equal(t, int64(1), OpsErrorLogQueueLength()) + + job := <-opsErrorLogQueue + require.NotNil(t, job.entry) + require.Equal(t, "rate_limit_error", job.entry.ErrorType) + require.Equal(t, "request", job.entry.ErrorPhase) + require.True(t, job.entry.IsBusinessLimited) + require.True(t, job.entry.Stream) + require.Equal(t, http.StatusOK, job.entry.StatusCode) // wire 状态码保持 200 + require.Equal(t, "P1", job.entry.Severity) // 用 IntendedStatus 429 分级 + require.Equal(t, "test-model", job.entry.Model) + require.Equal(t, "Concurrency limit exceeded for account, please retry later", job.entry.ErrorMessage) +} + +// 未标记流内错误时 logOpsStreamError 必须是 no-op(不误记正常的 200 流)。 +func TestLogOpsStreamError_NoopWhenNotMarked(t *testing.T) { + setupOpsErrorLogTestQueue(t, 4) + + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + + ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil) + logOpsStreamError(c, ops, http.StatusOK) + + require.Equal(t, int64(0), OpsErrorLogEnqueuedTotal()) +} + +// 命中 skip_monitoring=true 透传规则时不落库,与其它采集分支一致。 +func TestLogOpsStreamError_SkipWhenPassthroughSkipMonitoring(t *testing.T) { + setupOpsErrorLogTestQueue(t, 4) + + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + service.MarkOpsStreamError(c, "upstream_error", "Upstream request failed", http.StatusBadGateway) + c.Set(service.OpsSkipPassthroughKey, true) + + ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil) + logOpsStreamError(c, ops, http.StatusOK) + + require.Equal(t, int64(0), OpsErrorLogEnqueuedTotal()) +} + +// MarkOpsStreamError 采用「首个标记生效」:后续的通用兜底帧不得覆盖根因错误。 +func TestMarkOpsStreamError_FirstWins(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + + service.MarkOpsStreamError(c, "rate_limit_error", "Concurrency limit exceeded for account", http.StatusTooManyRequests) + service.MarkOpsStreamError(c, "upstream_error", "Upstream request failed", http.StatusBadGateway) + + se, ok := service.GetOpsStreamError(c) + require.True(t, ok) + require.Equal(t, "rate_limit_error", se.ErrType) + require.Equal(t, "Concurrency limit exceeded for account", se.Message) + require.Equal(t, http.StatusTooManyRequests, se.IntendedStatus) +} + func TestIsKnownOpsErrorType(t *testing.T) { known := []string{ "invalid_request_error", diff --git a/backend/internal/service/ops_upstream_context.go b/backend/internal/service/ops_upstream_context.go index 4a1e36aa6f..6389f1ecb5 100644 --- a/backend/internal/service/ops_upstream_context.go +++ b/backend/internal/service/ops_upstream_context.go @@ -32,6 +32,12 @@ const ( // ops_error_logger 中间件检查此 key,为 true 时跳过错误记录。 OpsSkipPassthroughKey = "ops_skip_passthrough" + // OpsStreamErrorKey 保存 handleStreamingAwareError 在「响应已固化为 HTTP 200 的 SSE 流」 + // 上就地(in-band)补发错误帧时记录的 OpsStreamError。因为 wire 状态码停留在 200, + // ops_error_logger 的 status>=400 采集路径永远不会触发,这类流内失败 + //(例如等待并发槽位超时后回退的限流、Wait 后二次计费校验失败)本会在错误看板里隐形。 + OpsStreamErrorKey = "ops_stream_error" + // Client-side configuration denials should remain visible in ops_error_logs, // but should be excluded from SLA/error-rate calculations. // ResponseCommittedKey 由 handleErrorResponse 系列函数在写完 HTTP 错误响应后设置。 @@ -87,6 +93,52 @@ func HasOpsClientBusinessLimited(c *gin.Context) bool { return marked } +// OpsStreamError 描述网关在「响应状态已固化为 200」之后(keepalive ping 或部分数据 +// 已 flush)就地以 SSE error 帧形式返回的错误。由于 HTTP 状态码停留在 200, +// 而 ops_error_logger 以 status>=400 为采集触发条件,这类流内失败 +// (并发限流回退、Wait 后二次计费校验失败、流开始后才无可用账号等)本会在错误看板里 +// 完全隐形。handler.handleStreamingAwareError 负责标记,ops_error_logger 中间件在 +// status<400 分支消费它并补记一条错误日志。 +type OpsStreamError struct { + // ErrType 是写入 SSE 帧的对客错误类型(如 rate_limit_error / upstream_error / api_error)。 + ErrType string + // Message 是写入 SSE 帧的对客错误消息。 + Message string + // IntendedStatus 是流若未固化本应返回的 HTTP 状态码(如并发限流的 429)。 + // 仅用于错误分级(severity/classification);实际 wire 状态码仍为 200。 + IntendedStatus int +} + +// MarkOpsStreamError 记录一次就地 SSE 错误,供 ops 日志采集。 +// 采用「首个标记生效」策略:同一请求若先后补发多帧(如上游透传错误后又追加通用兜底帧), +// 保留最先记录的根因错误,而不是被后续的 "Upstream request failed" 覆盖。 +func MarkOpsStreamError(c *gin.Context, errType, message string, intendedStatus int) { + if c == nil { + return + } + if _, exists := c.Get(OpsStreamErrorKey); exists { + return + } + c.Set(OpsStreamErrorKey, OpsStreamError{ + ErrType: strings.TrimSpace(errType), + Message: strings.TrimSpace(message), + IntendedStatus: intendedStatus, + }) +} + +// GetOpsStreamError 返回本请求记录的就地 SSE 错误(若有)。 +func GetOpsStreamError(c *gin.Context) (OpsStreamError, bool) { + if c == nil { + return OpsStreamError{}, false + } + v, ok := c.Get(OpsStreamErrorKey) + if !ok { + return OpsStreamError{}, false + } + se, ok := v.(OpsStreamError) + return se, ok +} + // SetOpsUpstreamError is the exported wrapper for setOpsUpstreamError, used by // handler-layer code (e.g. failover-exhausted paths) that needs to record the // original upstream status code before mapping it to a client-facing code.