From fe56aa46b9b2c7309b9890950fe7be0ecbafccc1 Mon Sep 17 00:00:00 2001 From: wangbo Date: Tue, 4 Aug 2026 10:46:46 +0800 Subject: [PATCH] =?UTF-8?q?fix(errors):=20=E5=8C=BA=E5=88=86=E5=B9=B3?= =?UTF-8?q?=E5=8F=B0=E9=99=90=E6=B5=81=E5=B9=B6=E4=BF=9D=E7=95=99=E4=B8=8A?= =?UTF-8?q?=E6=B8=B8=E7=8A=B6=E6=80=81=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原因：公开错误层将平台并发限流误标为上游限流，并把多种上游 4xx 统一压成 400，影响定位和客户端处理。 影响：新增公开错误 source，平台限流使用 gateway_rate_limited，上游请求按安全分类返回对应状态；数据库与管理端继续保留原始错误码、消息和状态用于审计。 验证：Go 全量测试、pnpm test、pnpm lint、pnpm build、pnpm openapi、gofmt 和 diff 检查均通过。 --- apps/api/docs/swagger.json | 3 + apps/api/docs/swagger.yaml | 2 + apps/api/internal/httpapi/compat_protocol.go | 4 +- .../internal/httpapi/compat_protocol_test.go | 11 +++ .../httpapi/core_flow_integration_test.go | 4 +- apps/api/internal/httpapi/easyai_compat.go | 1 + apps/api/internal/httpapi/gemini_compat.go | 4 +- apps/api/internal/httpapi/handlers.go | 5 +- apps/api/internal/httpapi/public_errors.go | 45 +++++++-- .../internal/httpapi/public_errors_test.go | 96 +++++++++++++++++++ .../httpapi/rate_limit_error_detail_test.go | 13 ++- apps/api/internal/httpapi/response.go | 1 + apps/api/internal/httpapi/task_idempotency.go | 10 +- apps/api/internal/publicerror/metrics.go | 6 ++ apps/api/internal/publicerror/public_error.go | 56 ++++++++++- .../internal/publicerror/public_error_test.go | 34 ++++++- apps/api/internal/runner/service.go | 3 +- .../store/admin_task_redaction_test.go | 7 ++ apps/api/internal/store/runtime_types.go | 3 +- apps/api/internal/store/tasks_runtime.go | 2 +- packages/contracts/src/index.ts | 1 + 21 files changed, 289 insertions(+), 22 deletions(-) create mode 100644 apps/api/internal/httpapi/public_errors_test.go diff --git a/apps/api/docs/swagger.json b/apps/api/docs/swagger.json index e8a1159..4edc374 100644 --- a/apps/api/docs/swagger.json +++ b/apps/api/docs/swagger.json @@ -13457,6 +13457,9 @@ "retryable": { "type": "boolean" }, + "source": { + "type": "string" + }, "taskId": { "type": "string" }, diff --git a/apps/api/docs/swagger.yaml b/apps/api/docs/swagger.yaml index ced6113..2cb3387 100644 --- a/apps/api/docs/swagger.yaml +++ b/apps/api/docs/swagger.yaml @@ -2001,6 +2001,8 @@ definitions: type: integer retryable: type: boolean + source: + type: string taskId: type: string version: diff --git a/apps/api/internal/httpapi/compat_protocol.go b/apps/api/internal/httpapi/compat_protocol.go index bd88faf..9d36353 100644 --- a/apps/api/internal/httpapi/compat_protocol.go +++ b/apps/api/internal/httpapi/compat_protocol.go @@ -109,7 +109,9 @@ func writeProtocolError(w http.ResponseWriter, protocol string, status int, mess case clients.ProtocolVolcesContents: writeVolcesPublicError(w, standard) case clients.ProtocolKlingV1Omni, clients.ProtocolKlingV2Omni: - writeKelingCompatError(w, "", newKelingCompatError(status, kelingCompatBusinessCode(code, message), message)) + compatErr := newKelingCompatError(status, kelingCompatBusinessCode(code, message), message) + compatErr.PublicError = &standard + writeKelingCompatError(w, "", compatErr) default: writeOpenAIError(w, status, message, details, code) } diff --git a/apps/api/internal/httpapi/compat_protocol_test.go b/apps/api/internal/httpapi/compat_protocol_test.go index 0d17be4..6b7d20c 100644 --- a/apps/api/internal/httpapi/compat_protocol_test.go +++ b/apps/api/internal/httpapi/compat_protocol_test.go @@ -84,6 +84,15 @@ func TestProtocolErrorsUseCompatibleShapesWithStandardPublicErrors(t *testing.T) assertNoKeys(t, errorBody, "status", "taskId", "gateway_status") }, }, + { + name: "keling", protocol: clients.ProtocolKlingV1Omni, status: http.StatusTooManyRequests, + assertBody: func(t *testing.T, body map[string]any) { + errorBody := requireObject(t, body["error"]) + if errorBody["code"] != "gateway_rate_limited" || errorBody["source"] != "gateway" || errorBody["httpStatus"] != float64(http.StatusTooManyRequests) { + t.Fatalf("unexpected Kling-compatible error: %+v", body) + } + }, + }, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { @@ -93,6 +102,8 @@ func TestProtocolErrorsUseCompatibleShapesWithStandardPublicErrors(t *testing.T) code = "rate_limit" } else if test.name == "volces" { code = "upstream_submission_unknown" + } else if test.name == "keling" { + code = "gateway_rate_limited" } writeProtocolError(recorder, test.protocol, test.status, "failed", nil, code) if recorder.Code != test.status { diff --git a/apps/api/internal/httpapi/core_flow_integration_test.go b/apps/api/internal/httpapi/core_flow_integration_test.go index a8120b4..f045c65 100644 --- a/apps/api/internal/httpapi/core_flow_integration_test.go +++ b/apps/api/internal/httpapi/core_flow_integration_test.go @@ -1233,8 +1233,8 @@ WHERE reference_type = 'gateway_task' "simulationDurationMs": 5, "messages": []map[string]any{{"role": "user", "content": "second"}}, }, "rate-limit-second-"+suffixText, http.StatusTooManyRequests, nil, &rateLimitTaskTwo.Task) - if rateLimitTaskTwo.Task.Status != "failed" || rateLimitTaskTwo.Task.ErrorCode != "rate_limit" { - t.Fatalf("runtime policy rate limit should fail second task with rate_limit: %+v", rateLimitTaskTwo.Task) + if rateLimitTaskTwo.Task.Status != "failed" || rateLimitTaskTwo.Task.ErrorCode != "gateway_rate_limited" { + t.Fatalf("runtime policy rate limit should fail second task with gateway_rate_limited: %+v", rateLimitTaskTwo.Task) } var asyncRateLimitTask struct { TaskID string `json:"taskId"` diff --git a/apps/api/internal/httpapi/easyai_compat.go b/apps/api/internal/httpapi/easyai_compat.go index 5beeb9d..cc8fc20 100644 --- a/apps/api/internal/httpapi/easyai_compat.go +++ b/apps/api/internal/httpapi/easyai_compat.go @@ -68,6 +68,7 @@ func writeEasyAIAsyncError(w http.ResponseWriter, status int, message string, de "status": status, "code": code, "category": standard.Category, + "source": standard.Source, "httpStatus": standard.HTTPStatus, "retryable": standard.Retryable, "action": standard.Action, diff --git a/apps/api/internal/httpapi/gemini_compat.go b/apps/api/internal/httpapi/gemini_compat.go index 4b8a651..6e9e099 100644 --- a/apps/api/internal/httpapi/gemini_compat.go +++ b/apps/api/internal/httpapi/gemini_compat.go @@ -235,7 +235,7 @@ func (s *Server) geminiGenerateContent(w http.ResponseWriter, r *http.Request) { } status := http.StatusConflict if task.Status == "failed" || task.Status == "cancelled" { - status = storedTaskErrorStatus(task.ErrorCode) + status = taskErrorHTTPStatus(task) } writeGeminiTaskError(status, firstNonEmpty(task.ErrorMessage, task.Error, task.Message, "task is not complete"), nil, firstNonEmpty(task.ErrorCode, task.Status)) return @@ -263,7 +263,7 @@ func (s *Server) geminiGenerateContent(w http.ResponseWriter, r *http.Request) { return } if task.Status != "succeeded" { - status := storedTaskErrorStatus(task.ErrorCode) + status := taskErrorHTTPStatus(task) writeGeminiTaskError(status, firstNonEmpty(task.ErrorMessage, task.Error, task.Message, "task failed"), nil, firstNonEmpty(task.ErrorCode, task.Status)) return } diff --git a/apps/api/internal/httpapi/handlers.go b/apps/api/internal/httpapi/handlers.go index 765ec89..209d7e6 100644 --- a/apps/api/internal/httpapi/handlers.go +++ b/apps/api/internal/httpapi/handlers.go @@ -1286,7 +1286,7 @@ func (s *Server) createTask(kind string, compatible bool) http.Handler { } status := http.StatusConflict if task.Status == "failed" || task.Status == "cancelled" { - status = storedTaskErrorStatus(task.ErrorCode) + status = taskErrorHTTPStatus(task) } writeTaskError(status, firstNonEmpty(task.ErrorMessage, task.Error, task.Message, "task is not complete"), nil, firstNonEmpty(task.ErrorCode, task.Status)) return @@ -1706,6 +1706,9 @@ func statusFromRunError(err error) int { case errors.Is(err, store.ErrInsufficientWalletBalance): return http.StatusPaymentRequired default: + if status := clients.ErrorResponseMetadata(err).StatusCode; status >= 400 && status <= 599 { + return status + } return http.StatusBadGateway } } diff --git a/apps/api/internal/httpapi/public_errors.go b/apps/api/internal/httpapi/public_errors.go index 1fdcd6b..b154250 100644 --- a/apps/api/internal/httpapi/public_errors.go +++ b/apps/api/internal/httpapi/public_errors.go @@ -10,8 +10,8 @@ import ( func publicGatewayTask(task store.GatewayTask) store.GatewayTask { if task.Status == "failed" || task.Status == "cancelled" || task.ErrorCode != "" || task.ErrorMessage != "" { - value := publicerror.FromFields(task.ErrorCode, firstNonEmpty(task.ErrorMessage, task.Error), storedTaskErrorStatus(task.ErrorCode), false) - if task.PublicError != nil && task.PublicError.Code != "" { + value := publicerror.FromFields(task.ErrorCode, firstNonEmpty(task.ErrorMessage, task.Error), taskErrorHTTPStatus(task), false) + if task.PublicError != nil && task.PublicError.Code != "" && task.PublicError.Source != "" { value = *task.PublicError } value = publicerror.WithIDs(value, task.RequestID, task.ID) @@ -21,6 +21,7 @@ func publicGatewayTask(task store.GatewayTask) store.GatewayTask { task.ErrorMessage = value.Message task.Error = value.Message } + task.Attempts = append([]store.TaskAttempt(nil), task.Attempts...) for index := range task.Attempts { attempt := &task.Attempts[index] if attempt.ErrorCode == "" && attempt.ErrorMessage == "" { @@ -31,7 +32,7 @@ func publicGatewayTask(task store.GatewayTask) store.GatewayTask { status = http.StatusBadGateway } value := publicerror.FromFields(attempt.ErrorCode, attempt.ErrorMessage, status, attempt.Retryable) - if attempt.PublicError != nil && attempt.PublicError.Code != "" { + if attempt.PublicError != nil && attempt.PublicError.Code != "" && attempt.PublicError.Source != "" { value = *attempt.PublicError } value = publicerror.WithIDs(value, attempt.RequestID, task.ID) @@ -43,16 +44,48 @@ func publicGatewayTask(task store.GatewayTask) store.GatewayTask { } func publicTaskError(task store.GatewayTask) publicerror.Error { - if task.PublicError != nil && task.PublicError.Code != "" { + if task.PublicError != nil && task.PublicError.Code != "" && task.PublicError.Source != "" { return publicerror.WithIDs(*task.PublicError, task.RequestID, task.ID) } - status := storedTaskErrorStatus(task.ErrorCode) + status := taskErrorHTTPStatus(task) if status <= 0 { status = http.StatusBadGateway } return publicerror.WithIDs(publicerror.FromFields(task.ErrorCode, firstNonEmpty(task.ErrorMessage, task.Error, task.Message), status, false), task.RequestID, task.ID) } +func taskErrorHTTPStatus(task store.GatewayTask) int { + if task.PublicError != nil && task.PublicError.Source != "" && task.PublicError.HTTPStatus >= 400 { + return task.PublicError.HTTPStatus + } + for index := len(task.Attempts) - 1; index >= 0; index-- { + if status := task.Attempts[index].StatusCode; status >= 400 && status <= 599 { + return status + } + } + if status := publicErrorInt(task.Metrics["statusCode"]); status >= 400 && status <= 599 { + return status + } + return storedTaskErrorStatus(task.ErrorCode) +} + +func publicErrorInt(value any) int { + switch typed := value.(type) { + case int: + return typed + case int32: + return int(typed) + case int64: + return int(typed) + case float32: + return int(typed) + case float64: + return int(typed) + default: + return 0 + } +} + func publicTaskList(items []store.GatewayTask) []store.GatewayTask { out := make([]store.GatewayTask, len(items)) for index, task := range items { @@ -63,7 +96,7 @@ func publicTaskList(items []store.GatewayTask) []store.GatewayTask { func publicErrorMap(value publicerror.Error) map[string]any { out := map[string]any{ - "code": value.Code, "message": value.Message, "category": value.Category, + "code": value.Code, "message": value.Message, "category": value.Category, "source": value.Source, "httpStatus": value.HTTPStatus, "retryable": value.Retryable, "action": value.Action, "version": value.Version, } diff --git a/apps/api/internal/httpapi/public_errors_test.go b/apps/api/internal/httpapi/public_errors_test.go new file mode 100644 index 0000000..ab8aed1 --- /dev/null +++ b/apps/api/internal/httpapi/public_errors_test.go @@ -0,0 +1,96 @@ +package httpapi + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/clients" + "github.com/easyai/easyai-ai-gateway/apps/api/internal/publicerror" + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +func TestPublicHTTPErrorReportsSourceAndPreservesSafeUpstreamStatus(t *testing.T) { + upstream := httptest.NewRecorder() + writeProtocolError(upstream, clients.ProtocolOpenAIImages, http.StatusNotFound, "404 page not found at private upstream route", nil, "http_404") + if upstream.Code != http.StatusNotFound { + t.Fatalf("upstream status = %d, want 404; body=%s", upstream.Code, upstream.Body.String()) + } + var upstreamBody map[string]any + if err := json.Unmarshal(upstream.Body.Bytes(), &upstreamBody); err != nil { + t.Fatal(err) + } + upstreamError := requireObject(t, upstreamBody["error"]) + upstreamDetails := requireObject(t, upstreamError["details"]) + upstreamPublic := requireObject(t, upstreamDetails["publicError"]) + if upstreamError["code"] != "upstream_not_found" || upstreamPublic["source"] != "upstream" { + t.Fatalf("unexpected upstream response: %+v", upstreamBody) + } + if strings.Contains(upstream.Body.String(), "private upstream route") { + t.Fatalf("upstream response leaked raw message: %s", upstream.Body.String()) + } + + gateway := httptest.NewRecorder() + writeError(gateway, http.StatusTooManyRequests, "concurrency limit is saturated and queueing is disabled", "gateway_rate_limited") + if gateway.Code != http.StatusTooManyRequests { + t.Fatalf("gateway status = %d, want 429; body=%s", gateway.Code, gateway.Body.String()) + } + var gatewayBody map[string]any + if err := json.Unmarshal(gateway.Body.Bytes(), &gatewayBody); err != nil { + t.Fatal(err) + } + gatewayError := requireObject(t, gatewayBody["error"]) + if gatewayError["code"] != "gateway_rate_limited" || gatewayError["source"] != "gateway" { + t.Fatalf("unexpected gateway response: %+v", gatewayBody) + } +} + +func TestPublicGatewayTaskSanitizesCopyAndKeepsRawAuditFields(t *testing.T) { + rawMessage := `404 page not found: {"privateProject":"secret"}` + task := store.GatewayTask{ + ID: "task-1", + Status: "failed", + ErrorCode: "http_404", + ErrorMessage: rawMessage, + Attempts: []store.TaskAttempt{{ + AttemptNo: 1, + Status: "failed", + StatusCode: http.StatusNotFound, + ErrorCode: "http_404", + ErrorMessage: rawMessage, + }}, + } + + public := publicGatewayTask(task) + if public.ErrorCode != "upstream_not_found" || public.ErrorMessage == rawMessage || public.PublicError == nil || public.PublicError.Source != "upstream" { + t.Fatalf("unexpected public task error: %+v", public) + } + if strings.Contains(public.ErrorMessage, "secret") || public.Attempts[0].ErrorMessage == rawMessage { + t.Fatalf("public task leaked upstream details: %+v", public) + } + if task.ErrorCode != "http_404" || task.ErrorMessage != rawMessage || task.Attempts[0].ErrorCode != "http_404" || task.Attempts[0].ErrorMessage != rawMessage { + t.Fatalf("public conversion mutated raw audit fields: %+v", task) + } +} + +func TestTaskErrorHTTPStatusRebuildsLegacySnapshotFromAttempt(t *testing.T) { + legacy := publicerror.Error{ + Code: "upstream_request_rejected", + HTTPStatus: http.StatusBadRequest, + Version: "v1", + } + task := store.GatewayTask{ + ErrorCode: "http_422", + PublicError: &legacy, + Attempts: []store.TaskAttempt{{StatusCode: http.StatusUnprocessableEntity}}, + } + if got := taskErrorHTTPStatus(task); got != http.StatusUnprocessableEntity { + t.Fatalf("taskErrorHTTPStatus = %d, want %d", got, http.StatusUnprocessableEntity) + } + public := publicTaskError(task) + if public.Code != "upstream_unprocessable_request" || public.HTTPStatus != http.StatusUnprocessableEntity || public.Source != "upstream" { + t.Fatalf("legacy public snapshot was not rebuilt: %+v", public) + } +} diff --git a/apps/api/internal/httpapi/rate_limit_error_detail_test.go b/apps/api/internal/httpapi/rate_limit_error_detail_test.go index a4089bd..7b6dbba 100644 --- a/apps/api/internal/httpapi/rate_limit_error_detail_test.go +++ b/apps/api/internal/httpapi/rate_limit_error_detail_test.go @@ -1,12 +1,14 @@ package httpapi import ( + "fmt" "net/http" "net/http/httptest" "strings" "testing" "time" + "github.com/easyai/easyai-ai-gateway/apps/api/internal/clients" "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" ) @@ -66,7 +68,7 @@ func TestQueueFullAndTimeoutPublicContracts(t *testing.T) { } recorder := httptest.NewRecorder() applyRunErrorHeaders(recorder, queueErr) - if statusFromRunError(queueErr) != http.StatusTooManyRequests || runErrorCode(queueErr) != "rate_limit" { + if statusFromRunError(queueErr) != http.StatusTooManyRequests || runErrorCode(queueErr) != "gateway_rate_limited" { t.Fatalf("queue full contract status=%d code=%s", statusFromRunError(queueErr), runErrorCode(queueErr)) } if recorder.Header().Get("Retry-After") != "2" { @@ -84,6 +86,15 @@ func TestQueueFullAndTimeoutPublicContracts(t *testing.T) { } } +func TestStatusFromRunErrorPreservesUpstreamHTTPStatus(t *testing.T) { + for _, status := range []int{http.StatusBadRequest, http.StatusNotFound, http.StatusMethodNotAllowed, http.StatusConflict, http.StatusRequestEntityTooLarge, http.StatusUnsupportedMediaType, http.StatusUnprocessableEntity, http.StatusTooManyRequests} { + err := &clients.ClientError{Code: fmt.Sprintf("http_%d", status), Message: "raw upstream error", StatusCode: status} + if got := statusFromRunError(err); got != status { + t.Fatalf("statusFromRunError(%d) = %d", status, got) + } + } +} + func TestRunErrorMessageIncludesRateLimitSummary(t *testing.T) { message := runErrorMessage(&store.RateLimitExceededError{ ScopeType: "user_group", diff --git a/apps/api/internal/httpapi/response.go b/apps/api/internal/httpapi/response.go index 79c7126..8ac7c99 100644 --- a/apps/api/internal/httpapi/response.go +++ b/apps/api/internal/httpapi/response.go @@ -33,6 +33,7 @@ func writeErrorWithDetails(w http.ResponseWriter, status int, message string, de "status": status, "code": standard.Code, "category": standard.Category, + "source": standard.Source, "httpStatus": standard.HTTPStatus, "retryable": standard.Retryable, "action": standard.Action, diff --git a/apps/api/internal/httpapi/task_idempotency.go b/apps/api/internal/httpapi/task_idempotency.go index 32c84d6..b4c223e 100644 --- a/apps/api/internal/httpapi/task_idempotency.go +++ b/apps/api/internal/httpapi/task_idempotency.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "net/http" + "strconv" "strings" "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" @@ -52,7 +53,7 @@ func writeIdempotentTaskReplay(w http.ResponseWriter, task store.GatewayTask, co writeJSON(w, http.StatusOK, task.Result) return } - status := storedTaskErrorStatus(task.ErrorCode) + status := taskErrorHTTPStatus(task) message := strings.TrimSpace(task.ErrorMessage) if message == "" { message = strings.TrimSpace(task.Error) @@ -81,9 +82,14 @@ func storedTaskErrorStatus(code string) int { return http.StatusBadRequest case "no_model_candidate", "cloned_voice_not_found": return http.StatusNotFound - case "rate_limit", "platform_cooling_down", "model_cooling_down": + case "gateway_rate_limited", "rate_limit", "platform_cooling_down", "model_cooling_down": return http.StatusTooManyRequests default: + if strings.HasPrefix(strings.TrimSpace(code), "http_") { + if status, err := strconv.Atoi(strings.TrimPrefix(strings.TrimSpace(code), "http_")); err == nil && status >= 400 && status <= 599 { + return status + } + } return http.StatusBadGateway } } diff --git a/apps/api/internal/publicerror/metrics.go b/apps/api/internal/publicerror/metrics.go index e8d1cda..9aed837 100644 --- a/apps/api/internal/publicerror/metrics.go +++ b/apps/api/internal/publicerror/metrics.go @@ -15,6 +15,7 @@ var observed = struct { upstreamInvalidResponse atomic.Uint64 upstreamAuthFailed atomic.Uint64 upstreamRequestRejected atomic.Uint64 + gatewayRateLimited atomic.Uint64 storageWriteFailed atomic.Uint64 storageReadFailed atomic.Uint64 resultExpired atomic.Uint64 @@ -39,6 +40,10 @@ func Observe(value Error) { observed.upstreamAuthFailed.Add(1) case "upstream_request_rejected": observed.upstreamRequestRejected.Add(1) + case "upstream_invalid_request", "upstream_not_found", "upstream_method_not_allowed", "upstream_conflict", "upstream_payload_too_large", "upstream_unsupported_media_type", "upstream_unprocessable_request": + observed.upstreamRequestRejected.Add(1) + case "gateway_rate_limited": + observed.gatewayRateLimited.Add(1) case "storage_write_failed": observed.storageWriteFailed.Add(1) case "storage_read_failed": @@ -63,6 +68,7 @@ func MetricSnapshot() []MetricCount { {Code: "upstream_invalid_response", Count: observed.upstreamInvalidResponse.Load()}, {Code: "upstream_auth_failed", Count: observed.upstreamAuthFailed.Load()}, {Code: "upstream_request_rejected", Count: observed.upstreamRequestRejected.Load()}, + {Code: "gateway_rate_limited", Count: observed.gatewayRateLimited.Load()}, {Code: "storage_write_failed", Count: observed.storageWriteFailed.Load()}, {Code: "storage_read_failed", Count: observed.storageReadFailed.Load()}, {Code: "result_expired", Count: observed.resultExpired.Load()}, diff --git a/apps/api/internal/publicerror/public_error.go b/apps/api/internal/publicerror/public_error.go index 9e74398..bba65be 100644 --- a/apps/api/internal/publicerror/public_error.go +++ b/apps/api/internal/publicerror/public_error.go @@ -2,6 +2,7 @@ package publicerror import ( "net/http" + "strconv" "strings" ) @@ -11,6 +12,7 @@ type Error struct { Code string `json:"code"` Message string `json:"message"` Category string `json:"category"` + Source string `json:"source"` HTTPStatus int `json:"httpStatus"` Retryable bool `json:"retryable"` Action string `json:"action"` @@ -79,6 +81,8 @@ func mappedError(code string, message string, status int, retryable bool) (Error return newError("upstream_connection_interrupted", "The upstream connection was interrupted before a complete response was received.", "upstream", http.StatusBadGateway, true, "retry"), true case "upstream_timeout", "timeout", "context_deadline_exceeded": return newError("upstream_timeout", "The upstream service did not respond in time.", "upstream", http.StatusGatewayTimeout, true, "retry"), true + case "gateway_rate_limited": + return newError("gateway_rate_limited", "The gateway rate limit was reached.", "rate_limit", http.StatusTooManyRequests, true, "retry_after"), true case "rate_limit", "upstream_rate_limited", "too_many_requests": return newError("upstream_rate_limited", "The upstream service rate limit was reached.", "rate_limit", http.StatusTooManyRequests, true, "retry_after"), true case "upstream_unavailable", "upstream_overloaded", "service_unavailable", "bad_gateway", "server_error", "provider_failed": @@ -107,8 +111,11 @@ func mappedError(code string, message string, status int, retryable bool) (Error if (status == http.StatusUnauthorized || status == http.StatusForbidden) && (strings.HasPrefix(code, "http_") || strings.Contains(code, "provider")) { return newError("upstream_auth_failed", "The upstream service rejected the configured credentials.", "upstream", http.StatusBadGateway, false, "contact_support"), true } + if strings.HasPrefix(code, "http_") && status >= 500 { + return newError("upstream_unavailable", "The upstream service is temporarily unavailable.", "upstream", http.StatusServiceUnavailable, true, "retry"), true + } if strings.HasPrefix(code, "http_") || (code == "provider_failed" && status > 0 && status < 500) { - return newError("upstream_request_rejected", "The upstream service rejected the request.", "upstream", http.StatusBadRequest, false, "fix_request"), true + return upstreamHTTPError(status), true } if status >= 500 && code != "upstream_submission_unknown" && (strings.Contains(code, "upstream") || strings.Contains(code, "provider")) { return newError("upstream_unavailable", "The upstream service is temporarily unavailable.", "upstream", http.StatusServiceUnavailable, true, "retry"), true @@ -142,7 +149,45 @@ func retryAction(retryable bool) string { } func newError(code string, message string, category string, status int, retryable bool, action string) Error { - return Error{Code: code, Message: message, Category: category, HTTPStatus: status, Retryable: retryable, Action: action, Version: Version} + return Error{Code: code, Message: message, Category: category, Source: errorSource(code, category), HTTPStatus: status, Retryable: retryable, Action: action, Version: Version} +} + +func upstreamHTTPError(status int) Error { + if status < 400 || status >= 500 { + status = http.StatusBadRequest + } + switch status { + case http.StatusBadRequest: + return newError("upstream_invalid_request", "The upstream service rejected the request parameters.", "upstream", status, false, "fix_request") + case http.StatusNotFound: + return newError("upstream_not_found", "The upstream service could not find the requested resource.", "upstream", status, false, "contact_support") + case http.StatusMethodNotAllowed: + return newError("upstream_method_not_allowed", "The upstream service did not accept the configured request method.", "upstream", status, false, "contact_support") + case http.StatusConflict: + return newError("upstream_conflict", "The upstream service reported a request conflict.", "upstream", status, false, "fix_request") + case http.StatusRequestEntityTooLarge: + return newError("upstream_payload_too_large", "The upstream service rejected the request because its payload was too large.", "upstream", status, false, "fix_request") + case http.StatusUnsupportedMediaType: + return newError("upstream_unsupported_media_type", "The upstream service rejected the request media type.", "upstream", status, false, "fix_request") + case http.StatusUnprocessableEntity: + return newError("upstream_unprocessable_request", "The upstream service could not process the request parameters.", "upstream", status, false, "fix_request") + default: + return newError("upstream_request_rejected", "The upstream service rejected the request.", "upstream", status, false, "fix_request") + } +} + +func errorSource(code string, category string) string { + code = strings.ToLower(strings.TrimSpace(code)) + switch { + case strings.HasPrefix(code, "upstream_"): + return "upstream" + case strings.HasPrefix(code, "gateway_"), category == "gateway", category == "storage": + return "gateway" + case category == "request": + return "client" + default: + return "gateway" + } } func sensitiveTransportMessage(message string) bool { @@ -156,8 +201,13 @@ func sensitiveTransportMessage(message string) bool { } func statusFromCode(code string) int { + if strings.HasPrefix(code, "http_") { + if status, err := strconv.Atoi(strings.TrimPrefix(code, "http_")); err == nil && status >= 400 && status <= 599 { + return status + } + } switch code { - case "rate_limit", "upstream_rate_limited", "too_many_requests": + case "gateway_rate_limited", "rate_limit", "upstream_rate_limited", "too_many_requests": return http.StatusTooManyRequests case "upstream_timeout", "timeout", "context_deadline_exceeded": return http.StatusGatewayTimeout diff --git a/apps/api/internal/publicerror/public_error_test.go b/apps/api/internal/publicerror/public_error_test.go index 374db31..f290dc1 100644 --- a/apps/api/internal/publicerror/public_error_test.go +++ b/apps/api/internal/publicerror/public_error_test.go @@ -44,7 +44,11 @@ func TestProviderHTTPErrorDoesNotExposeProviderBody(t *testing.T) { wantCode string wantStatus int }{ - {code: "http_400", status: http.StatusBadRequest, wantCode: "upstream_request_rejected", wantStatus: http.StatusBadRequest}, + {code: "http_400", status: http.StatusBadRequest, wantCode: "upstream_invalid_request", wantStatus: http.StatusBadRequest}, + {code: "http_404", status: http.StatusNotFound, wantCode: "upstream_not_found", wantStatus: http.StatusNotFound}, + {code: "http_405", status: http.StatusMethodNotAllowed, wantCode: "upstream_method_not_allowed", wantStatus: http.StatusMethodNotAllowed}, + {code: "http_413", status: http.StatusRequestEntityTooLarge, wantCode: "upstream_payload_too_large", wantStatus: http.StatusRequestEntityTooLarge}, + {code: "http_422", status: http.StatusUnprocessableEntity, wantCode: "upstream_unprocessable_request", wantStatus: http.StatusUnprocessableEntity}, {code: "auth_failed", status: http.StatusUnauthorized, wantCode: "upstream_auth_failed", wantStatus: http.StatusBadGateway}, {code: "provider_failed", status: http.StatusForbidden, wantCode: "upstream_auth_failed", wantStatus: http.StatusBadGateway}, {code: "provider_failed", status: http.StatusTooManyRequests, wantCode: "upstream_rate_limited", wantStatus: http.StatusTooManyRequests}, @@ -55,12 +59,40 @@ func TestProviderHTTPErrorDoesNotExposeProviderBody(t *testing.T) { if got.Code != test.wantCode || got.HTTPStatus != test.wantStatus { t.Fatalf("%s: unexpected public error: %+v", test.code, got) } + if got.Source != "upstream" { + t.Fatalf("%s: source = %q, want upstream", test.code, got.Source) + } if strings.Contains(got.Message, "private-a") || strings.Contains(got.Message, "secret-project") { t.Fatalf("%s: provider body leaked: %+v", test.code, got) } } } +func TestGatewayRateLimitIsDistinctFromUpstreamRateLimit(t *testing.T) { + gateway := FromFields("gateway_rate_limited", "concurrency limit is saturated and queueing is disabled", http.StatusTooManyRequests, true) + if gateway.Code != "gateway_rate_limited" || gateway.Source != "gateway" || gateway.HTTPStatus != http.StatusTooManyRequests { + t.Fatalf("unexpected gateway rate limit: %+v", gateway) + } + if strings.Contains(gateway.Message, "concurrency") { + t.Fatalf("gateway rate limit leaked internal details: %+v", gateway) + } + + upstream := FromFields("rate_limit", "provider quota exhausted for private account", http.StatusTooManyRequests, true) + if upstream.Code != "upstream_rate_limited" || upstream.Source != "upstream" || upstream.HTTPStatus != http.StatusTooManyRequests { + t.Fatalf("unexpected upstream rate limit: %+v", upstream) + } + if upstream.Message == "provider quota exhausted for private account" { + t.Fatalf("upstream rate limit leaked provider details: %+v", upstream) + } +} + +func TestHTTPCodeDerivesOriginalUpstreamStatusWhenStatusIsMissing(t *testing.T) { + got := FromFields("http_404", "404 page not found", 0, false) + if got.Code != "upstream_not_found" || got.Source != "upstream" || got.HTTPStatus != http.StatusNotFound { + t.Fatalf("unexpected derived upstream error: %+v", got) + } +} + func TestValidationGateKeepsStablePublicCode(t *testing.T) { got := FromFields("validation_in_progress", "new production tasks are paused while validation is running", http.StatusServiceUnavailable, true) if got.Code != "validation_in_progress" || got.HTTPStatus != http.StatusServiceUnavailable || !got.Retryable { diff --git a/apps/api/internal/runner/service.go b/apps/api/internal/runner/service.go index d27b324..de069f5 100644 --- a/apps/api/internal/runner/service.go +++ b/apps/api/internal/runner/service.go @@ -1363,7 +1363,7 @@ func (s *Service) runCandidate( s.observeProviderQuotaWait(limitErr.Metric) } retryable := store.RateLimitRetryable(err) - clientErr := &clients.ClientError{Code: "rate_limit", Message: err.Error(), Retryable: retryable} + clientErr := &clients.ClientError{Code: clients.ErrorCode(err), Message: err.Error(), Retryable: retryable} return clients.Response{}, &localRateLimitError{clientErr: clientErr, cause: err, retryAfter: localRateLimitRetryAfter(err)} } attemptOwnedLeases := append([]store.ConcurrencyLease(nil), limitResult.Leases...) @@ -2026,6 +2026,7 @@ func (s *Service) failTask(ctx context.Context, taskID string, executionToken st ExecutionToken: executionToken, Code: code, Message: message, + StatusCode: clients.ErrorResponseMetadata(cause).StatusCode, Result: buildFailureResult(code, message, requestID, cause), RequestID: requestID, Metrics: metrics, diff --git a/apps/api/internal/store/admin_task_redaction_test.go b/apps/api/internal/store/admin_task_redaction_test.go index 21bc9b9..f5f7bd9 100644 --- a/apps/api/internal/store/admin_task_redaction_test.go +++ b/apps/api/internal/store/admin_task_redaction_test.go @@ -8,6 +8,8 @@ import ( func TestMaskAdminGatewayTaskRecursivelyMasksSecretsWithoutMutatingSource(t *testing.T) { source := AdminGatewayTask{ GatewayTask: GatewayTask{ + ErrorCode: "http_404", + ErrorMessage: "404 page not found from upstream route /private/v1/images", Request: map[string]any{ "model": "example", "headers": map[string]any{ @@ -20,12 +22,17 @@ func TestMaskAdminGatewayTaskRecursivelyMasksSecretsWithoutMutatingSource(t *tes "nested": []any{map[string]any{"password": "private-password"}}, }, Attempts: []TaskAttempt{{ + ErrorCode: "http_404", + ErrorMessage: "404 page not found from upstream route /private/v1/images", RequestSnapshot: map[string]any{"client_secret": "private-client-secret"}, }}, }, } masked := MaskAdminGatewayTask(source) + if masked.ErrorCode != source.ErrorCode || masked.ErrorMessage != source.ErrorMessage || masked.Attempts[0].ErrorMessage != source.Attempts[0].ErrorMessage { + t.Fatalf("admin error audit fields must preserve raw upstream values: %#v", masked) + } headers := masked.Request["headers"].(map[string]any) if headers["Authorization"] != maskedAdminTaskValue || headers["X-Api-Key"] != maskedAdminTaskValue { t.Fatalf("sensitive headers were not masked: %#v", headers) diff --git a/apps/api/internal/store/runtime_types.go b/apps/api/internal/store/runtime_types.go index fac8757..71ea9e1 100644 --- a/apps/api/internal/store/runtime_types.go +++ b/apps/api/internal/store/runtime_types.go @@ -94,7 +94,7 @@ func (e *RateLimitExceededError) Error() string { } func (e *RateLimitExceededError) ErrorCode() string { - return "rate_limit" + return "gateway_rate_limited" } func (e *RateLimitExceededError) Unwrap() error { @@ -349,6 +349,7 @@ type FinishTaskFailureInput struct { ExecutionToken string Code string Message string + StatusCode int Result map[string]any RequestID string Metrics map[string]any diff --git a/apps/api/internal/store/tasks_runtime.go b/apps/api/internal/store/tasks_runtime.go index b5fd34f..345ef04 100644 --- a/apps/api/internal/store/tasks_runtime.go +++ b/apps/api/internal/store/tasks_runtime.go @@ -2068,7 +2068,7 @@ func (s *Store) FinishTaskFailure(ctx context.Context, input FinishTaskFailureIn metricsJSON, _ := json.Marshal(sanitizeJSONForStorage(emptyObjectIfNil(input.Metrics))) resultJSON, _ := json.Marshal(minimalTaskResult(nil)) message := truncateUTF8Bytes(input.Message, 2048) - publicErrorJSON := encodePublicErrorSnapshot(input.Code, message, 0, true, input.RequestID, input.TaskID) + publicErrorJSON := encodePublicErrorSnapshot(input.Code, message, input.StatusCode, true, input.RequestID, input.TaskID) finalizedAdmission := false err := s.beginTransaction(ctx, func(tx pgx.Tx) error { tag, err := tx.Exec(ctx, ` diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 431e6aa..f81aa82 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -1255,6 +1255,7 @@ export interface PublicErrorV1 { code: string; message: string; category: string; + source: 'gateway' | 'upstream' | 'client' | string; httpStatus: number; retryable: boolean; action: string;