From 6327573534d903efc1d63d0e3af92d683d3d293e Mon Sep 17 00:00:00 2001 From: alfadb Date: Tue, 28 Apr 2026 19:12:48 +0800 Subject: [PATCH 1/3] fix(gateway): wrap Anthropic stream EOF as failover error before client output MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Anthropic streaming path (gateway_service.go) returned a plain error on upstream SSE read failure, so the handler-level UpstreamFailoverError check never fired and the client received a bare `stream_read_error` event, breaking long-running tasks even when no bytes had been written yet. The most common trigger is HTTP/2 GOAWAY from api.anthropic.com edge backends doing graceful rotation: Go's http.Transport surfaces this as `unexpected EOF` and never auto-retries. Mirror what the OpenAI and antigravity gateways already do: when the read error happens before any byte has reached the client (`!c.Writer.Written()`), return `*UpstreamFailoverError{StatusCode: 502, RetryableOnSameAccount: true}` so the handler can retry on the same or another account. After client output has begun, SSE has no resume protocol — keep the existing passthrough behavior. Tests cover both branches via streamReadCloser-based fixtures. Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/internal/service/gateway_service.go | 14 ++++ .../service/gateway_streaming_test.go | 70 +++++++++++++++++++ 2 files changed, 84 insertions(+) diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 6be19ba6..911bc6fc 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -7041,6 +7041,20 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http sendErrorEvent("response_too_large") return &streamingResult{usage: usage, firstTokenMs: firstTokenMs}, ev.err } + // 上游中途读错误（unexpected EOF / connection reset 等，常见于 HTTP/2 GOAWAY）： + // 若尚未向客户端写过任何字节，包成 UpstreamFailoverError 让 handler 层走 failover/重试。 + // 已经开始写流时 SSE 协议无 resume，只能透传错误事件给客户端。 + if !c.Writer.Written() { + logger.LegacyPrintf("service.gateway", "Upstream stream read error before any client output (account=%d), failing over: %v", account.ID, ev.err) + body, _ := json.Marshal(map[string]string{ + "error": fmt.Sprintf("upstream stream disconnected: %s", ev.err), + }) + return nil, &UpstreamFailoverError{ + StatusCode: http.StatusBadGateway, + ResponseBody: body, + RetryableOnSameAccount: true, + } + } sendErrorEvent("stream_read_error") return &streamingResult{usage: usage, firstTokenMs: firstTokenMs}, fmt.Errorf("stream read error: %w", ev.err) } diff --git a/backend/internal/service/gateway_streaming_test.go b/backend/internal/service/gateway_streaming_test.go index b1584827..389831fa 100644 --- a/backend/internal/service/gateway_streaming_test.go +++ b/backend/internal/service/gateway_streaming_test.go @@ -4,9 +4,11 @@ package service import ( "context" + "errors" "io" "net/http" "net/http/httptest" + "strings" "testing" "time" @@ -218,3 +220,71 @@ func TestHandleStreamingResponse_SpecialCharactersInJSON(t *testing.T) { body := rec.Body.String() require.Contains(t, body, "content_block_delta", "响应应包含转发的 SSE 事件") } + +// 上游中途读错误（如 HTTP/2 GOAWAY 触发的 unexpected EOF）发生在向客户端写入任何字节前： +// 网关应返回 *UpstreamFailoverError 触发账号 failover/重试，而不是把错误事件直接发给客户端。 +func TestHandleStreamingResponse_StreamReadErrorBeforeOutput_TriggersFailover(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := newMinimalGatewayService() + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: &streamReadCloser{err: io.ErrUnexpectedEOF}, + } + + result, err := svc.handleStreamingResponse(context.Background(), resp, c, &Account{ID: 1}, time.Now(), "model", "model", false) + + require.Error(t, err) + require.Nil(t, result, "失败移交场景下不应返回 streamingResult") + + var failoverErr *UpstreamFailoverError + require.True(t, errors.As(err, &failoverErr), "未输出过字节时 stream read error 必须包成 UpstreamFailoverError，期望: %v", err) + require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode) + require.True(t, failoverErr.RetryableOnSameAccount, "GOAWAY 类错误应允许同账号重试") + require.Contains(t, string(failoverErr.ResponseBody), "upstream stream disconnected") + + // 客户端应收不到任何 stream_read_error 事件，由 handler 层根据 failover 结果再决定 + require.NotContains(t, rec.Body.String(), "stream_read_error") +} + +// 上游已经发送过事件（c.Writer 已写过字节）后再发生读错误： +// SSE 协议无 resume，网关只能透传 stream_read_error 错误事件给客户端，不能 failover。 +func TestHandleStreamingResponse_StreamReadErrorAfterOutput_PassesThrough(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := newMinimalGatewayService() + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + + // 第一次 Read 返回完整 SSE 事件让网关向 client 写入字节，第二次 Read 返回 EOF + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: &streamReadCloser{ + payload: []byte("data: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":5}}}\n\n"), + err: io.ErrUnexpectedEOF, + }, + } + + result, err := svc.handleStreamingResponse(context.Background(), resp, c, &Account{ID: 1}, time.Now(), "model", "model", false) + + require.Error(t, err) + require.Contains(t, err.Error(), "stream read error", "已开始流后应透传普通 stream read error") + require.NotNil(t, result, "透传场景下应返回已收集的 streamingResult") + + // 不应被错误地包成 failover error + var failoverErr *UpstreamFailoverError + require.False(t, errors.As(err, &failoverErr), "已经向客户端写过字节时不能再 failover") + + // 客户端必须收到 stream_read_error 事件 + body := rec.Body.String() + require.True(t, + strings.Contains(body, "stream_read_error"), + "已开始流后必须发送 stream_read_error 事件给客户端，实际响应: %q", body) +} From 4c474616b994665a104c5eeb1ccd8c5e96a31ddf Mon Sep 17 00:00:00 2001 From: alfadb Date: Tue, 28 Apr 2026 20:24:17 +0800 Subject: [PATCH 2/3] fix(gateway): emit Anthropic-standard SSE error events and failover body MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two follow-ups to PR #2066's failover-wrap fix: 1. Failover ResponseBody (`UpstreamFailoverError.ResponseBody`) was encoded as `{"error": ""}` (string field). `ExtractUpstreamErrorMessage` probes for `error.message`, `detail`, or top-level `message` only — so `handleFailoverExhausted` and downstream passthrough rules saw an empty message, losing the EOF root cause in ops logs. Re-encode as the Anthropic standard shape `{"type":"error","error":{"type":"upstream_disconnected","message":"..."}}`. (Addresses the inline review comment from copilot-pull-request-reviewer on Wei-Shaw/sub2api#2066.) 2. The streaming `event: error` SSE frame for `response_too_large`, `stream_read_error`, and `stream_timeout` was non-standard (`{"error":""}`). Anthropic SDKs (and Claude Code) expect `{"type":"error","error":{"type":"...","message":"..."}}` and parse `error.type`/`error.message` accordingly. Refactor `sendErrorEvent` to take both reason and message, and emit the standard frame so client SDKs surface a real diagnostic message instead of a generic stream error. This does not by itself prevent task interruption on long-stream EOF (SSE has no resume; client-side retry remains the only complete fix), but it gives both server-side ops logs and client-side error UIs a meaningful upstream message so users know the next step is to retry. Tests updated to assert the new body shape on both branches plus a new assertion that `ExtractUpstreamErrorMessage` returns a non-empty string. Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/internal/service/gateway_service.go | 38 +++++++++++++++---- .../service/gateway_streaming_test.go | 21 +++++++--- 2 files changed, 45 insertions(+), 14 deletions(-) diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 911bc6fc..4c4a9b82 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -6871,14 +6871,31 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http } lastDataAt := time.Now() - // 仅发送一次错误事件，避免多次写入导致协议混乱（写失败时尽力通知客户端） + // 仅发送一次错误事件，避免多次写入导致协议混乱（写失败时尽力通知客户端）。 + // 事件格式遵循 Anthropic SSE 标准：{"type":"error","error":{"type":,"message":}} + // 这样 Anthropic SDK / Claude Code 等客户端能按标准 error 类型解析，UI 能显示具体错误文案， + // 服务端 ExtractUpstreamErrorMessage 也能从透传的 body 中提取 message。 errorEventSent := false - sendErrorEvent := func(reason string) { + sendErrorEvent := func(reason, message string) { if errorEventSent { return } errorEventSent = true - _, _ = fmt.Fprintf(w, "event: error\ndata: {\"error\":\"%s\"}\n\n", reason) + if message == "" { + message = reason + } + body, err := json.Marshal(map[string]any{ + "type": "error", + "error": map[string]string{ + "type": reason, + "message": message, + }, + }) + if err != nil { + // json.Marshal 不可能在已知 string-only 输入上失败，保守 fallback + body = []byte(fmt.Sprintf(`{"type":"error","error":{"type":%q,"message":%q}}`, reason, message)) + } + _, _ = fmt.Fprintf(w, "event: error\ndata: %s\n\n", body) flusher.Flush() } @@ -7038,16 +7055,21 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http // 客户端未断开，正常的错误处理 if errors.Is(ev.err, bufio.ErrTooLong) { logger.LegacyPrintf("service.gateway", "SSE line too long: account=%d max_size=%d error=%v", account.ID, maxLineSize, ev.err) - sendErrorEvent("response_too_large") + sendErrorEvent("response_too_large", fmt.Sprintf("upstream SSE line exceeded %d bytes", maxLineSize)) return &streamingResult{usage: usage, firstTokenMs: firstTokenMs}, ev.err } // 上游中途读错误（unexpected EOF / connection reset 等，常见于 HTTP/2 GOAWAY）： // 若尚未向客户端写过任何字节，包成 UpstreamFailoverError 让 handler 层走 failover/重试。 // 已经开始写流时 SSE 协议无 resume，只能透传错误事件给客户端。 + disconnectMsg := fmt.Sprintf("upstream stream disconnected: %s", ev.err) if !c.Writer.Written() { logger.LegacyPrintf("service.gateway", "Upstream stream read error before any client output (account=%d), failing over: %v", account.ID, ev.err) - body, _ := json.Marshal(map[string]string{ - "error": fmt.Sprintf("upstream stream disconnected: %s", ev.err), + body, _ := json.Marshal(map[string]any{ + "type": "error", + "error": map[string]string{ + "type": "upstream_disconnected", + "message": disconnectMsg, + }, }) return nil, &UpstreamFailoverError{ StatusCode: http.StatusBadGateway, @@ -7055,7 +7077,7 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http RetryableOnSameAccount: true, } } - sendErrorEvent("stream_read_error") + sendErrorEvent("stream_read_error", disconnectMsg) return &streamingResult{usage: usage, firstTokenMs: firstTokenMs}, fmt.Errorf("stream read error: %w", ev.err) } line := ev.line @@ -7114,7 +7136,7 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http if s.rateLimitService != nil { s.rateLimitService.HandleStreamTimeout(ctx, account, originalModel) } - sendErrorEvent("stream_timeout") + sendErrorEvent("stream_timeout", fmt.Sprintf("upstream stream idle for %s", streamInterval)) return &streamingResult{usage: usage, firstTokenMs: firstTokenMs}, fmt.Errorf("stream data interval timeout") case <-keepaliveCh: diff --git a/backend/internal/service/gateway_streaming_test.go b/backend/internal/service/gateway_streaming_test.go index 389831fa..f3a52553 100644 --- a/backend/internal/service/gateway_streaming_test.go +++ b/backend/internal/service/gateway_streaming_test.go @@ -8,7 +8,6 @@ import ( "io" "net/http" "net/http/httptest" - "strings" "testing" "time" @@ -246,7 +245,15 @@ func TestHandleStreamingResponse_StreamReadErrorBeforeOutput_TriggersFailover(t require.True(t, errors.As(err, &failoverErr), "未输出过字节时 stream read error 必须包成 UpstreamFailoverError，期望: %v", err) require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode) require.True(t, failoverErr.RetryableOnSameAccount, "GOAWAY 类错误应允许同账号重试") - require.Contains(t, string(failoverErr.ResponseBody), "upstream stream disconnected") + + // ResponseBody 必须是 Anthropic 标准 error 格式： + // 1) ExtractUpstreamErrorMessage 能正确从 error.message 提取消息（被 handleFailoverExhausted / ops 日志依赖） + // 2) error.type 标记为 upstream_disconnected + extractedMsg := ExtractUpstreamErrorMessage(failoverErr.ResponseBody) + require.NotEmpty(t, extractedMsg, "ExtractUpstreamErrorMessage 必须从 ResponseBody 取到非空 message，否则 ops 日志会丢失诊断信息") + require.Contains(t, extractedMsg, "upstream stream disconnected") + require.Contains(t, string(failoverErr.ResponseBody), `"type":"error"`) + require.Contains(t, string(failoverErr.ResponseBody), `"upstream_disconnected"`) // 客户端应收不到任何 stream_read_error 事件，由 handler 层根据 failover 结果再决定 require.NotContains(t, rec.Body.String(), "stream_read_error") @@ -282,9 +289,11 @@ func TestHandleStreamingResponse_StreamReadErrorAfterOutput_PassesThrough(t *tes var failoverErr *UpstreamFailoverError require.False(t, errors.As(err, &failoverErr), "已经向客户端写过字节时不能再 failover") - // 客户端必须收到 stream_read_error 事件 + // 客户端必须收到 Anthropic 标准格式的 SSE error 事件，error.type=stream_read_error， + // error.message 含具体根因（让 SDK 能解析、UI 能显示具体错误） body := rec.Body.String() - require.True(t, - strings.Contains(body, "stream_read_error"), - "已开始流后必须发送 stream_read_error 事件给客户端，实际响应: %q", body) + require.Contains(t, body, "event: error\n", "必须按 Anthropic SSE 标准发送 error 事件帧") + require.Contains(t, body, `"type":"error"`, "data 必须含 type:error 顶层字段（Anthropic 标准）") + require.Contains(t, body, `"stream_read_error"`, "error.type 必须为 stream_read_error") + require.Contains(t, body, "upstream stream disconnected", "error.message 必须包含具体根因，Claude Code 等客户端才能显示有效错误文案") } From d78478e8668f0547f9639c812f2bb2641f80166f Mon Sep 17 00:00:00 2001 From: alfadb Date: Wed, 29 Apr 2026 15:44:54 +0800 Subject: [PATCH 3/3] fix(gateway): sanitize stream errors to avoid leaking infrastructure topology (*net.OpError).Error() concatenates Source/Addr fields, so the previous disconnectMsg surfaced internal source IP/port and upstream server address to clients via SSE error frames and UpstreamFailoverError.ResponseBody (reported by @Wei-Shaw on PR #2066). - Add sanitizeStreamError that maps known errors (io.ErrUnexpectedEOF, context.Canceled, syscall.ECONNRESET/EPIPE/ETIMEDOUT/...) to fixed descriptions and falls back to a generic placeholder, with an explicit *net.OpError branch that drops Source/Addr fields entirely. - Use sanitized message in client-facing disconnectMsg; full ev.err is still preserved in the existing operator log line for diagnosis. - Tests cover net.OpError redaction, the failover ResponseBody path, and every known sanitized error mapping. --- backend/internal/service/gateway_service.go | 50 +++++++++- .../service/gateway_streaming_test.go | 96 +++++++++++++++++++ 2 files changed, 145 insertions(+), 1 deletion(-) diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 4c4a9b82..aea0ba94 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -11,6 +11,7 @@ import ( "io" "log/slog" mathrand "math/rand" + "net" "net/http" "net/url" "os" @@ -20,6 +21,7 @@ import ( "strconv" "strings" "sync/atomic" + "syscall" "time" "github.com/Wei-Shaw/sub2api/internal/config" @@ -6434,6 +6436,49 @@ func (s *GatewayService) shouldFailoverOn400(respBody []byte) bool { return false } +// sanitizeStreamError 返回不含网络地址的客户端可见错误描述。 +// 默认 (*net.OpError).Error() 会拼接 Source/Addr 字段，泄露内部 IP/端口与上游 +// 服务器地址（例如 "read tcp 10.0.0.1:54321->52.1.2.3:443: read: connection +// reset by peer"）。该函数只保留可识别的错误类别，原始 err 仍在调用点写入日志。 +func sanitizeStreamError(err error) string { + if err == nil { + return "" + } + switch { + case errors.Is(err, io.ErrUnexpectedEOF): + return "unexpected EOF" + case errors.Is(err, io.EOF): + return "EOF" + case errors.Is(err, context.Canceled): + return "canceled" + case errors.Is(err, context.DeadlineExceeded): + return "deadline exceeded" + case errors.Is(err, syscall.ECONNRESET): + return "connection reset by peer" + case errors.Is(err, syscall.ECONNABORTED): + return "connection aborted" + case errors.Is(err, syscall.ETIMEDOUT): + return "connection timed out" + case errors.Is(err, syscall.EPIPE): + return "broken pipe" + case errors.Is(err, syscall.ECONNREFUSED): + return "connection refused" + } + var netErr *net.OpError + if errors.As(err, &netErr) { + if netErr.Timeout() { + if netErr.Op != "" { + return netErr.Op + " timeout" + } + return "i/o timeout" + } + if netErr.Op != "" { + return netErr.Op + " network error" + } + } + return "upstream connection error" +} + // ExtractUpstreamErrorMessage 从上游响应体中提取错误消息 // 支持 Claude 风格的错误格式：{"type":"error","error":{"type":"...","message":"..."}} func ExtractUpstreamErrorMessage(body []byte) string { @@ -7061,7 +7106,10 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http // 上游中途读错误（unexpected EOF / connection reset 等，常见于 HTTP/2 GOAWAY）： // 若尚未向客户端写过任何字节，包成 UpstreamFailoverError 让 handler 层走 failover/重试。 // 已经开始写流时 SSE 协议无 resume，只能透传错误事件给客户端。 - disconnectMsg := fmt.Sprintf("upstream stream disconnected: %s", ev.err) + // 注意:面向客户端的 disconnectMsg 必须用 sanitizeStreamError 剥离地址, + // 默认 *net.OpError 的 Error() 会泄露内部 IP/端口和上游地址。完整 ev.err + // 仅在下方 LegacyPrintf 内部日志中保留供运维诊断。 + disconnectMsg := "upstream stream disconnected: " + sanitizeStreamError(ev.err) if !c.Writer.Written() { logger.LegacyPrintf("service.gateway", "Upstream stream read error before any client output (account=%d), failing over: %v", account.ID, ev.err) body, _ := json.Marshal(map[string]any{ diff --git a/backend/internal/service/gateway_streaming_test.go b/backend/internal/service/gateway_streaming_test.go index f3a52553..ef09a882 100644 --- a/backend/internal/service/gateway_streaming_test.go +++ b/backend/internal/service/gateway_streaming_test.go @@ -6,8 +6,10 @@ import ( "context" "errors" "io" + "net" "net/http" "net/http/httptest" + "syscall" "testing" "time" @@ -297,3 +299,97 @@ func TestHandleStreamingResponse_StreamReadErrorAfterOutput_PassesThrough(t *tes require.Contains(t, body, `"stream_read_error"`, "error.type 必须为 stream_read_error") require.Contains(t, body, "upstream stream disconnected", "error.message 必须包含具体根因，Claude Code 等客户端才能显示有效错误文案") } + +// 默认 (*net.OpError).Error() 会拼接 Source/Addr 字段，泄露内部 IP/端口与上游 +// 服务器地址。sanitizeStreamError 必须剥离这些信息，避免基础设施拓扑通过 +// failover ResponseBody 或 SSE error 帧返回给客户端。 +func TestSanitizeStreamError_StripsNetworkAddresses(t *testing.T) { + src, err := net.ResolveTCPAddr("tcp", "10.0.0.1:54321") + require.NoError(t, err) + dst, err := net.ResolveTCPAddr("tcp", "52.1.2.3:443") + require.NoError(t, err) + + raw := &net.OpError{ + Op: "read", + Net: "tcp", + Source: src, + Addr: dst, + Err: syscall.ECONNRESET, + } + + // 前置：原始 Error() 确实包含会泄露的字段（避免测试在 Go 行为变化时静默通过） + require.Contains(t, raw.Error(), "10.0.0.1") + require.Contains(t, raw.Error(), "52.1.2.3") + + got := sanitizeStreamError(raw) + require.NotContains(t, got, "10.0.0.1", "不得泄露内部源 IP") + require.NotContains(t, got, "54321", "不得泄露源端口") + require.NotContains(t, got, "52.1.2.3", "不得泄露上游目标 IP") + require.NotContains(t, got, "443", "不得泄露上游端口") + require.Equal(t, "connection reset by peer", got) +} + +func TestSanitizeStreamError_KnownErrors(t *testing.T) { + cases := []struct { + name string + err error + want string + }{ + {"unexpected EOF", io.ErrUnexpectedEOF, "unexpected EOF"}, + {"EOF", io.EOF, "EOF"}, + {"context canceled", context.Canceled, "canceled"}, + {"deadline exceeded", context.DeadlineExceeded, "deadline exceeded"}, + {"ECONNRESET 直接", syscall.ECONNRESET, "connection reset by peer"}, + {"EPIPE", syscall.EPIPE, "broken pipe"}, + {"ETIMEDOUT", syscall.ETIMEDOUT, "connection timed out"}, + {"未识别错误兜底", errors.New("weird internal error"), "upstream connection error"}, + {"nil 返回空串", nil, ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, sanitizeStreamError(tc.err)) + }) + } +} + +// failover ResponseBody 必须用 sanitize 过的消息，避免泄露给客户端 / 写入 ops 日志 +// 时携带内部地址信息。 +func TestHandleStreamingResponse_FailoverBodyDoesNotLeakAddresses(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := newMinimalGatewayService() + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + + src, _ := net.ResolveTCPAddr("tcp", "10.0.0.1:54321") + dst, _ := net.ResolveTCPAddr("tcp", "52.1.2.3:443") + netErr := &net.OpError{ + Op: "read", + Net: "tcp", + Source: src, + Addr: dst, + Err: syscall.ECONNRESET, + } + + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: &streamReadCloser{err: netErr}, + } + + _, err := svc.handleStreamingResponse(context.Background(), resp, c, &Account{ID: 1}, time.Now(), "model", "model", false) + require.Error(t, err) + + var failoverErr *UpstreamFailoverError + require.True(t, errors.As(err, &failoverErr)) + + body := string(failoverErr.ResponseBody) + require.NotContains(t, body, "10.0.0.1", "failover ResponseBody 不得泄露内部源 IP") + require.NotContains(t, body, "54321") + require.NotContains(t, body, "52.1.2.3", "failover ResponseBody 不得泄露上游 IP") + require.NotContains(t, body, "443") + // 仍然包含可诊断的根因 + require.Contains(t, body, "connection reset by peer") + require.Contains(t, body, "upstream stream disconnected") +}