fix(ops): close proxy attribution gaps and bound queued upstream events

Follow-up to the event-time proxy snapshot: every upstream attempt now
carries attribution, the direct/unknown contract is enforced everywhere,
and the queued upstream_errors array has hard bounds again.

Attribution coverage
- compact fallback retry helper and the streaming compact failover event
  now stamp proxy_id/proxy_name; the non-streaming SSE compact retry
  previously dropped its first attempt on `continue` and now records it
- raw chat-completions stream truncation events (new on main) attributed
- Anthropic/Bedrock transport errors get attribution inside the shared
  handleUpstreamTransportError helper, so all five callers inherit it
- OpenAI first-output timeout takes attribution from the caller: HTTP
  callers pass the HTTP rule, the WS passthrough relay passes the WS rule
  (no managed proxy => unknown, never direct, because http.DefaultClient
  may honor environment proxies)

Contract hardening
- single opsUpstreamProxyAttribution returns (id, name) from one decision;
  a hydrated Proxy without a durable ID is unknown, not id=null+name
- both normalizers (struct at append time, JSON on detail reads) enforce
  proxy_id==null <=> sentinel name; blank managed name => "proxy"
- detail-read normalization edits only the attribution keys via sjson, so
  keys from older struct versions and stored key order are preserved

Queue bounds
- keep at most 256 attempts and ~512 KiB serialized per entry, newest
  first; the oldest retained event records dropped_earlier_attempts
- attempts outside the 16-event body window trim url/message to 512 bytes
- restore the empty-event drop rule, judged before the body window clears
  detail so older real attempts survive

Tests cover each site above (managed proxy, direct, WS unknown), the
normalizer invariants, key/order preservation, and the count/byte bounds.
This commit is contained in:
Wei-Shaw
2026-09-03 15:11:36 +08:00
parent e9e3c46cb7
commit 4c1f920d58
18 changed files with 749 additions and 57 deletions
@@ -30,6 +30,12 @@ The two sentinel names have strict meanings:
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,
@@ -59,6 +65,10 @@ 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
@@ -68,22 +78,40 @@ 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.
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 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.
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 and the shared parser
materialize missing attribution as:
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
{
@@ -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
@@ -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.
@@ -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,
@@ -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)
}
@@ -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,8 +268,8 @@ func (s *OpenAIGatewayService) newOpenAIFirstOutputTimeoutError(
)
requestID := strings.TrimSpace(responseHeaders.Get("x-request-id"))
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: opsUpstreamProxyID(account),
ProxyName: opsUpstreamProxyName(account),
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",
@@ -152,6 +152,32 @@ 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 := raw.([]*OpsUpstreamErrorEvent)
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,
@@ -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 空流。
@@ -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,
)
}
@@ -1175,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,
@@ -1206,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
@@ -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,
)
@@ -892,14 +892,7 @@ 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
}
proxyID, proxyName := opsUpstreamWSProxyAttribution(account)
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
ProxyID: proxyID,
ProxyName: proxyName,
@@ -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,
@@ -184,6 +184,40 @@ func TestWriteOpenAIWSFallbackErrorResponseMarksDefaultClientRouteUnknown(t *tes
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 := raw.([]*OpsUpstreamErrorEvent)
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
@@ -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,
@@ -45,6 +45,12 @@ func TestSanitizeOpsUpstreamErrorsForQueueBoundsAndRedacts(t *testing.T) {
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")
}
@@ -56,3 +62,140 @@ func TestSanitizeOpsUpstreamErrorsForQueueBoundsAndRedacts(t *testing.T) {
}
}
}
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])
}
}
+72 -9
View File
@@ -539,32 +539,53 @@ 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
}
events := entry.UpstreamErrors
const maxEventsWithBodies = 16
firstEventWithBody := len(events) - maxEventsWithBodies
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 i, ev := range events {
if ev == nil {
continue
}
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)
keepBody := i >= firstEventWithBody
out.UpstreamURL = truncateString(strings.TrimSpace(out.UpstreamURL), urlMaxLen)
if body := strings.TrimSpace(out.UpstreamResponseBody); keepBody && body != "" {
out.UpstreamResponseBody, _ = sanitizeErrorBodyForStorage(body, OpsErrorLogQueueBodyMaxBytes)
} else {
@@ -586,10 +607,16 @@ 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)
// 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)
@@ -602,11 +629,47 @@ func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error {
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
+104 -23
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.
@@ -372,6 +376,10 @@ type OpsUpstreamErrorEvent struct {
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"`
@@ -404,6 +412,9 @@ type OpsUpstreamErrorEvent struct {
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) {
@@ -442,27 +453,57 @@ 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
// 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
return &proxyID
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 {
if account == nil {
return opsProxyNameUnknown
_, 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
}
if account.ProxyID == nil || account.Proxy == nil {
return opsProxyNameDirect
}
if name := strings.TrimSpace(account.Proxy.Name); name != "" {
return name
}
return "proxy"
return proxyID, name
}
func setUnknownOpsUpstreamProxy(ev *OpsUpstreamErrorEvent) {
@@ -480,10 +521,19 @@ func normalizeOpsUpstreamProxyAttribution(ev *OpsUpstreamErrorEvent) {
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.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)
}
}
@@ -544,20 +594,51 @@ func ParseOpsUpstreamErrors(raw string) ([]*OpsUpstreamErrorEvent, error) {
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
}
events, err := ParseOpsUpstreamErrors(raw)
if err != nil {
return "", err
if !gjson.Valid(raw) {
return "", errors.New("upstream_errors is not valid JSON")
}
normalized, err := json.Marshal(events)
if err != nil {
return "", err
parsed := gjson.Parse(raw)
if !parsed.IsArray() {
return "", errors.New("upstream_errors is not a JSON array")
}
return string(normalized), nil
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
}
// safeUpstreamURL returns scheme + host + path from a URL, stripping query/fragment
@@ -3,6 +3,7 @@ package service
import (
"context"
"net/http/httptest"
"strings"
"testing"
"github.com/gin-gonic/gin"
@@ -118,6 +119,17 @@ func TestOpsUpstreamProxyFieldAccessors(t *testing.T) {
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{},
@@ -189,6 +201,81 @@ func TestOpsUpstreamErrorEventRetryKeepsExplicitAttemptProxy(t *testing.T) {
}
}
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)