github/sub2api
4
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-21 11:25:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #6179 from wwcchh0123/codex/upstream-error-proxy-attribution

Browse Source 
fix(ops): preserve event-time proxy attribution for upstream errors
This commit is contained in:
Wesley Liddick
2026-09-03 15:34:59 +08:00
committed by GitHub
parent 5097b31457 abc07bb079
commit 432ebc498c
55 changed files with 1472 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/dev-docs/upstream-error-proxy-attribution.md
+136
View File
@@ -0,0 +1,136 @@
# Upstream error proxy attribution
## Scope
Gateway inference errors are stored in `ops_error_logs.upstream_errors`. Each
array item represents one upstream attempt, including same-account retries and
cross-account failover attempts. Proxy attribution is stored on that item; it
must not be reconstructed later from `accounts.proxy_id`.
This repository has one gateway implementation under `backend/internal/service`.
There is no separate `gateway/backend` source copy to synchronize.
## Event contract
Every newly persisted event contains these fields. `proxy_id` is emitted even
when its value is JSON `null`:
| Field | Meaning |
| --- | --- |
| `proxy_id` | Managed proxy ID used by the attempt; `null` for direct or unknown routes. |
| `proxy_name` | Snapshotted proxy name, `direct/no_proxy`, or `unknown`. |
Only proxy ID and name are stored. Proxy URL, protocol endpoint, host, port,
username, password, and authorization data are excluded.
The two sentinel names have strict meanings:
- `direct/no_proxy`: the transport was explicitly given no proxy.
- `unknown`: the route cannot be proven from event-time evidence. This includes
legacy events, pre-transport credential failures, and OpenAI WebSocket use of
the default HTTP client.
Invariant: `proxy_id` is `null` if and only if `proxy_name` is one of the two
sentinels. A managed proxy with a blank name (defensive only; `proxies.name` is
non-empty) is labeled `proxy`. Both the struct normalizer used at append time
and the JSON normalizer used on detail reads enforce this, so an event can never
carry `proxy_id=null` together with a real proxy name.
## Why the account snapshot is event-time evidence
Account selection loads `ProxyID` and `Proxy` into one in-memory snapshot. HTTP,
TLS-fingerprint, Bedrock, Gemini, Antigravity, OpenAI, and OpenAI WebSocket
forwarding use that snapshot for managed proxy routing.
Inference transports use a managed proxy only when both the binding ID and its
hydrated proxy object are present; the attribution accessors use the same rule.
OpenAI WebSocket additionally retains the existing default-client environment
proxy behavior when that snapshot does not yield a usable proxy. The transport
does not switch to a database backup proxy during a request.
Expired proxy fallback happens before scheduling by atomically rewriting the
account binding:
- backup proxy: `proxy_id` becomes the backup ID and
`proxy_fallback_origin_id` stores the expired proxy ID;
- direct fallback: `proxy_id` becomes null and
`proxy_fallback_origin_id` stores the expired proxy ID.
Custom Anthropic relays receive the same snapshotted proxy as their encoded
relay proxy parameter. OpenAI WebSocket requests keep the existing transport
behavior: configured account proxies use the proxy client, while an empty
account proxy leaves the HTTP client unset so `coder/websocket` uses
`http.DefaultClient`, including its `HTTP_PROXY`, `HTTPS_PROXY`, and `NO_PROXY`
handling.
For WebSocket errors without a usable account proxy, the event stores
`proxy_id=null` and `proxy_name=unknown`. It does not infer whether the default
client selected an environment proxy or a direct connection after the fact.
This applies to every event produced while a request is on the WebSocket
transport, including WS handshake/fallback errors and the first-output timeout
raised by the WS passthrough relay; `opsUpstreamWSProxyAttribution` is the
accessor for those sites and never returns `direct/no_proxy`.
Credential acquisition failures that occur before the inference transport is
opened are also recorded as unknown. The account's proxy binding is not treated
as proof that the credential operation or an inference request used that route.
Each failure site constructs a complete `OpsUpstreamErrorEvent`. Its literal
sets `proxy_id` and `proxy_name` from the same route input used by the attempt.
Credential-free accessors copy only the managed proxy ID and name. A fallback
proxy needs no special event flag: its event-time ID and name are the historical
route evidence. Shared helpers that assemble the event on behalf of a caller
(`handleUpstreamTransportError`, `handleOpenAIUpstreamTransportError`,
`appendOpenAICompactFallbackRetryOps`, `recordOpenAIRawStreamTruncation`,
`newOpenAIFirstOutputTimeoutError`) stamp attribution themselves, so a caller
cannot forget it. Every non-test `OpsUpstreamErrorEvent{` literal in
`backend/internal/service` must either set `ProxyName` or be passed to one of
those helpers; grep for literals missing `ProxyName` before merging changes
that add a failure site.
`appendOpsUpstreamError` accepts only the completed event. It does not receive
an `Account`, query current account state, or infer proxy attribution while
appending. Once appended, the event contains only scalar snapshot values, so
later account mutation or failover cannot change an earlier event.
Queue sanitization keeps attempt events newest-first under hard bounds:
- the newest 16 attempts retain `detail` and `upstream_response_body` (each
capped at the queue body limit) plus `upstream_url`/`message` up to 2048
bytes;
- older attempts keep timestamps, account, proxy, status, kind, stage, scope,
reason and a `upstream_url`/`message` trimmed to 512 bytes; an attempt whose
only payload was a cleared `detail` is still retained;
- at most 256 attempts and roughly 512 KiB of serialized JSON are kept per
request. When either bound drops attempts, the oldest retained event carries
`dropped_earlier_attempts` with the number of discarded earlier attempts. The
newest attempt is always retained because it decides the request outcome.
## Legacy events and analytics
Legacy JSON is not rewritten in the database. Detail reads edit only the
`proxy_id`/`proxy_name` keys of events that lack valid attribution, so keys
written by older struct versions and the stored key order are preserved. The
shared parser applies the same rule to the decoded structs. Both materialize
missing attribution as:
```json
{
"proxy_id": null,
"proxy_name": "unknown"
}
```
Proxy and region aggregation must use the event fields first. A safe grouping
rule is:
1. A non-null event `proxy_id` groups by that ID and uses the event
`proxy_name` as the historical label. A join to `proxies` may enrich current
metadata but must not replace the event label.
2. `proxy_id=null` and `proxy_name=direct/no_proxy` groups as direct.
3. Missing attribution, or `proxy_id=null` and `proxy_name=unknown`, groups as
unknown. Do not join through the account's current `proxy_id` and present it
as historical attribution.
If an operator deliberately produces a current-account cohort for legacy data,
the report must be labeled as a current snapshot; it is not historical proxy
attribution.
backend/internal/service/antigravity_gateway_claude.go
+14
View File
@@ -146,6 +146,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
logBody, maxBytes := s.getLogConfig()
upstreamDetail := s.getUpstreamErrorDetail(respBody)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -203,6 +205,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
})
if retryErr != nil {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -242,6 +246,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
retryUpstreamDetail = truncateString(string(retryBody), maxBytes)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -278,6 +284,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
errMsg := strings.TrimSpace(extractAntigravityErrorMessage(respBody))
if isThinkingBudgetConstraintError(errMsg) && s.settingService.IsBudgetRectifierEnabled(ctx) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -359,6 +367,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
logger.LegacyPrintf("service.antigravity_gateway", "%s status=400 prompt_too_long=true upstream_message=%q request_id=%s body=%s", prefix, upstreamMsg, resp.Header.Get("x-request-id"), truncateForLog(respBody, maxBytes))
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -385,6 +395,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
upstreamDetail := s.getUpstreamErrorDetail(respBody)
log.Printf("%s status=400 google_config_error failover=true upstream_message=%q account=%d", prefix, upstreamMsg, account.ID)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -403,6 +415,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
upstreamDetail := s.getUpstreamErrorDetail(respBody)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/antigravity_gateway_compat.go
+4
View File
@@ -424,6 +424,8 @@ func (s *AntigravityGatewayService) handleAntigravityCompatHTTPError(
if s.shouldFailoverUpstreamError(resp.StatusCode) {
message := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractAntigravityErrorMessage(body)))
event := OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -493,6 +495,8 @@ func (s *AntigravityGatewayService) writeMappedAntigravityCompatError(
message := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractAntigravityErrorMessage(body)))
setOpsUpstreamError(c, upstreamStatus, message, s.getUpstreamErrorDetail(body))
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/antigravity_gateway_gemini.go
+14
View File
@@ -227,6 +227,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
upstreamMsg := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractAntigravityErrorMessage(signatureCheckBody)))
upstreamDetail := s.getUpstreamErrorDetail(signatureCheckBody)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -272,6 +274,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
retryOpsBody = retryUnwrapped
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -292,6 +296,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
} else {
if switchErr, ok := IsAntigravityAccountSwitchError(retryErr); ok {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -305,6 +311,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
}
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -346,6 +354,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
if resp.StatusCode == http.StatusBadRequest && isGoogleProjectConfigError(strings.ToLower(upstreamMsg)) {
log.Printf("%s status=400 google_config_error failover=true upstream_message=%q account=%d", prefix, upstreamMsg, account.ID)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -360,6 +370,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
if s.shouldFailoverUpstreamError(resp.StatusCode) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -375,6 +387,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
contentType = "application/json"
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/antigravity_gateway_retry.go
+6
View File
@@ -548,6 +548,8 @@ urlFallbackLoop:
if err != nil {
safeErr := sanitizeUpstreamErrorMessage(err.Error())
appendOpsUpstreamError(p.c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(p.account),
ProxyName: opsUpstreamProxyName(p.account),
Platform: p.account.Platform,
AccountID: p.account.ID,
AccountName: p.account.Name,
@@ -625,6 +627,8 @@ urlFallbackLoop:
upstreamMsg := strings.TrimSpace(extractAntigravityErrorMessage(respBody))
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
appendOpsUpstreamError(p.c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(p.account),
ProxyName: opsUpstreamProxyName(p.account),
Platform: p.account.Platform,
AccountID: p.account.ID,
AccountName: p.account.Name,
@@ -660,6 +664,8 @@ urlFallbackLoop:
upstreamMsg := strings.TrimSpace(extractAntigravityErrorMessage(respBody))
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
appendOpsUpstreamError(p.c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(p.account),
ProxyName: opsUpstreamProxyName(p.account),
Platform: p.account.Platform,
AccountID: p.account.ID,
AccountName: p.account.Name,
backend/internal/service/antigravity_gateway_streaming.go
+2
View File
@@ -695,6 +695,8 @@ func (s *AntigravityGatewayService) writeMappedClaudeError(c *gin.Context, accou
upstreamDetail := s.getUpstreamErrorDetail(body)
setOpsUpstreamError(c, upstreamStatus, upstreamMsg, upstreamDetail)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/gateway_anthropic_passthrough.go
+6
View File
@@ -140,6 +140,8 @@ func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput(
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -184,6 +186,8 @@ func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput(
s.handleRetryExhaustedSideEffects(ctx, resp, account)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -218,6 +222,8 @@ func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput(
s.handleFailoverSideEffects(ctx, resp, account, input.RequestModel)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/gateway_bedrock.go
+6
View File
@@ -225,6 +225,8 @@ func (s *GatewayService) executeBedrockUpstream(
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -276,6 +278,8 @@ func (s *GatewayService) handleBedrockUpstreamErrors(
s.handleRetryExhaustedSideEffects(ctx, resp, account)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -300,6 +304,8 @@ func (s *GatewayService) handleBedrockUpstreamErrors(
s.handleFailoverSideEffects(ctx, resp, account)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/gateway_count_tokens.go
+4
View File
@@ -270,6 +270,8 @@ func (s *GatewayService) forwardCountTokensAnthropicAPIKeyPassthrough(ctx contex
if err != nil {
setOpsUpstreamError(c, 0, sanitizeUpstreamErrorMessage(err.Error()), "")
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -324,6 +326,8 @@ func (s *GatewayService) forwardCountTokensAnthropicAPIKeyPassthrough(ctx contex
}
setOpsUpstreamError(c, resp.StatusCode, upstreamMsg, upstreamDetail)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/gateway_forward.go
+18
View File
@@ -405,6 +405,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
if s.shouldRectifySignatureError(ctx, account, respBody, reqModel) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -466,6 +468,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
_ = retryResp.Body.Close()
if retryReadErr == nil && retryResp.StatusCode == 400 && s.isSignatureErrorPattern(ctx, account, retryRespBody) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -506,6 +510,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
_ = retryResp2.Body.Close()
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -545,6 +551,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
errMsg := extractUpstreamErrorMessage(respBody)
if isThinkingBudgetConstraintError(errMsg) && s.settingService.IsBudgetRectifierEnabled(ctx) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -615,6 +623,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -669,6 +679,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
s.handleRetryExhaustedSideEffects(ctx, resp, account)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -704,6 +716,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
s.handleFailoverSideEffects(ctx, resp, account, reqModel)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
UpstreamStatusCode: resp.StatusCode,
@@ -746,6 +760,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
upstreamDetail = truncateString(string(respBody), maxBytes)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -822,6 +838,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/gateway_forward_as_chat_completions.go
+2
View File
@@ -148,6 +148,8 @@ func (s *GatewayService) ForwardAsChatCompletions(
if s.shouldFailoverUpstreamError(resp.StatusCode) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/gateway_forward_as_responses.go
+2
View File
@@ -161,6 +161,8 @@ func (s *GatewayService) ForwardAsResponses(
if s.shouldFailoverUpstreamError(resp.StatusCode) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/gateway_upstream_response.go
+4
View File
@@ -398,6 +398,8 @@ func (s *GatewayService) handleErrorResponse(ctx context.Context, resp *http.Res
}
setOpsUpstreamError(c, resp.StatusCode, upstreamMsg, upstreamDetail)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
UpstreamStatusCode: resp.StatusCode,
@@ -578,6 +580,8 @@ func (s *GatewayService) handleRetryExhaustedError(ctx context.Context, resp *ht
}
setOpsUpstreamError(c, resp.StatusCode, upstreamMsg, upstreamDetail)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
UpstreamStatusCode: resp.StatusCode,
backend/internal/service/gateway_upstream_transport_error.go
+3 -1
View File
@@ -27,7 +27,8 @@ var gatewayTransportFailoverBody = []byte(`{"type":"error","error":{"type":"upst
// proxy / DNS / TCP / TLS). It:
// 1. records the failure in Ops error logs (status 0, kind=request_error) —
// the caller passes path-specific fields (UpstreamURL, Passthrough) via
// event; identity and classification fields are filled here;
// event; identity, proxy attribution and classification fields are
// filled here from the same account snapshot that built the transport;
// 2. for durable faults (expired/rejected proxy creds, dead proxy,
// DNS/routing) temporarily unschedules the account and logs a stable warn
// event that alert rules can key on;
@@ -40,6 +41,7 @@ var gatewayTransportFailoverBody = []byte(`{"type":"error","error":{"type":"upst
func (s *GatewayService) handleUpstreamTransportError(ctx context.Context, c *gin.Context, account *Account, err error, event OpsUpstreamErrorEvent) error {
safeErr := sanitizeUpstreamErrorMessage(err.Error())
setOpsUpstreamError(c, 0, safeErr, "")
event.ProxyID, event.ProxyName = opsUpstreamProxyAttribution(account)
event.Platform = account.Platform
event.AccountID = account.ID
event.AccountName = account.Name
backend/internal/service/gateway_upstream_transport_error_test.go
+55
View File
@@ -72,6 +72,61 @@ func TestHandleUpstreamTransportError_TransientFailsOverWithoutEviction(t *testi
}
}
// TestHandleUpstreamTransportError_AttributesEventTimeProxy pins that the
// shared helper stamps proxy attribution from the same account snapshot the
// transport used, so every Anthropic/Bedrock caller inherits it.
func TestHandleUpstreamTransportError_AttributesEventTimeProxy(t *testing.T) {
proxyID := int64(10060)
tests := []struct {
name string
account *Account
wantID *int64
wantName string
}{
{
name: "managed proxy",
account: &Account{ID: 149, Name: "acc", Platform: PlatformAnthropic, ProxyID: &proxyID, Proxy: &Proxy{ID: proxyID, Name: "wldsg82-ipv6-10060"}},
wantID: &proxyID,
wantName: "wldsg82-ipv6-10060",
},
{
name: "direct",
account: &Account{ID: 149, Name: "acc", Platform: PlatformAnthropic},
wantName: opsProxyNameDirect,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
s := &GatewayService{accountRepo: &transportTempUnschedRepoStub{}}
c := newTransportErrorTestGin(t)
_ = s.handleUpstreamTransportError(context.Background(), c, tt.account,
errors.New("EOF"), OpsUpstreamErrorEvent{UpstreamURL: "https://api.anthropic.com/v1/messages", Passthrough: true})
raw, ok := c.Get(OpsUpstreamErrorsKey)
if !ok {
t.Fatal("no upstream events recorded")
}
events := raw.([]*OpsUpstreamErrorEvent)
if len(events) != 1 {
t.Fatalf("events = %d, want 1", len(events))
}
ev := events[0]
if tt.wantID == nil {
if ev.ProxyID != nil {
t.Fatalf("proxy_id = %d, want null", *ev.ProxyID)
}
} else if ev.ProxyID == nil || *ev.ProxyID != *tt.wantID {
t.Fatalf("proxy_id = %v, want %d", ev.ProxyID, *tt.wantID)
}
if ev.ProxyName != tt.wantName {
t.Fatalf("proxy_name = %q, want %q", ev.ProxyName, tt.wantName)
}
if !ev.Passthrough || ev.UpstreamURL == "" || ev.Kind != "request_error" {
t.Fatalf("caller-supplied fields lost: %+v", ev)
}
})
}
}
// TestHandleUpstreamTransportError_PersistentEvictsAccount pins the contract
// for durable faults (dead endpoint / DNS / proxy credentials): fail over AND
// temporarily unschedule the account for the transport cooldown.
backend/internal/service/gemini_chat_completions_compat_service.go
+8
View File
@@ -127,6 +127,8 @@ func (s *GeminiMessagesCompatService) forwardClaudeBodyAsChatCompletions(
if err != nil {
safeErr := sanitizeUpstreamErrorMessage(err.Error())
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -172,6 +174,8 @@ func (s *GeminiMessagesCompatService) forwardClaudeBodyAsChatCompletions(
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -225,6 +229,8 @@ func (s *GeminiMessagesCompatService) forwardClaudeBodyAsChatCompletions(
if s.shouldFailoverGeminiUpstreamError(resp.StatusCode) {
upstreamMsg := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(evBody)))
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -816,6 +822,8 @@ func (s *GeminiMessagesCompatService) writeGeminiChatCompletionsMappedError(
setOpsUpstreamError(c, upstreamStatus, upstreamMsg, "")
if account != nil {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/gemini_messages_compat_service.go
+30
View File
@@ -785,6 +785,8 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex
if err != nil {
safeErr := sanitizeUpstreamErrorMessage(err.Error())
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -823,6 +825,8 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex
upstreamDetail = truncateString(string(respBody), maxBytes)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -909,6 +913,8 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex
upstreamDetail = truncateString(string(respBody), maxBytes)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -976,6 +982,8 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex
upstreamDetail = truncateString(string(respBody), maxBytes)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1010,6 +1018,8 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex
}
log.Printf("[Gemini] status=400 google_config_error failover=true upstream_message=%q account=%d", upstreamMsg, account.ID)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1038,6 +1048,8 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex
upstreamDetail = truncateString(string(respBody), maxBytes)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1324,6 +1336,8 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin.
if err != nil {
safeErr := sanitizeUpstreamErrorMessage(err.Error())
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1392,6 +1406,8 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin.
upstreamDetail = truncateString(string(respBody), maxBytes)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1492,6 +1508,8 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin.
upstreamDetail = truncateString(string(evBody), maxBytes)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1523,6 +1541,8 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin.
}
log.Printf("[Gemini] status=400 google_config_error failover=true upstream_message=%q account=%d", upstreamMsg, account.ID)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1548,6 +1568,8 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin.
upstreamDetail = truncateString(string(evBody), maxBytes)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1678,6 +1700,8 @@ func (s *GeminiMessagesCompatService) skippedErrorPolicyFailoverError(c *gin.Con
upstreamMsg := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(respBody)))
upstreamDetail := s.upstreamErrorDetail(respBody)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1719,6 +1743,8 @@ func (s *GeminiMessagesCompatService) writeGeminiCustomCodeSkippedError(c *gin.C
upstreamDetail := s.upstreamErrorDetail(body)
setOpsUpstreamError(c, upstreamStatus, upstreamMsg, upstreamDetail)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1747,6 +1773,8 @@ func (s *GeminiMessagesCompatService) writeGeminiNativeUpstreamError(c *gin.Cont
}
setOpsUpstreamError(c, resp.StatusCode, upstreamMsg, upstreamDetail)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1811,6 +1839,8 @@ func (s *GeminiMessagesCompatService) writeGeminiMappedError(c *gin.Context, acc
}
setOpsUpstreamError(c, upstreamStatus, upstreamMsg, upstreamDetail)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/grok_credential_failure.go
+4
View File
@@ -647,6 +647,10 @@ func (s *OpenAIGatewayService) newGrokCredentialFailover(c *gin.Context, account
class.message = "Grok OAuth credentials are unavailable"
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
// Credential acquisition happens before the inference transport opens,
// so the account binding is not evidence of an inference proxy route.
ProxyID: nil,
ProxyName: opsProxyNameUnknown,
Platform: PlatformGrok,
AccountID: account.ID,
Stage: string(GatewayFailureStageAccountAuth),
backend/internal/service/grok_credential_failure_test.go
+28
View File
@@ -294,6 +294,34 @@ func TestGetRequestCredentialMapsPermanentGrokOAuthFailureAndRedactsSecrets(t *t
require.NotContains(t, events[0].Message, "leaked-refresh")
}
func TestNewGrokCredentialFailoverDoesNotAttributeInferenceProxy(t *testing.T) {
gin.SetMode(gin.TestMode)
proxyID := int64(43)
account := &Account{
ID: 701,
Platform: PlatformGrok,
ProxyID: &proxyID,
Proxy: &Proxy{ID: proxyID, Name: "bound-proxy"},
}
c, _ := gin.CreateTestContext(httptest.NewRecorder())
err := (&OpenAIGatewayService{}).newGrokCredentialFailover(c, account, grokCredentialFailureClass{
scope: GatewayFailureScopeAccount,
reason: GrokCredentialReasonAccountChanged,
action: NextAccountRetry,
message: "credential unavailable",
})
require.Error(t, err)
rawEvents, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events, ok := rawEvents.([]*OpsUpstreamErrorEvent)
require.True(t, ok)
require.Len(t, events, 1)
require.Nil(t, events[0].ProxyID)
require.Equal(t, opsProxyNameUnknown, events[0].ProxyName)
}
func TestGetRequestCredentialPermanentMappingsPersistAndInvalidate(t *testing.T) {
gin.SetMode(gin.TestMode)
tests := []struct {
backend/internal/service/grok_media.go
+6
View File
@@ -1239,6 +1239,8 @@ func (s *OpenAIGatewayService) handleGrokMediaErrorResponse(
if isGrokContentPolicyRejection(resp.StatusCode, body) {
clientMsg := grokContentPolicyClientMessage(body)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1269,6 +1271,8 @@ func (s *OpenAIGatewayService) handleGrokMediaErrorResponse(
if !account.ShouldHandleErrorCode(resp.StatusCode) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1288,6 +1292,8 @@ func (s *OpenAIGatewayService) handleGrokMediaErrorResponse(
kind = "failover"
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/openai_compact_fallback.go
+2
View File
@@ -215,6 +215,8 @@ func (s *OpenAIGatewayService) appendOpenAICompactFallbackRetryOps(
detail = truncateString(string(payload), maxBytes)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/openai_compact_fallback_test.go
+121 -5
View File
@@ -225,6 +225,120 @@ func TestOpenAIGatewayForwardRetriesExplicitNativeCompactHTTPFailureOnce(t *test
require.Equal(t, "retry", events[0].Kind)
require.Equal(t, "compact_model_fallback", events[0].Reason)
require.Equal(t, http.StatusBadRequest, events[0].UpstreamStatusCode)
require.Nil(t, events[0].ProxyID)
require.Equal(t, opsProxyNameDirect, events[0].ProxyName)
}
func compactFallbackManagedProxyAccount() (*Account, *Proxy) {
proxy := &Proxy{ID: 10060, Name: "wldsg82-ipv6-10060", Protocol: "http", Host: "proxy.example", Port: 8080}
return &Account{
ID: 1, Name: "openai-oauth", Platform: PlatformOpenAI, Type: AccountTypeOAuth, Concurrency: 1,
Credentials: map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-account"},
Status: StatusActive, Schedulable: true,
ProxyID: &proxy.ID, Proxy: proxy,
}, proxy
}
func requireCompactEventsAttributedTo(t *testing.T, events []*OpsUpstreamErrorEvent, proxy *Proxy) {
t.Helper()
for i, ev := range events {
require.NotNil(t, ev.ProxyID, "event %d proxy_id", i)
require.Equal(t, proxy.ID, *ev.ProxyID, "event %d proxy_id", i)
require.Equal(t, proxy.Name, ev.ProxyName, "event %d proxy_name", i)
}
}
// Non-streaming SSE compact retry: the first attempt's failure must be recorded
// as its own attempt (it previously vanished on `continue`), carrying the
// managed proxy the transport actually used.
func TestOpenAIGatewayForwardNonStreamCompactRetryRecordsAttemptWithManagedProxy(t *testing.T) {
gin.SetMode(gin.TestMode)
body := []byte(`{"model":"gpt-5.5","stream":false,"instructions":"compact-test","input":[{"type":"message","role":"user","content":"hello"},{"type":"compaction_trigger"}]}`)
c := newOpenAICompactFallbackTestContext(t, "/v1/responses")
c.Request.Body = io.NopCloser(bytes.NewReader(body))
c.Request.Header.Set("Content-Type", "application/json")
MarkOpenAINativeCompactionV2(c)
upstream := &httpUpstreamRecorder{responses: []*http.Response{
{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"text/event-stream"}, "X-Request-Id": []string{"rid_compact_1"}},
Body: io.NopCloser(strings.NewReader("event: response.failed\n" +
`data: {"type":"response.failed","response":{"status":"failed","error":{"code":"context_length_exceeded","message":"context window exceeded"}}}` + "\n\n")),
},
{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: io.NopCloser(strings.NewReader(`{"id":"resp_compact","status":"completed","model":"gpt-5.4","output":[],"usage":{"input_tokens":1,"output_tokens":1}}`)),
},
}}
svc := &OpenAIGatewayService{
cfg: &config.Config{Gateway: config.GatewayConfig{OpenAICompactModel: "gpt-5.4"}},
httpUpstream: upstream,
}
account, proxy := compactFallbackManagedProxyAccount()
result, err := svc.Forward(context.Background(), c, account, body)
require.NoError(t, err)
require.NotNil(t, result)
require.Len(t, upstream.bodies, 2)
require.Equal(t, proxy.URL(), upstream.lastProxyURL)
rawEvents, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events, ok := rawEvents.([]*OpsUpstreamErrorEvent)
require.True(t, ok)
require.Len(t, events, 1)
require.Equal(t, "retry", events[0].Kind)
require.Equal(t, "compact_model_fallback", events[0].Reason)
require.Equal(t, "rid_compact_1", events[0].UpstreamRequestID)
requireCompactEventsAttributedTo(t, events, proxy)
}
// Streaming compact fallback whose second attempt fails with a failover-class
// error: both the retry event and the failover event must carry the proxy.
func TestOpenAIGatewayForwardCompactFailoverEventCarriesManagedProxy(t *testing.T) {
gin.SetMode(gin.TestMode)
body := []byte(`{"model":"gpt-5.5","stream":true,"instructions":"compact-test","input":[{"type":"message","role":"user","content":"hello"},{"type":"compaction_trigger"}]}`)
recorder := httptest.NewRecorder()
c, _ := gin.CreateTestContext(recorder)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil)
c.Request.Body = io.NopCloser(bytes.NewReader(body))
c.Request.Header.Set("Content-Type", "application/json")
MarkOpenAINativeCompactionV2(c)
contextFailed := "event: response.failed\n" +
`data: {"type":"response.failed","response":{"status":"failed","error":{"code":"context_length_exceeded","message":"context window exceeded"}}}` + "\n\n"
// The fallback model's failure is an account-state error, which
// shouldFailoverOpenAIUpstreamResponse classifies as failover-worthy.
workspaceFailed := "event: response.failed\n" +
`data: {"type":"response.failed","response":{"status":"failed","error":{"code":"deactivated_workspace","message":"workspace deactivated"}}}` + "\n\n"
upstream := &httpUpstreamRecorder{responses: []*http.Response{
{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(contextFailed))},
{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(workspaceFailed))},
}}
svc := &OpenAIGatewayService{
cfg: &config.Config{Gateway: config.GatewayConfig{OpenAICompactModel: "gpt-5.4"}},
httpUpstream: upstream,
}
account, proxy := compactFallbackManagedProxyAccount()
result, err := svc.Forward(context.Background(), c, account, body)
require.Error(t, err)
require.Nil(t, result)
var failoverErr *UpstreamFailoverError
require.ErrorAs(t, err, &failoverErr)
require.Len(t, upstream.bodies, 2)
rawEvents, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events, ok := rawEvents.([]*OpsUpstreamErrorEvent)
require.True(t, ok)
require.Len(t, events, 2)
require.Equal(t, "retry", events[0].Kind)
require.Equal(t, "compact_model_fallback", events[0].Reason)
require.Equal(t, "failover", events[1].Kind)
requireCompactEventsAttributedTo(t, events, proxy)
}
func TestOpenAIGatewayForwardRetriesExplicitNativeCompactSSEFailureBeforeOutput(t *testing.T) {
@@ -355,6 +469,10 @@ func TestOpenAIGatewayForwardDoesNotRecurseWhenCompactFallbackAlsoFails(t *testi
require.Equal(t, "retry", events[0].Kind)
require.Equal(t, "compact_model_fallback", events[0].Reason)
require.Equal(t, "http_error", events[1].Kind)
for _, ev := range events {
require.Nil(t, ev.ProxyID)
require.Equal(t, opsProxyNameDirect, ev.ProxyName)
}
}
func TestOpenAIPassthroughCompactFallbackSecondStreamFailureUsesStandardErrorPath(t *testing.T) {
@@ -376,11 +494,7 @@ func TestOpenAIPassthroughCompactFallbackSecondStreamFailureUsesStandardErrorPat
cfg: &config.Config{Gateway: config.GatewayConfig{OpenAICompactModel: "gpt-5.4"}},
httpUpstream: upstream,
}
account := &Account{
ID: 1, Name: "openai-oauth", Platform: PlatformOpenAI, Type: AccountTypeOAuth, Concurrency: 1,
Credentials: map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-account"},
Status: StatusActive, Schedulable: true,
}
account, proxy := compactFallbackManagedProxyAccount()
result, err := svc.forwardOpenAIPassthrough(
context.Background(), c, account, body, body, "gpt-5.5", false, nil, true, time.Now(),
@@ -402,4 +516,6 @@ func TestOpenAIPassthroughCompactFallbackSecondStreamFailureUsesStandardErrorPat
require.Equal(t, "compact_model_fallback", events[0].Reason)
require.Equal(t, "http_error", events[1].Kind)
require.True(t, events[1].Passthrough)
require.Equal(t, proxy.URL(), upstream.lastProxyURL)
requireCompactEventsAttributedTo(t, events, proxy)
}
backend/internal/service/openai_embeddings.go
+5 -1
View File
@@ -89,7 +89,7 @@ func (s *OpenAIGatewayService) ForwardEmbeddings(
account.ApplyHeaderOverrides(upstreamReq.Header)
proxyURL := ""
if account.Proxy != nil {
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
resp, err := s.doOpenAIUpstream(upstreamReq, proxyURL, account)
@@ -97,6 +97,8 @@ func (s *OpenAIGatewayService) ForwardEmbeddings(
safeErr := sanitizeUpstreamErrorMessage(err.Error())
setOpsUpstreamError(c, 0, safeErr, "")
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -126,6 +128,8 @@ func (s *OpenAIGatewayService) ForwardEmbeddings(
upstreamDetail = truncateString(string(respBody), maxBytes)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/openai_first_output_timeout.go
+9 -1
View File
@@ -243,10 +243,16 @@ func (s *OpenAIGatewayService) openAIFirstOutputTimeout(reasoningEffort string)
return time.Duration(seconds) * time.Second
}
// newOpenAIFirstOutputTimeoutError records the timeout as an upstream attempt
// and returns the failover error. proxyID/proxyName are supplied by the caller
// because the same deadline is enforced over HTTP and WebSocket transports,
// whose direct-route semantics differ (see opsUpstreamWSProxyAttribution).
func (s *OpenAIGatewayService) newOpenAIFirstOutputTimeoutError(
ctx context.Context,
c *gin.Context,
account *Account,
proxyID *int64,
proxyName string,
startTime time.Time,
originalModel string,
reasoningEffort string,
@@ -262,7 +268,9 @@ func (s *OpenAIGatewayService) newOpenAIFirstOutputTimeoutError(
)
requestID := strings.TrimSpace(responseHeaders.Get("x-request-id"))
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
Platform: account.Platform, AccountID: account.ID, AccountName: account.Name,
ProxyID: proxyID,
ProxyName: proxyName,
Platform: account.Platform, AccountID: account.ID, AccountName: account.Name,
UpstreamStatusCode: http.StatusGatewayTimeout, UpstreamRequestID: requestID,
Kind: "first_output_timeout", Message: "OpenAI upstream produced no semantic output before the deadline",
Detail: fmt.Sprintf("phase=%s elapsed_ms=%d timeout_ms=%d", phase, elapsed.Milliseconds(), timeout.Milliseconds()),
backend/internal/service/openai_first_output_timeout_test.go
+27
View File
@@ -152,6 +152,33 @@ func TestOpenAINativeFirstOutputTimeoutIgnoresPreambleAndCleansReader(t *testing
}
}
func TestNewOpenAIFirstOutputTimeoutErrorRecordsCallerProxyAttribution(t *testing.T) {
gin.SetMode(gin.TestMode)
c, _ := gin.CreateTestContext(httptest.NewRecorder())
svc := &OpenAIGatewayService{}
proxyID := int64(10060)
account := &Account{ID: 1, Name: "acc", Platform: PlatformOpenAI, ProxyID: &proxyID, Proxy: &Proxy{ID: proxyID, Name: "ws-proxy"}}
// WebSocket callers pass the WS attribution; HTTP callers pass the HTTP one.
wsID, wsName := opsUpstreamWSProxyAttribution(&Account{ID: 2, Platform: PlatformOpenAI})
_ = svc.newOpenAIFirstOutputTimeoutError(context.Background(), c, &Account{ID: 2, Name: "ws", Platform: PlatformOpenAI},
wsID, wsName, time.Now(), "gpt-5.5", "", time.Second, "websocket_first_semantic_output", nil)
_ = svc.newOpenAIFirstOutputTimeoutError(context.Background(), c, account,
opsUpstreamProxyID(account), opsUpstreamProxyName(account), time.Now(), "gpt-5.5", "", time.Second, "semantic_output", nil)
raw, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events, ok := raw.([]*OpsUpstreamErrorEvent)
require.True(t, ok)
require.Len(t, events, 2)
require.Equal(t, "first_output_timeout", events[0].Kind)
require.Nil(t, events[0].ProxyID)
require.Equal(t, opsProxyNameUnknown, events[0].ProxyName)
require.NotNil(t, events[1].ProxyID)
require.Equal(t, proxyID, *events[1].ProxyID)
require.Equal(t, "ws-proxy", events[1].ProxyName)
}
func TestOpenAIFirstOutputTimeoutForReasoningEffort(t *testing.T) {
svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{
OpenAIFirstOutputTimeoutSeconds: 120,
backend/internal/service/openai_gateway_cc_pipeline.go
+3 -1
View File
@@ -112,6 +112,8 @@ func (s *OpenAIGatewayService) failoverOpenAIUpstreamHTTPError(
upstreamDetail = truncateString(string(respBody), maxBytes)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -223,7 +225,7 @@ func (s *OpenAIGatewayService) sendCCUpstreamRequest(
account.ApplyHeaderOverrides(upstreamReq.Header)
proxyURL := ""
if account.Proxy != nil {
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
resp, err := s.doOpenAIUpstream(upstreamReq, proxyURL, account)
backend/internal/service/openai_gateway_chat_completions.go
+1 -1
View File
@@ -348,7 +348,7 @@ func (s *OpenAIGatewayService) forwardAsChatCompletions(
// 7. Send request
proxyURL := ""
if account.Proxy != nil {
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
resp, err := s.doOpenAIUpstream(upstreamReq, proxyURL, account)
backend/internal/service/openai_gateway_chat_completions_anthropic_native.go
+1 -1
View File
@@ -106,7 +106,7 @@ func (s *OpenAIGatewayService) forwardChatCompletionsViaNativeAnthropic(
}
proxyURL := ""
if account.Proxy != nil {
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
backend/internal/service/openai_gateway_chat_completions_raw.go
+2
View File
@@ -193,6 +193,8 @@ func (s *OpenAIGatewayService) forwardAsRawChatCompletions(
kind = "failover"
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/openai_gateway_chat_completions_raw_test.go
+44
View File
@@ -727,6 +727,50 @@ func TestForwardAsRawChatCompletions_TruncatedStreamAfterOutputFailsRequest(t *t
// 已写出的内容保持原样透传，客户端拿到的仍是它已经收到的那部分。
require.Contains(t, rec.Body.String(), `"content":"half an ans"`)
require.NotContains(t, rec.Body.String(), "data: [DONE]")
rawEvents, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events := rawEvents.([]*OpsUpstreamErrorEvent)
require.Len(t, events, 1)
require.Equal(t, "http_error", events[0].Kind)
require.Nil(t, events[0].ProxyID)
require.Equal(t, opsProxyNameDirect, events[0].ProxyName)
}
// 截断 failover 事件必须带上本次传输真实使用的托管代理快照。
func TestForwardAsRawChatCompletions_TruncationFailoverAttributesManagedProxy(t *testing.T) {
gin.SetMode(gin.TestMode)
body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
c.Request.Header.Set("Content-Type", "application/json")
upstream := &httpUpstreamRecorder{resp: &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_empty_proxy"}},
Body: io.NopCloser(strings.NewReader("")),
}}
svc := &OpenAIGatewayService{cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream}
proxy := &Proxy{ID: 8001, Name: "oxylabs-uk-8001", Protocol: "http", Host: "proxy.example", Port: 8080}
account := rawChatCompletionsTestAccount()
account.ProxyID = &proxy.ID
account.Proxy = proxy
_, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "")
var failoverErr *UpstreamFailoverError
require.ErrorAs(t, err, &failoverErr)
require.Equal(t, proxy.URL(), upstream.lastProxyURL)
rawEvents, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events := rawEvents.([]*OpsUpstreamErrorEvent)
require.Len(t, events, 1)
require.Equal(t, "failover", events[0].Kind)
require.NotNil(t, events[0].ProxyID)
require.Equal(t, proxy.ID, *events[0].ProxyID)
require.Equal(t, proxy.Name, events[0].ProxyName)
}
// 上游 200 但一个 SSE 字节都没发：响应头尚未提交，应换号重试而不是回 200 空流。
backend/internal/service/openai_gateway_count_tokens.go
+2 -2
View File
@@ -80,7 +80,7 @@ func (s *OpenAIGatewayService) ForwardResponsesInputTokens(
}
proxyURL := ""
if account.Proxy != nil {
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
resp, err := s.doOpenAIUpstream(upstreamReq, proxyURL, account)
@@ -320,7 +320,7 @@ func (s *OpenAIGatewayService) ForwardCountTokensAsAnthropic(
}
proxyURL := ""
if account.Proxy != nil {
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
resp, err := s.doOpenAIUpstream(upstreamReq, proxyURL, account)
backend/internal/service/openai_gateway_forward.go
+7 -1
View File
@@ -1015,7 +1015,8 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
}
headerGuard.close()
return nil, s.newOpenAIFirstOutputTimeoutError(
ctx, c, account, startTime, originalModel, reasoningEffortValue,
ctx, c, account, opsUpstreamProxyID(account), opsUpstreamProxyName(account),
startTime, originalModel, reasoningEffortValue,
firstOutputTimeout, "response_headers", nil,
)
}
@@ -1108,6 +1109,8 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
upstreamDetail = truncateString(string(respBody), maxBytes)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1173,6 +1176,8 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
compactResp, compactBody := openAICompactFallbackErrorResponse(resp, signal)
if s.shouldFailoverOpenAIUpstreamResponse(compactResp.StatusCode, signal.message, compactBody) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1204,6 +1209,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
if retryBody, fallbackModel, retry := s.prepareOpenAICompactFallbackRetry(
c, account, requestedModel, body, http.StatusBadRequest, signal.message, signal.payload, compactModelFallbackRetried,
); retry {
s.appendOpenAICompactFallbackRetryOps(c, account, resp, signal.payload, signal.message, false)
body = retryBody
requestView = newOpenAIRequestView(body)
upstreamModel = fallbackModel
backend/internal/service/openai_gateway_grok.go
+4
View File
@@ -170,6 +170,8 @@ func (s *OpenAIGatewayService) forwardGrokResponses(
kind = "failover"
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1412,6 +1414,8 @@ func (s *OpenAIGatewayService) describeGrokComposerImage(
kind = "failover"
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/openai_gateway_grok_chat_bridge.go
+2
View File
@@ -623,6 +623,8 @@ func (s *OpenAIGatewayService) forwardGrokChatCompletionsViaResponses(
kind = "failover"
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/openai_gateway_messages.go
+3 -1
View File
@@ -374,7 +374,7 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic(
// 7. Send request
proxyURL := ""
if account.Proxy != nil {
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
// Grok may reject encrypted reasoning replayed under a different OAuth
@@ -689,6 +689,8 @@ func (s *OpenAIGatewayService) recordOpenAIMessagesStreamUpstreamError(c *gin.Co
message = sanitizeUpstreamErrorMessage(message)
setOpsUpstreamError(c, http.StatusBadGateway, message, "")
event := OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: PlatformOpenAI,
UpstreamStatusCode: http.StatusBadGateway,
UpstreamRequestID: strings.TrimSpace(upstreamRequestID),
backend/internal/service/openai_gateway_messages_anthropic_native.go
+1 -1
View File
@@ -86,7 +86,7 @@ func (s *OpenAIGatewayService) forwardAnthropicViaNativeAnthropicEndpoint(
}
proxyURL := ""
if account.Proxy != nil {
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
backend/internal/service/openai_gateway_passthrough.go
+6
View File
@@ -894,6 +894,8 @@ func (s *OpenAIGatewayService) handleFailoverErrorResponsePassthrough(
canonicalModel := canonicalOpenAIAccountSchedulingModel(account, reqModel)
shouldDisable := s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, body, canonicalModel)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -960,6 +962,8 @@ func (s *OpenAIGatewayService) handleErrorResponsePassthrough(
_ = s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, body, canonicalModel)
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1676,6 +1680,8 @@ func (s *OpenAIGatewayService) recordOpenAIStreamUpstreamError(
if c != nil {
setOpsUpstreamError(c, statusCode, message, detail)
event := OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: PlatformOpenAI,
UpstreamStatusCode: statusCode,
UpstreamRequestID: strings.TrimSpace(upstreamRequestID),
backend/internal/service/openai_gateway_response_handling.go
+2 -1
View File
@@ -925,7 +925,8 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context.
markEventProcessed(ev)
}
return resultWithUsage(), s.newOpenAIFirstOutputTimeoutError(
ctx, c, account, startTime, originalModel, reasoningEffort,
ctx, c, account, opsUpstreamProxyID(account), opsUpstreamProxyName(account),
startTime, originalModel, reasoningEffort,
firstOutputTimeout, "semantic_output", resp.Header,
)
backend/internal/service/openai_gateway_responses_anthropic_native.go
+1 -1
View File
@@ -111,7 +111,7 @@ func (s *OpenAIGatewayService) forwardResponsesViaNativeAnthropic(
}
proxyURL := ""
if account.Proxy != nil {
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
backend/internal/service/openai_gateway_service.go
+3
View File
@@ -892,7 +892,10 @@ func (s *OpenAIGatewayService) writeOpenAIWSFallbackErrorResponse(c *gin.Context
setOpsUpstreamError(c, statusCode, upstreamMessage, "")
if account != nil {
proxyID, proxyName := opsUpstreamWSProxyAttribution(account)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: proxyID,
ProxyName: proxyName,
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/openai_gateway_upstream_errors.go
+10
View File
@@ -549,6 +549,8 @@ func (s *OpenAIGatewayService) handleErrorResponse(
if isOpenAIRequestBodyTooLargeError(resp.StatusCode, upstreamMsg, body) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -596,6 +598,8 @@ func (s *OpenAIGatewayService) handleErrorResponse(
// Check custom error codes
if !account.ShouldHandleErrorCode(resp.StatusCode) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -633,6 +637,8 @@ func (s *OpenAIGatewayService) handleErrorResponse(
kind = "failover"
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -798,6 +804,8 @@ func (s *OpenAIGatewayService) handleCompatErrorResponse(
// return a generic error without exposing upstream details.
if !account.ShouldHandleErrorCode(resp.StatusCode) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -828,6 +836,8 @@ func (s *OpenAIGatewayService) handleCompatErrorResponse(
kind = "failover"
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/openai_gateway_usage_integrity.go
+2
View File
@@ -53,6 +53,8 @@ func newGrokMissingUsageFailoverError(c *gin.Context, account *Account, upstream
setOpsUpstreamError(c, http.StatusBadGateway, grokMissingUsageMessage, "")
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: PlatformGrok,
AccountID: accountID,
AccountName: accountName,
backend/internal/service/openai_images.go
+4
View File
@@ -631,6 +631,8 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesAPIKey(
safeErr := sanitizeUpstreamErrorMessage(err.Error())
setOpsUpstreamError(c, 0, safeErr, "")
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -650,6 +652,8 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesAPIKey(
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/openai_images_responses.go
+13 -1
View File
@@ -957,6 +957,8 @@ func (s *OpenAIGatewayService) handleOpenAIImagesErrorResponse(
// handleCompatErrorResponse).
if !account.ShouldHandleErrorCode(resp.StatusCode) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -997,6 +999,8 @@ func (s *OpenAIGatewayService) handleOpenAIImagesErrorResponse(
kind = "failover"
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1826,6 +1830,8 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesOAuth(
safeErr := sanitizeUpstreamErrorMessage(err.Error())
setOpsUpstreamError(c, 0, safeErr, "")
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -1852,6 +1858,8 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesOAuth(
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -2024,7 +2032,9 @@ func (s *OpenAIGatewayService) handleOpenAIImagesOAuthResponseError(
kind = "retry_exhausted_failover"
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
Platform: account.Platform, AccountID: account.ID, AccountName: account.Name,
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform, AccountID: account.ID, AccountName: account.Name,
UpstreamStatusCode: statusCode, UpstreamRequestID: requestID, UpstreamURL: upstreamURL,
Kind: kind, Message: message,
})
@@ -2066,6 +2076,8 @@ func (s *OpenAIGatewayService) handleOpenAIImagesOAuthResponseError(
}
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/openai_raw_stream_truncation.go
+2
View File
@@ -123,6 +123,8 @@ func recordOpenAIRawStreamTruncation(
setOpsUpstreamError(c, http.StatusBadGateway, message, "")
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: platform,
AccountID: accountID,
AccountName: accountName,
backend/internal/service/openai_silent_refusal.go
+4
View File
@@ -247,6 +247,8 @@ func newOpenAISilentRefusalFailoverError(c *gin.Context, account *Account, upstr
setOpsUpstreamError(c, http.StatusBadGateway, openAISilentRefusalUpstreamMessage, "")
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: platform,
AccountID: accountID,
AccountName: accountName,
@@ -284,6 +286,8 @@ func newOpenAIResponsesEmptyCompletedFailoverError(c *gin.Context, account *Acco
setOpsUpstreamError(c, http.StatusBadGateway, openAIResponsesEmptyCompletedMessage, "")
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: platform,
AccountID: accountID,
AccountName: accountName,
backend/internal/service/openai_upstream_transport_error.go
+2
View File
@@ -109,6 +109,8 @@ func (s *OpenAIGatewayService) handleOpenAIUpstreamTransportError(ctx context.Co
safeErr := sanitizeUpstreamErrorMessage(err.Error())
setOpsUpstreamError(c, 0, safeErr, "")
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
backend/internal/service/openai_upstream_transport_error_handle_test.go
+88 -5
View File
@@ -40,17 +40,23 @@ func newOpenAITransportErrTestContext() (*gin.Context, *httptest.ResponseRecorde
}
type failingOpenAIHTTPUpstream struct {
err error
calls int
err error
calls int
proxyURL string
proxyURLs []string
}
func (u *failingOpenAIHTTPUpstream) Do(_ *http.Request, _ string, _ int64, _ int) (*http.Response, error) {
func (u *failingOpenAIHTTPUpstream) Do(_ *http.Request, proxyURL string, _ int64, _ int) (*http.Response, error) {
u.calls++
u.proxyURL = proxyURL
u.proxyURLs = append(u.proxyURLs, proxyURL)
return nil, u.err
}
func (u *failingOpenAIHTTPUpstream) DoWithTLS(_ *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) {
func (u *failingOpenAIHTTPUpstream) DoWithTLS(_ *http.Request, proxyURL string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) {
u.calls++
u.proxyURL = proxyURL
u.proxyURLs = append(u.proxyURLs, proxyURL)
return nil, u.err
}
@@ -60,7 +66,14 @@ func (u *failingOpenAIHTTPUpstream) DoWithTLS(_ *http.Request, _ string, _ int64
func TestHandleOpenAIUpstreamTransportError_PersistentEvictsAndFailsOver(t *testing.T) {
repo := &openaiTransportAccountRepoStub{}
svc := &OpenAIGatewayService{accountRepo: repo}
account := &Account{ID: 4627, Name: "proxy-expired", Platform: PlatformOpenAI}
proxyID := int64(10060)
account := &Account{
ID: 4627,
Name: "proxy-expired",
Platform: PlatformOpenAI,
ProxyID: &proxyID,
Proxy: &Proxy{ID: proxyID, Name: "wldsg82-ipv6-10060"},
}
c, rec := newOpenAITransportErrTestContext()
before := time.Now()
@@ -85,6 +98,15 @@ func TestHandleOpenAIUpstreamTransportError_PersistentEvictsAndFailsOver(t *test
// Must NOT write a response body — the handler owns the (failover) response.
require.Equal(t, 0, rec.Body.Len())
rawEvents, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events, ok := rawEvents.([]*OpsUpstreamErrorEvent)
require.True(t, ok)
require.Len(t, events, 1)
require.NotNil(t, events[0].ProxyID)
require.Equal(t, proxyID, *events[0].ProxyID)
require.Equal(t, "wldsg82-ipv6-10060", events[0].ProxyName)
}
// A transient blip should fail over but must NOT evict the account.
@@ -191,12 +213,22 @@ func TestForwardAsRawChatCompletions_TransportErrorFailsOver(t *testing.T) {
},
},
}
proxyID := int64(8001)
proxy := &Proxy{
ID: proxyID,
Name: "oxylabs-uk-8001",
Protocol: "http",
Host: "proxy.example",
Port: 8080,
}
account := &Account{
ID: 81,
Name: "oc-20053",
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "sk-test", "base_url": "https://opencode.ai/zen/v1"},
ProxyID: &proxyID,
Proxy: proxy,
}
c, rec := newOpenAITransportErrTestContext()
body := []byte(`{"model":"deepseek-v4-flash-free","messages":[{"role":"user","content":"hello"}]}`)
@@ -204,11 +236,62 @@ func TestForwardAsRawChatCompletions_TransportErrorFailsOver(t *testing.T) {
_, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "")
require.Equal(t, 1, upstream.calls)
require.Equal(t, proxy.URL(), upstream.proxyURL)
var fo *UpstreamFailoverError
require.True(t, errors.As(err, &fo), "transport error must trigger account failover")
require.Equal(t, http.StatusBadGateway, fo.StatusCode)
require.Empty(t, repo.tempUnschedCalls, "plain EOF is transient: fail over but do not evict")
require.Equal(t, 0, rec.Body.Len(), "service must not write a hard 502 before handler can fail over")
rawEvents, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events, ok := rawEvents.([]*OpsUpstreamErrorEvent)
require.True(t, ok)
require.Len(t, events, 1)
require.NotNil(t, events[0].ProxyID)
require.Equal(t, proxyID, *events[0].ProxyID)
require.Equal(t, proxy.Name, events[0].ProxyName)
}
func TestForwardAsRawChatCompletions_RecordsProxyPerAccountAttempt(t *testing.T) {
upstream := &failingOpenAIHTTPUpstream{err: errors.New("EOF")}
svc := &OpenAIGatewayService{
accountRepo: &openaiTransportAccountRepoStub{},
httpUpstream: upstream,
cfg: &config.Config{Security: config.SecurityConfig{
URLAllowlist: config.URLAllowlistConfig{Enabled: false},
}},
}
proxyA := &Proxy{ID: 10060, Name: "proxy-a", Protocol: "http", Host: "proxy-a.example", Port: 8080}
proxyB := &Proxy{ID: 8001, Name: "proxy-b", Protocol: "http", Host: "proxy-b.example", Port: 8080}
fallbackOriginID := int64(10150)
accounts := []*Account{
{ID: 81, Name: "account-a", Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "a", "base_url": "https://example.com/v1"}, ProxyID: &proxyA.ID, Proxy: proxyA},
{ID: 82, Name: "account-b", Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "b", "base_url": "https://example.com/v1"}, ProxyID: &proxyB.ID, Proxy: proxyB, ProxyFallbackOriginID: &fallbackOriginID},
{ID: 83, Name: "account-direct", Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "c", "base_url": "https://example.com/v1"}},
}
c, _ := newOpenAITransportErrTestContext()
body := []byte(`{"model":"test","messages":[{"role":"user","content":"hello"}]}`)
for _, account := range accounts {
_, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "")
var failoverErr *UpstreamFailoverError
require.ErrorAs(t, err, &failoverErr)
}
require.Equal(t, []string{proxyA.URL(), proxyB.URL(), ""}, upstream.proxyURLs)
rawEvents, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events, ok := rawEvents.([]*OpsUpstreamErrorEvent)
require.True(t, ok)
require.Len(t, events, 3)
for i, proxy := range []*Proxy{proxyA, proxyB} {
require.NotNil(t, events[i].ProxyID)
require.Equal(t, proxy.ID, *events[i].ProxyID)
require.Equal(t, proxy.Name, events[i].ProxyName)
}
require.Nil(t, events[2].ProxyID)
require.Equal(t, opsProxyNameDirect, events[2].ProxyName)
}
func TestHandleOpenAIUpstreamTransportError_RecordsOllamaActivityOnly(t *testing.T) {
backend/internal/service/openai_ws_fallback_test.go
+59
View File
@@ -4,11 +4,13 @@ import (
"context"
"errors"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/Wei-Shaw/sub2api/internal/config"
coderws "github.com/coder/websocket"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
)
@@ -160,6 +162,63 @@ func TestResolveOpenAIWSFallbackErrorResponse(t *testing.T) {
})
}
func TestWriteOpenAIWSFallbackErrorResponseMarksDefaultClientRouteUnknown(t *testing.T) {
gin.SetMode(gin.TestMode)
c, _ := gin.CreateTestContext(httptest.NewRecorder())
svc := &OpenAIGatewayService{}
account := &Account{ID: 42, Name: "default-client", Platform: PlatformOpenAI}
written := svc.writeOpenAIWSFallbackErrorResponse(
c,
account,
wrapOpenAIWSFallback("auth_failed", errors.New("unauthorized")),
)
require.True(t, written)
raw, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events, ok := raw.([]*OpsUpstreamErrorEvent)
require.True(t, ok)
require.Len(t, events, 1)
require.Nil(t, events[0].ProxyID)
require.Equal(t, opsProxyNameUnknown, events[0].ProxyName)
}
func TestWriteOpenAIWSFallbackErrorResponseKeepsManagedProxy(t *testing.T) {
gin.SetMode(gin.TestMode)
c, _ := gin.CreateTestContext(httptest.NewRecorder())
svc := &OpenAIGatewayService{}
proxyID := int64(10060)
account := &Account{ID: 42, Name: "proxied", Platform: PlatformOpenAI, ProxyID: &proxyID, Proxy: &Proxy{ID: proxyID, Name: "ws-proxy"}}
require.True(t, svc.writeOpenAIWSFallbackErrorResponse(c, account, wrapOpenAIWSFallback("auth_failed", errors.New("unauthorized"))))
raw, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events, ok := raw.([]*OpsUpstreamErrorEvent)
require.True(t, ok)
require.Len(t, events, 1)
require.NotNil(t, events[0].ProxyID)
require.Equal(t, proxyID, *events[0].ProxyID)
require.Equal(t, "ws-proxy", events[0].ProxyName)
}
func TestOpsUpstreamWSProxyAttributionNeverReportsDirect(t *testing.T) {
proxyID := int64(7)
id, name := opsUpstreamWSProxyAttribution(nil)
require.Nil(t, id)
require.Equal(t, opsProxyNameUnknown, name)
id, name = opsUpstreamWSProxyAttribution(&Account{})
require.Nil(t, id)
require.Equal(t, opsProxyNameUnknown, name, "no managed proxy => http.DefaultClient => unknown, never direct")
id, name = opsUpstreamWSProxyAttribution(&Account{ProxyID: &proxyID, Proxy: &Proxy{ID: proxyID, Name: "ws-proxy"}})
require.NotNil(t, id)
require.Equal(t, proxyID, *id)
require.Equal(t, "ws-proxy", name)
}
func TestOpenAIWSFallbackCooling(t *testing.T) {
svc := &OpenAIGatewayService{cfg: &config.Config{}}
svc.cfg.Gateway.OpenAIWS.FallbackCooldownSeconds = 1
backend/internal/service/openai_ws_v2_passthrough_adapter.go
+5
View File
@@ -1396,10 +1396,15 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough(
var firstOutputTimeoutErr *openAIWSPassthroughFirstOutputTimeoutError
if errors.As(relayErr, &firstOutputTimeoutErr) {
deadline := firstOutputTimeoutErr.deadline
// The relay ran over the WebSocket transport, so a missing managed
// proxy is an unknown route (http.DefaultClient), not a direct one.
wsProxyID, wsProxyName := opsUpstreamWSProxyAttribution(account)
failoverErr := s.newOpenAIFirstOutputTimeoutError(
ctx,
c,
account,
wsProxyID,
wsProxyName,
deadline.startedAt,
deadline.requestModel,
deadline.reasoningEffort,
backend/internal/service/ops_queue_sanitize_test.go
+155 -3
View File
@@ -8,7 +8,10 @@ import (
func TestSanitizeOpsUpstreamErrorsForQueueBoundsAndRedacts(t *testing.T) {
entry := &OpsInsertErrorLogInput{}
for i := 0; i < 20; i++ {
proxyID := int64(i + 1)
entry.UpstreamErrors = append(entry.UpstreamErrors, &OpsUpstreamErrorEvent{
ProxyID: &proxyID,
ProxyName: "proxy-" + string(rune('a'+i)),
Platform: strings.Repeat("p", 100),
AccountName: strings.Repeat("a", 300),
UpstreamStatusCode: 500,
@@ -32,18 +35,167 @@ func TestSanitizeOpsUpstreamErrorsForQueueBoundsAndRedacts(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if len(events) != 16 {
t.Fatalf("event count = %d, want 16", len(events))
if len(events) != 20 {
t.Fatalf("event count = %d, want 20", len(events))
}
for _, event := range events {
for i, event := range events {
if event.ProxyID == nil || *event.ProxyID != int64(i+1) {
t.Fatalf("event %d proxy id = %v, want %d", i, event.ProxyID, i+1)
}
if len(event.Platform) > 32 || len(event.AccountName) > 128 || len(event.UpstreamURL) > 2048 || len(event.Message) > 2048 {
t.Fatalf("event fields were not bounded: %+v", event)
}
if i < 4 && (len(event.UpstreamURL) > opsUpstreamErrorsOlderURLMaxLen || len(event.Message) > opsUpstreamErrorsOlderMessageMaxLen) {
t.Fatalf("older event %d kept a full-size url/message: url=%d message=%d", i, len(event.UpstreamURL), len(event.Message))
}
if event.DroppedEarlierAttempts != 0 {
t.Fatalf("event %d reported dropped attempts without any drop: %d", i, event.DroppedEarlierAttempts)
}
if len(event.UpstreamResponseBody) > OpsErrorLogQueueBodyMaxBytes || len(event.Detail) > OpsErrorLogQueueBodyMaxBytes {
t.Fatal("event body/detail exceeded queue limit")
}
if strings.Contains(event.UpstreamResponseBody, "Bearer secret") || strings.Contains(event.Detail, `"secret"`) {
t.Fatal("credential material was not redacted")
}
if i < 4 && (event.UpstreamResponseBody != "" || event.Detail != "") {
t.Fatal("older event retained a large body outside the queue body window")
}
}
}
func TestSanitizeOpsUpstreamErrorsForQueueHardBoundsEventCount(t *testing.T) {
entry := &OpsInsertErrorLogInput{}
total := opsUpstreamErrorsMaxEvents + 40
for i := 0; i < total; i++ {
entry.UpstreamErrors = append(entry.UpstreamErrors, &OpsUpstreamErrorEvent{
AtUnixMs: int64(i + 1),
ProxyName: opsProxyNameDirect,
UpstreamStatusCode: 500,
Message: "attempt",
})
}
if err := SanitizeOpsUpstreamErrorsForQueue(entry); err != nil {
t.Fatal(err)
}
events, err := ParseOpsUpstreamErrors(*entry.UpstreamErrorsJSON)
if err != nil {
t.Fatal(err)
}
if len(events) != opsUpstreamErrorsMaxEvents {
t.Fatalf("event count = %d, want %d", len(events), opsUpstreamErrorsMaxEvents)
}
// Newest attempts win; the oldest retained one carries the drop count.
if events[len(events)-1].AtUnixMs != int64(total) {
t.Fatalf("newest event lost: at=%d want %d", events[len(events)-1].AtUnixMs, total)
}
if events[0].AtUnixMs != int64(total-opsUpstreamErrorsMaxEvents+1) {
t.Fatalf("oldest retained at=%d", events[0].AtUnixMs)
}
if events[0].DroppedEarlierAttempts != 40 {
t.Fatalf("dropped_earlier_attempts = %d, want 40", events[0].DroppedEarlierAttempts)
}
if events[1].DroppedEarlierAttempts != 0 {
t.Fatal("drop count must only be stamped on the oldest retained event")
}
}
func TestSanitizeOpsUpstreamErrorsForQueueHardBoundsSerializedBytes(t *testing.T) {
entry := &OpsInsertErrorLogInput{}
// Every bounded field at its cap: the 16 newest attempts weigh ~20KB each
// (8KB body + 8KB detail + 2KB url + 2KB message), older ones ~2KB. At the
// count cap that is well past the byte budget, so bytes must win here.
const total = opsUpstreamErrorsMaxEvents
for i := 0; i < total; i++ {
entry.UpstreamErrors = append(entry.UpstreamErrors, &OpsUpstreamErrorEvent{
AtUnixMs: int64(i + 1),
Platform: strings.Repeat("p", 40),
AccountName: strings.Repeat("a", 200),
ProxyName: strings.Repeat("n", 200),
UpstreamRequestID: strings.Repeat("r", 200),
UpstreamStatusCode: 502,
UpstreamURL: "https://example.com/" + strings.Repeat("u", 3000),
// Plain-text payloads are truncated, not JSON-trimmed, so they keep
// their full queue-limit weight.
UpstreamResponseBody: strings.Repeat("b", 10_000),
Kind: strings.Repeat("k", 80),
Stage: strings.Repeat("s", 80),
Scope: strings.Repeat("c", 80),
Reason: strings.Repeat("e", 200),
Message: strings.Repeat("m", 3000),
Detail: strings.Repeat("d", 10_000),
})
}
if err := SanitizeOpsUpstreamErrorsForQueue(entry); err != nil {
t.Fatal(err)
}
// The newest event is always kept even when it alone exceeds the remaining
// budget, so allow one max-size event of slack.
if len(*entry.UpstreamErrorsJSON) > opsUpstreamErrorsQueueMaxBytes+24*1024 {
t.Fatalf("serialized upstream_errors = %d bytes, budget %d", len(*entry.UpstreamErrorsJSON), opsUpstreamErrorsQueueMaxBytes)
}
events, err := ParseOpsUpstreamErrors(*entry.UpstreamErrorsJSON)
if err != nil {
t.Fatal(err)
}
if len(events) == 0 || len(events) >= total {
t.Fatalf("expected the byte budget to drop some events, kept %d of %d", len(events), total)
}
if events[len(events)-1].AtUnixMs != int64(total) {
t.Fatal("newest attempt must always be retained")
}
if events[0].DroppedEarlierAttempts != total-len(events) {
t.Fatalf("dropped_earlier_attempts = %d, want %d", events[0].DroppedEarlierAttempts, total-len(events))
}
}
func TestSanitizeOpsUpstreamErrorsForQueueDropsEmptyButKeepsOlderTrimmedEvents(t *testing.T) {
entry := &OpsInsertErrorLogInput{}
// 20 events: the first 4 fall outside the body window. Event 0 is a real
// attempt whose only payload is a detail (status 0, no message); it must
// survive the window even though its detail is cleared. Event 1 is truly
// empty and must be dropped.
for i := 0; i < 20; i++ {
ev := &OpsUpstreamErrorEvent{AtUnixMs: int64(i + 1), ProxyName: opsProxyNameDirect}
switch i {
case 0:
ev.Detail = "proxy handshake failed"
case 1:
// fully empty
default:
ev.UpstreamStatusCode = 500
}
entry.UpstreamErrors = append(entry.UpstreamErrors, ev)
}
if err := SanitizeOpsUpstreamErrorsForQueue(entry); err != nil {
t.Fatal(err)
}
events, err := ParseOpsUpstreamErrors(*entry.UpstreamErrorsJSON)
if err != nil {
t.Fatal(err)
}
if len(events) != 19 {
t.Fatalf("event count = %d, want 19", len(events))
}
if events[0].AtUnixMs != 1 || events[0].Detail != "" {
t.Fatalf("detail-only older attempt must be kept with detail cleared: %+v", events[0])
}
if events[1].AtUnixMs != 3 {
t.Fatalf("fully-empty attempt must be dropped, got at=%d", events[1].AtUnixMs)
}
}
func TestSanitizeOpsUpstreamErrorsForQueueIgnoresCallerSuppliedDropCount(t *testing.T) {
entry := &OpsInsertErrorLogInput{UpstreamErrors: []*OpsUpstreamErrorEvent{
{UpstreamStatusCode: 500, ProxyName: opsProxyNameDirect, DroppedEarlierAttempts: 99},
}}
if err := SanitizeOpsUpstreamErrorsForQueue(entry); err != nil {
t.Fatal(err)
}
events, err := ParseOpsUpstreamErrors(*entry.UpstreamErrorsJSON)
if err != nil {
t.Fatal(err)
}
if len(events) != 1 || events[0].DroppedEarlierAttempts != 0 {
t.Fatalf("caller-supplied drop count must be reset: %+v", events[0])
}
}
backend/internal/service/ops_repo_mock_test.go
+4
View File
@@ -9,6 +9,7 @@ import (
type opsRepoMock struct {
InsertErrorLogFn func(ctx context.Context, input *OpsInsertErrorLogInput) (int64, error)
BatchInsertErrorLogsFn func(ctx context.Context, inputs []*OpsInsertErrorLogInput) (int64, error)
GetErrorLogByIDFn func(ctx context.Context, id int64) (*OpsErrorLogDetail, error)
BatchInsertSystemLogsFn func(ctx context.Context, inputs []*OpsInsertSystemLogInput) (int64, error)
ListSystemLogsFn func(ctx context.Context, filter *OpsSystemLogFilter) (*OpsSystemLogList, error)
DeleteSystemLogsFn func(ctx context.Context, filter *OpsSystemLogCleanupFilter) (int64, error)
@@ -34,6 +35,9 @@ func (m *opsRepoMock) ListErrorLogs(ctx context.Context, filter *OpsErrorLogFilt
}
func (m *opsRepoMock) GetErrorLogByID(ctx context.Context, id int64) (*OpsErrorLogDetail, error) {
if m.GetErrorLogByIDFn != nil {
return m.GetErrorLogByIDFn(ctx, id)
}
return &OpsErrorLogDetail{}, nil
}
backend/internal/service/ops_service.go
+84 -17
View File
@@ -539,29 +539,54 @@ func (s *OpsService) prepareErrorLogInput(ctx context.Context, entry *OpsInsertE
return entry, true, nil
}
const (
// opsUpstreamErrorsBodyWindow is how many of the newest attempts keep their
// larger detail / upstream_response_body payloads while queued.
opsUpstreamErrorsBodyWindow = 16
// Attempts older than the body window keep scalar metadata only, with URL
// and message trimmed harder than the newest attempts (2048).
opsUpstreamErrorsOlderURLMaxLen = 512
opsUpstreamErrorsOlderMessageMaxLen = 512
// Hard bounds for the upstream_errors array of one queued entry. Newest
// attempts win; the oldest retained attempt records the dropped count.
opsUpstreamErrorsMaxEvents = 256
opsUpstreamErrorsQueueMaxBytes = 512 * 1024
)
func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error {
if entry == nil || len(entry.UpstreamErrors) == 0 {
return nil
}
const maxEvents = 16
events := entry.UpstreamErrors
if len(events) > maxEvents {
events = events[len(events)-maxEvents:]
events := make([]*OpsUpstreamErrorEvent, 0, len(entry.UpstreamErrors))
for _, ev := range entry.UpstreamErrors {
if ev != nil {
events = append(events, ev)
}
}
firstEventWithBody := len(events) - opsUpstreamErrorsBodyWindow
if firstEventWithBody < 0 {
firstEventWithBody = 0
}
sanitized := make([]*OpsUpstreamErrorEvent, 0, len(events))
for _, ev := range events {
if ev == nil {
continue
}
for i, ev := range events {
out := *ev
normalizeOpsUpstreamProxyAttribution(&out)
// Only boundOpsUpstreamErrors may stamp this; never trust caller input.
out.DroppedEarlierAttempts = 0
keepBody := i >= firstEventWithBody
urlMaxLen, messageMaxLen := 2048, 2048
if !keepBody {
urlMaxLen, messageMaxLen = opsUpstreamErrorsOlderURLMaxLen, opsUpstreamErrorsOlderMessageMaxLen
}
out.Platform = truncateString(strings.TrimSpace(out.Platform), 32)
out.AccountName = truncateString(strings.TrimSpace(out.AccountName), 128)
out.ProxyName = truncateString(strings.TrimSpace(out.ProxyName), 128)
out.UpstreamRequestID = truncateString(strings.TrimSpace(out.UpstreamRequestID), 128)
out.UpstreamURL = truncateString(strings.TrimSpace(out.UpstreamURL), 2048)
if body := strings.TrimSpace(out.UpstreamResponseBody); body != "" {
out.UpstreamURL = truncateString(strings.TrimSpace(out.UpstreamURL), urlMaxLen)
if body := strings.TrimSpace(out.UpstreamResponseBody); keepBody && body != "" {
out.UpstreamResponseBody, _ = sanitizeErrorBodyForStorage(body, OpsErrorLogQueueBodyMaxBytes)
} else {
out.UpstreamResponseBody = ""
@@ -582,11 +607,17 @@ func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error {
}
msg := sanitizeUpstreamErrorMessage(strings.TrimSpace(out.Message))
msg = truncateString(msg, 2048)
msg = truncateString(msg, messageMaxLen)
out.Message = msg
detail := strings.TrimSpace(out.Detail)
if detail != "" {
// Drop fully-empty events (can happen if only status code was known).
// Judged on the original detail so an older attempt whose detail is
// cleared by the body window below is still retained.
if out.UpstreamStatusCode == 0 && out.Message == "" && detail == "" {
continue
}
if keepBody && detail != "" {
// Keep upstream detail small while the event waits in the queue.
sanitizedDetail, _ := sanitizeErrorBodyForStorage(detail, OpsErrorLogQueueBodyMaxBytes)
out.Detail = sanitizedDetail
@@ -594,20 +625,51 @@ func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error {
out.Detail = ""
}
// Drop fully-empty events (can happen if only status code was known).
if out.UpstreamStatusCode == 0 && out.Message == "" && out.Detail == "" {
continue
}
evCopy := out
sanitized = append(sanitized, &evCopy)
}
sanitized, _ = boundOpsUpstreamErrors(sanitized)
entry.UpstreamErrorsJSON = marshalOpsUpstreamErrors(sanitized)
entry.UpstreamErrors = nil
return nil
}
// boundOpsUpstreamErrors enforces the per-entry event count and serialized
// byte budget. It walks from the newest attempt backwards so the final,
// outcome-deciding attempts are always retained (the newest event is kept even
// if it alone exceeds the budget), and stamps the oldest retained event with
// the number of dropped earlier attempts. Events must already be sanitized
// copies owned by the caller.
func boundOpsUpstreamErrors(events []*OpsUpstreamErrorEvent) ([]*OpsUpstreamErrorEvent, int) {
if len(events) == 0 {
return events, 0
}
keepFrom := len(events)
budget := opsUpstreamErrorsQueueMaxBytes
for i := len(events) - 1; i >= 0; i-- {
if len(events)-i > opsUpstreamErrorsMaxEvents {
break
}
raw, err := json.Marshal(events[i])
size := 0
if err == nil {
size = len(raw) + 1 // trailing comma / bracket share
}
if i != len(events)-1 && size > budget {
break
}
budget -= size
keepFrom = i
}
kept := events[keepFrom:]
dropped := keepFrom
if dropped > 0 {
kept[0].DroppedEarlierAttempts = dropped
}
return kept, dropped
}
func (s *OpsService) GetErrorLogs(ctx context.Context, filter *OpsErrorLogFilter) (*OpsErrorLogList, error) {
if err := s.RequireMonitoringEnabled(ctx); err != nil {
return nil, err
@@ -682,6 +744,11 @@ func (s *OpsService) GetErrorLogByID(ctx context.Context, id int64) (*OpsErrorLo
}
return nil, infraerrors.InternalServer("OPS_ERROR_LOAD_FAILED", "Failed to load ops error log").WithCause(err)
}
if detail != nil && strings.TrimSpace(detail.UpstreamErrors) != "" {
if normalized, normalizeErr := normalizeOpsUpstreamErrorsJSON(detail.UpstreamErrors); normalizeErr == nil {
detail.UpstreamErrors = normalized
}
}
return detail, nil
}
backend/internal/service/ops_upstream_context.go
+159
View File
@@ -2,11 +2,15 @@ package service
import (
"encoding/json"
"errors"
"strconv"
"strings"
"time"
"github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey"
"github.com/gin-gonic/gin"
"github.com/tidwall/gjson"
"github.com/tidwall/sjson"
)
// Gin context keys used by Ops error logger for capturing upstream error details.
@@ -365,6 +369,17 @@ type OpsUpstreamErrorEvent struct {
AccountID int64 `json:"account_id,omitempty"`
AccountName string `json:"account_name,omitempty"`
// Proxy attribution is an immutable, credential-free snapshot of the route
// used by this attempt. ProxyID is null for direct and unknown routes;
// ProxyName distinguishes direct/no_proxy from unknown.
// Never add proxy URLs or credentials to this event.
ProxyID *int64 `json:"proxy_id"`
ProxyName string `json:"proxy_name"`
// DroppedEarlierAttempts is set on the oldest retained event when queue
// bounds forced earlier attempts of the same request to be discarded.
DroppedEarlierAttempts int `json:"dropped_earlier_attempts,omitempty"`
// Outcome
UpstreamStatusCode int `json:"upstream_status_code,omitempty"`
UpstreamRequestID string `json:"upstream_request_id,omitempty"`
@@ -394,6 +409,14 @@ type OpsUpstreamErrorEvent struct {
SkipMonitoring bool `json:"-"`
}
const (
opsProxyNameDirect = "direct/no_proxy"
opsProxyNameUnknown = "unknown"
// opsProxyNameUnnamed labels a managed proxy whose name is blank. The
// proxies.name column is NOT NULL/non-empty, so this is a defensive value.
opsProxyNameUnnamed = "proxy"
)
func appendOpsUpstreamError(c *gin.Context, ev OpsUpstreamErrorEvent) {
if c == nil {
return
@@ -402,6 +425,7 @@ func appendOpsUpstreamError(c *gin.Context, ev OpsUpstreamErrorEvent) {
ev.AtUnixMs = time.Now().UnixMilli()
}
ev.Platform = strings.TrimSpace(ev.Platform)
normalizeOpsUpstreamProxyAttribution(&ev)
ev.UpstreamRequestID = strings.TrimSpace(ev.UpstreamRequestID)
ev.UpstreamResponseBody = strings.TrimSpace(ev.UpstreamResponseBody)
ev.Kind = strings.TrimSpace(ev.Kind)
@@ -429,6 +453,91 @@ func appendOpsUpstreamError(c *gin.Context, ev OpsUpstreamErrorEvent) {
checkSkipMonitoringForUpstreamEvent(c, &evCopy)
}
// opsUpstreamProxyAttribution derives both attribution fields from one
// decision so they can never disagree. The event is assembled at the failure
// site; forwarding code builds managed proxy routes from Proxy only when the
// binding ID is also set, so the same rule decides the label here:
//
// - nil account -> (nil, unknown)
// - no binding or no hydrated Proxy -> (nil, direct/no_proxy)
// - hydrated Proxy without durable ID -> (nil, unknown): the transport did
// use that proxy, but nothing durable identifies it
// - otherwise -> (Proxy.ID, Proxy.Name or "proxy")
//
// Invariant: proxy_id == null implies proxy_name is one of the two sentinels.
func opsUpstreamProxyAttribution(account *Account) (*int64, string) {
if account == nil {
return nil, opsProxyNameUnknown
}
if account.ProxyID == nil || account.Proxy == nil {
return nil, opsProxyNameDirect
}
if account.Proxy.ID <= 0 {
return nil, opsProxyNameUnknown
}
proxyID := account.Proxy.ID
name := strings.TrimSpace(account.Proxy.Name)
if name == "" {
name = opsProxyNameUnnamed
}
return &proxyID, name
}
func opsUpstreamProxyID(account *Account) *int64 {
proxyID, _ := opsUpstreamProxyAttribution(account)
return proxyID
}
func opsUpstreamProxyName(account *Account) string {
_, name := opsUpstreamProxyAttribution(account)
return name
}
// opsUpstreamWSProxyAttribution is the OpenAI WebSocket variant. The WS dialer
// sets an explicit proxy client only when the account has a usable managed
// proxy; otherwise coder/websocket falls back to http.DefaultClient, which
// honors HTTP_PROXY/HTTPS_PROXY/NO_PROXY. A missing managed proxy is therefore
// unknown, not direct.
func opsUpstreamWSProxyAttribution(account *Account) (*int64, string) {
proxyID, name := opsUpstreamProxyAttribution(account)
if proxyID == nil {
return nil, opsProxyNameUnknown
}
return proxyID, name
}
func setUnknownOpsUpstreamProxy(ev *OpsUpstreamErrorEvent) {
if ev == nil {
return
}
ev.ProxyID = nil
ev.ProxyName = opsProxyNameUnknown
}
// normalizeOpsUpstreamProxyAttribution makes legacy events explicit without
// pretending that the account's current proxy is historical evidence.
func normalizeOpsUpstreamProxyAttribution(ev *OpsUpstreamErrorEvent) {
if ev == nil {
return
}
if ev.ProxyID != nil && *ev.ProxyID <= 0 {
// A non-positive ID never identifies a managed proxy.
ev.ProxyID = nil
}
ev.ProxyName = strings.TrimSpace(ev.ProxyName)
if ev.ProxyID != nil {
if ev.ProxyName == "" {
ev.ProxyName = opsProxyNameUnnamed
}
return
}
// Invariant: proxy_id == null implies a sentinel name. Any other name
// without a durable ID is not historical evidence of a managed route.
if ev.ProxyName != opsProxyNameDirect {
setUnknownOpsUpstreamProxy(ev)
}
}
// checkSkipMonitoringForUpstreamEvent snapshots whether this attempt matches a
// skip_monitoring passthrough rule. The final failure decides whether the
// request error is hidden; an intermediate recovered attempt cannot suppress a
@@ -479,6 +588,56 @@ func ParseOpsUpstreamErrors(raw string) ([]*OpsUpstreamErrorEvent, error) {
if err := json.Unmarshal([]byte(raw), &out); err != nil {
return nil, err
}
for _, ev := range out {
normalizeOpsUpstreamProxyAttribution(ev)
}
return out, nil
}
// normalizeOpsUpstreamErrorsJSON materializes missing proxy attribution on
// stored JSON for detail reads. It edits only the attribution keys of events
// that lack them, so keys written by older struct versions and the original
// key order survive; nothing is re-marshaled through the current struct.
func normalizeOpsUpstreamErrorsJSON(raw string) (string, error) {
raw = strings.TrimSpace(raw)
if raw == "" {
return raw, nil
}
if !gjson.Valid(raw) {
return "", errors.New("upstream_errors is not valid JSON")
}
parsed := gjson.Parse(raw)
if !parsed.IsArray() {
return "", errors.New("upstream_errors is not a JSON array")
}
out := raw
for i, ev := range parsed.Array() {
if !ev.IsObject() {
continue
}
prefix := strconv.Itoa(i) + "."
proxyID := ev.Get("proxy_id")
proxyName := strings.TrimSpace(ev.Get("proxy_name").String())
hasValidID := proxyID.Exists() && proxyID.Type == gjson.Number && proxyID.Int() > 0
var err error
switch {
case hasValidID:
if proxyName == "" {
out, err = sjson.Set(out, prefix+"proxy_name", opsProxyNameUnnamed)
}
case proxyName == opsProxyNameDirect:
if !proxyID.Exists() || proxyID.Type != gjson.Null {
out, err = sjson.Set(out, prefix+"proxy_id", nil)
}
default:
if out, err = sjson.Set(out, prefix+"proxy_id", nil); err == nil {
out, err = sjson.Set(out, prefix+"proxy_name", opsProxyNameUnknown)
}
}
if err != nil {
return "", err
}
}
return out, nil
}
backend/internal/service/ops_upstream_context_test.go
+283
View File
@@ -1,8 +1,12 @@
package service
import (
"context"
"net/http/httptest"
"strings"
"testing"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
)
@@ -26,3 +30,282 @@ func TestSafeUpstreamURL(t *testing.T) {
})
}
}
func TestOpsUpstreamErrorEventKeepsExplicitProxySnapshotPerAttempt(t *testing.T) {
gin.SetMode(gin.TestMode)
c, _ := gin.CreateTestContext(httptest.NewRecorder())
oldProxyID := int64(10060)
oldProxy := &Proxy{
ID: oldProxyID,
Name: "wldsg82-ipv6-10060",
Protocol: "socks5",
Host: "old-proxy.example",
Username: "proxy-user",
Password: "proxy-secret",
}
oldAccount := &Account{
ID: 11,
Name: "old-account",
Platform: PlatformOpenAI,
ProxyID: &oldProxyID,
Proxy: oldProxy,
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
AccountID: oldAccount.ID,
ProxyID: opsUpstreamProxyID(oldAccount),
ProxyName: opsUpstreamProxyName(oldAccount),
Kind: "retry",
})
newProxyID := int64(8001)
newAccount := &Account{
ID: 12,
Name: "new-account",
Platform: PlatformOpenAI,
ProxyID: &newProxyID,
Proxy: &Proxy{ID: newProxyID, Name: "oxylabs-uk-8001"},
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
AccountID: newAccount.ID,
ProxyID: opsUpstreamProxyID(newAccount),
ProxyName: opsUpstreamProxyName(newAccount),
Kind: "failover",
})
// Mutating the selected account after both attempts must not rewrite history.
oldProxy.Name = "mutated-current-proxy"
oldAccount.ProxyID = &newProxyID
oldAccount.Proxy = newAccount.Proxy
raw, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events, ok := raw.([]*OpsUpstreamErrorEvent)
require.True(t, ok)
require.Len(t, events, 2)
require.Equal(t, int64(10060), *events[0].ProxyID)
require.Equal(t, "wldsg82-ipv6-10060", events[0].ProxyName)
require.Equal(t, int64(8001), *events[1].ProxyID)
require.Equal(t, "oxylabs-uk-8001", events[1].ProxyName)
encoded := marshalOpsUpstreamErrors(events)
require.NotNil(t, encoded)
require.NotContains(t, *encoded, "old-proxy.example")
require.NotContains(t, *encoded, "proxy-user")
require.NotContains(t, *encoded, "proxy-secret")
}
func TestOpsUpstreamProxyFieldAccessors(t *testing.T) {
gin.SetMode(gin.TestMode)
boundID := int64(7)
backupID := int64(8)
tests := []struct {
name string
account *Account
wantID *int64
wantName string
}{
{
name: "actual proxy object is authoritative",
account: &Account{ProxyID: &boundID, Proxy: &Proxy{ID: backupID, Name: "backup-uk"}},
wantID: &backupID,
wantName: "backup-uk",
},
{
name: "proxy object without binding id is not a configured route",
account: &Account{Proxy: &Proxy{ID: backupID, Name: "hydrated-proxy"}},
wantName: opsProxyNameDirect,
},
{
name: "hydrated proxy without durable id is unknown, never id=null plus a real name",
account: &Account{ProxyID: &boundID, Proxy: &Proxy{Name: "transient-proxy"}},
wantName: opsProxyNameUnknown,
},
{
name: "blank proxy name gets the unnamed placeholder",
account: &Account{ProxyID: &boundID, Proxy: &Proxy{ID: boundID, Name: " "}},
wantID: &boundID,
wantName: opsProxyNameUnnamed,
},
{
name: "account has no proxy",
account: &Account{},
wantName: opsProxyNameDirect,
},
{
name: "configured proxy was not hydrated and transport is direct",
account: &Account{ProxyID: &backupID},
wantName: opsProxyNameDirect,
},
{
name: "missing attempt account",
account: nil,
wantName: opsProxyNameUnknown,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c, _ := gin.CreateTestContext(httptest.NewRecorder())
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(tt.account),
ProxyName: opsUpstreamProxyName(tt.account),
Kind: "request_error",
})
raw, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events, ok := raw.([]*OpsUpstreamErrorEvent)
require.True(t, ok)
require.Len(t, events, 1)
event := events[0]
if tt.wantID == nil {
require.Nil(t, event.ProxyID)
} else {
require.NotNil(t, event.ProxyID)
require.Equal(t, *tt.wantID, *event.ProxyID)
}
require.Equal(t, tt.wantName, event.ProxyName)
})
}
}
func TestOpsUpstreamErrorEventRetryKeepsExplicitAttemptProxy(t *testing.T) {
gin.SetMode(gin.TestMode)
c, _ := gin.CreateTestContext(httptest.NewRecorder())
proxyID := int64(10060)
account := &Account{ProxyID: &proxyID, Proxy: &Proxy{ID: proxyID, Name: "retry-proxy"}}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Kind: "retry",
})
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
Kind: "retry_exhausted",
})
raw, ok := c.Get(OpsUpstreamErrorsKey)
require.True(t, ok)
events, ok := raw.([]*OpsUpstreamErrorEvent)
require.True(t, ok)
require.Len(t, events, 2)
for _, event := range events {
require.NotNil(t, event.ProxyID)
require.Equal(t, proxyID, *event.ProxyID)
require.Equal(t, "retry-proxy", event.ProxyName)
}
}
func TestNormalizeOpsUpstreamProxyAttributionEnforcesSentinelInvariant(t *testing.T) {
positive := int64(7)
zero := int64(0)
tests := []struct {
name string
in OpsUpstreamErrorEvent
wantID *int64
wantName string
}{
{name: "id with name kept", in: OpsUpstreamErrorEvent{ProxyID: &positive, ProxyName: "eu-1"}, wantID: &positive, wantName: "eu-1"},
{name: "id without name gets placeholder", in: OpsUpstreamErrorEvent{ProxyID: &positive}, wantID: &positive, wantName: opsProxyNameUnnamed},
{name: "null id with real name collapses to unknown", in: OpsUpstreamErrorEvent{ProxyName: "eu-1"}, wantName: opsProxyNameUnknown},
{name: "zero id with real name collapses to unknown", in: OpsUpstreamErrorEvent{ProxyID: &zero, ProxyName: "eu-1"}, wantName: opsProxyNameUnknown},
{name: "explicit direct preserved", in: OpsUpstreamErrorEvent{ProxyName: opsProxyNameDirect}, wantName: opsProxyNameDirect},
{name: "empty is unknown", in: OpsUpstreamErrorEvent{}, wantName: opsProxyNameUnknown},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
ev := tt.in
normalizeOpsUpstreamProxyAttribution(&ev)
if tt.wantID == nil {
require.Nil(t, ev.ProxyID)
} else {
require.NotNil(t, ev.ProxyID)
require.Equal(t, *tt.wantID, *ev.ProxyID)
}
require.Equal(t, tt.wantName, ev.ProxyName)
})
}
}
func TestNormalizeOpsUpstreamErrorsJSONPreservesUnknownKeysAndOrder(t *testing.T) {
// A legacy row written by an older struct version: it carries a key the
// current struct no longer has, and mixes explicit/legacy attribution.
raw := `[` +
`{"at_unix_ms":1,"account_id":42,"upstream_request_body":"legacy-only-key","kind":"http_error"},` +
`{"at_unix_ms":2,"proxy_id":null,"proxy_name":"direct/no_proxy","kind":"request_error"},` +
`{"at_unix_ms":3,"proxy_id":9,"proxy_name":"eu-9","kind":"failover"},` +
`{"at_unix_ms":4,"proxy_id":0,"proxy_name":"stale-name","kind":"retry"},` +
`{"at_unix_ms":5,"proxy_id":11,"proxy_name":"","kind":"retry"}` +
`]`
out, err := normalizeOpsUpstreamErrorsJSON(raw)
require.NoError(t, err)
// Legacy-only key and original key order survive.
require.Contains(t, out, `"upstream_request_body":"legacy-only-key"`)
require.True(t, strings.Index(out, `"at_unix_ms":1`) < strings.Index(out, `"account_id":42`))
events, err := ParseOpsUpstreamErrors(out)
require.NoError(t, err)
require.Len(t, events, 5)
require.Nil(t, events[0].ProxyID)
require.Equal(t, opsProxyNameUnknown, events[0].ProxyName)
require.Nil(t, events[1].ProxyID)
require.Equal(t, opsProxyNameDirect, events[1].ProxyName)
require.NotNil(t, events[2].ProxyID)
require.Equal(t, int64(9), *events[2].ProxyID)
require.Equal(t, "eu-9", events[2].ProxyName)
require.Nil(t, events[3].ProxyID)
require.Equal(t, opsProxyNameUnknown, events[3].ProxyName)
require.NotNil(t, events[4].ProxyID)
require.Equal(t, opsProxyNameUnnamed, events[4].ProxyName)
// Already-normalized input is returned byte-for-byte.
again, err := normalizeOpsUpstreamErrorsJSON(out)
require.NoError(t, err)
require.Equal(t, out, again)
_, err = normalizeOpsUpstreamErrorsJSON(`{"not":"an array"}`)
require.Error(t, err)
_, err = normalizeOpsUpstreamErrorsJSON(`[not json`)
require.Error(t, err)
}
func TestParseOpsUpstreamErrorsMarksLegacyProxyAttributionUnknown(t *testing.T) {
events, err := ParseOpsUpstreamErrors(`[{"account_id":42,"account_name":"legacy","kind":"http_error"}]`)
require.NoError(t, err)
require.Len(t, events, 1)
require.Nil(t, events[0].ProxyID)
require.Equal(t, opsProxyNameUnknown, events[0].ProxyName)
}
func TestParseOpsUpstreamErrorsPreservesExplicitDirectAttribution(t *testing.T) {
events, err := ParseOpsUpstreamErrors(`[{"proxy_id":null,"proxy_name":"direct/no_proxy","kind":"request_error"}]`)
require.NoError(t, err)
require.Len(t, events, 1)
require.Nil(t, events[0].ProxyID)
require.Equal(t, opsProxyNameDirect, events[0].ProxyName)
}
func TestOpsServiceGetErrorLogByIDNormalizesLegacyProxyAttribution(t *testing.T) {
repo := &opsRepoMock{
GetErrorLogByIDFn: func(context.Context, int64) (*OpsErrorLogDetail, error) {
return &OpsErrorLogDetail{UpstreamErrors: `[{"account_id":42,"kind":"http_error"}]`}, nil
},
}
svc := NewOpsService(repo, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
detail, err := svc.GetErrorLogByID(context.Background(), 1)
require.NoError(t, err)
require.NotNil(t, detail)
require.Contains(t, detail.UpstreamErrors, `"proxy_name":"unknown"`)
require.Contains(t, detail.UpstreamErrors, `"proxy_id":null`)
require.NotContains(t, detail.UpstreamErrors, `"proxy_mode"`)
require.NotContains(t, detail.UpstreamErrors, `"proxy_source"`)
require.NotContains(t, detail.UpstreamErrors, `"proxy_fallback"`)
}