fix(ops): snapshot proxy on upstream errors

This commit is contained in:
wwcchh0123
2026-09-03 14:20:39 +08:00
committed by shaw
parent 5097b31457
commit e9e3c46cb7
46 changed files with 764 additions and 30 deletions
@@ -0,0 +1,108 @@
# 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.
## 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.
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.
`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 preserves every non-nil attempt event. To bound asynchronous
queue memory, only the last 16 attempts retain the larger `detail` and
`upstream_response_body` values; older attempts keep their timestamps, account,
proxy, status, kind, message, and other scalar metadata.
## Legacy events and analytics
Legacy JSON is not rewritten in the database. Detail reads and the shared parser
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.
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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),
@@ -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 {
+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,
@@ -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,
@@ -262,7 +262,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: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
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()),
@@ -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)
@@ -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)
@@ -106,7 +106,7 @@ func (s *OpenAIGatewayService) forwardChatCompletionsViaNativeAnthropic(
}
proxyURL := ""
if account.Proxy != nil {
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
@@ -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,
@@ -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)
@@ -1108,6 +1108,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,
@@ -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,
@@ -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,
@@ -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),
@@ -86,7 +86,7 @@ func (s *OpenAIGatewayService) forwardAnthropicViaNativeAnthropicEndpoint(
}
proxyURL := ""
if account.Proxy != nil {
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
@@ -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),
@@ -111,7 +111,7 @@ func (s *OpenAIGatewayService) forwardResponsesViaNativeAnthropic(
}
proxyURL := ""
if account.Proxy != nil {
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
@@ -892,7 +892,17 @@ func (s *OpenAIGatewayService) writeOpenAIWSFallbackErrorResponse(c *gin.Context
setOpsUpstreamError(c, statusCode, upstreamMessage, "")
if account != nil {
proxyID := opsUpstreamProxyID(account)
proxyName := opsUpstreamProxyName(account)
// The existing WS client uses the configured proxy only when both fields
// are present; otherwise http.DefaultClient may select an environment proxy.
if account.ProxyID == nil || account.Proxy == nil {
proxyID = nil
proxyName = opsProxyNameUnknown
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: proxyID,
ProxyName: proxyName,
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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) {
@@ -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,28 @@ 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 TestOpenAIWSFallbackCooling(t *testing.T) {
svc := &OpenAIGatewayService{cfg: &config.Config{}}
svc.cfg.Gateway.OpenAIWS.FallbackCooldownSeconds = 1
@@ -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,10 +35,13 @@ 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)
}
@@ -45,5 +51,8 @@ func TestSanitizeOpsUpstreamErrorsForQueueBoundsAndRedacts(t *testing.T) {
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")
}
}
}
@@ -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
}
+15 -11
View File
@@ -544,24 +544,28 @@ func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error {
return nil
}
const maxEvents = 16
events := entry.UpstreamErrors
if len(events) > maxEvents {
events = events[len(events)-maxEvents:]
const maxEventsWithBodies = 16
firstEventWithBody := len(events) - maxEventsWithBodies
if firstEventWithBody < 0 {
firstEventWithBody = 0
}
sanitized := make([]*OpsUpstreamErrorEvent, 0, len(events))
for _, ev := range events {
for i, ev := range events {
if ev == nil {
continue
}
out := *ev
normalizeOpsUpstreamProxyAttribution(&out)
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 != "" {
keepBody := i >= firstEventWithBody
if body := strings.TrimSpace(out.UpstreamResponseBody); keepBody && body != "" {
out.UpstreamResponseBody, _ = sanitizeErrorBodyForStorage(body, OpsErrorLogQueueBodyMaxBytes)
} else {
out.UpstreamResponseBody = ""
@@ -586,7 +590,7 @@ func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error {
out.Message = msg
detail := strings.TrimSpace(out.Detail)
if detail != "" {
if keepBody && detail != "" {
// Keep upstream detail small while the event waits in the queue.
sanitizedDetail, _ := sanitizeErrorBodyForStorage(detail, OpsErrorLogQueueBodyMaxBytes)
out.Detail = sanitizedDetail
@@ -594,11 +598,6 @@ 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)
}
@@ -682,6 +681,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
}
@@ -365,6 +365,13 @@ 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"`
// Outcome
UpstreamStatusCode int `json:"upstream_status_code,omitempty"`
UpstreamRequestID string `json:"upstream_request_id,omitempty"`
@@ -394,6 +401,11 @@ type OpsUpstreamErrorEvent struct {
SkipMonitoring bool `json:"-"`
}
const (
opsProxyNameDirect = "direct/no_proxy"
opsProxyNameUnknown = "unknown"
)
func appendOpsUpstreamError(c *gin.Context, ev OpsUpstreamErrorEvent) {
if c == nil {
return
@@ -402,6 +414,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 +442,52 @@ func appendOpsUpstreamError(c *gin.Context, ev OpsUpstreamErrorEvent) {
checkSkipMonitoringForUpstreamEvent(c, &evCopy)
}
// The event is assembled at the failure site. Forwarding code builds managed
// proxy routes from Proxy, so that event-time object is authoritative.
func opsUpstreamProxyID(account *Account) *int64 {
if account == nil || account.ProxyID == nil || account.Proxy == nil || account.Proxy.ID <= 0 {
return nil
}
proxyID := account.Proxy.ID
return &proxyID
}
func opsUpstreamProxyName(account *Account) string {
if account == nil {
return opsProxyNameUnknown
}
if account.ProxyID == nil || account.Proxy == nil {
return opsProxyNameDirect
}
if name := strings.TrimSpace(account.Proxy.Name); name != "" {
return name
}
return "proxy"
}
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 {
ev.ProxyID = nil
}
ev.ProxyName = strings.TrimSpace(ev.ProxyName)
if ev.ProxyName == "" {
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,9 +538,28 @@ 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
}
func normalizeOpsUpstreamErrorsJSON(raw string) (string, error) {
raw = strings.TrimSpace(raw)
if raw == "" {
return raw, nil
}
events, err := ParseOpsUpstreamErrors(raw)
if err != nil {
return "", err
}
normalized, err := json.Marshal(events)
if err != nil {
return "", err
}
return string(normalized), nil
}
// safeUpstreamURL returns scheme + host + path from a URL, stripping query/fragment
// to avoid leaking sensitive query parameters (e.g. OAuth tokens).
func safeUpstreamURL(rawURL string) string {
@@ -1,8 +1,11 @@
package service
import (
"context"
"net/http/httptest"
"testing"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
)
@@ -26,3 +29,196 @@ 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: "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 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"`)
}