mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-08-29 03:09:50 +08:00
Merge pull request #6310 from fengshao1227/fix/nonstreaming-terminal-failure-failover-v2
fix(openai): 非流式路径按流式同一套判定处理 HTTP 200 终止失败事件
This commit is contained in:
Wesley Liddick
@@ -1695,6 +1695,65 @@ func (s *OpenAIGatewayService) newOpenAIStreamFailoverError(
|
||||
return failoverErr
|
||||
}
|
||||
|
||||
// nonStreamingTerminalFailureFailover applies the streaming path's terminal-event
|
||||
// verdict to a stream=false request whose upstream answered with SSE anyway
|
||||
// (other sub2api instances and several OpenAI-compatible upstreams do this).
|
||||
//
|
||||
// Both handleSSEToJSON and handlePassthroughSSEToJSON collapsed every terminal
|
||||
// `response.failed` / `error` frame into writeOpenAINonStreamingProtocolError, a
|
||||
// fixed 502. The streaming readers sitting a few hundred lines away classify the
|
||||
// very same frame with openAIStreamFailedEventShouldFailover /
|
||||
// openAIStreamErrorEventShouldFailover and return an UpstreamFailoverError, so an
|
||||
// upstream capacity error switched accounts when stream=true and was handed to the
|
||||
// caller verbatim when stream=false — with other schedulable accounts still in the
|
||||
// pool. Same event, same upstream, opposite outcome, decided only by a request flag
|
||||
// the upstream never saw.
|
||||
//
|
||||
// Dispatching on terminalType keeps the two verdicts distinct exactly as the
|
||||
// streaming readers do: a bare `error` frame goes through the conservative
|
||||
// classifier that fails over only on positively transient markers, while
|
||||
// `response.failed` uses the fuller one.
|
||||
//
|
||||
// Replay is safe here because the body was fully buffered by
|
||||
// ReadUpstreamResponseBody and this runs before any semantic byte is written.
|
||||
// Whether a failover actually happens stays the handler's call:
|
||||
// openAIForwardMayFailover compares OpenAICompactKeepaliveAdjustedWrittenSize
|
||||
// against its pre-Forward snapshot, so a request that already emitted output is
|
||||
// refused there (#3887). This function only declines to propose failover once the
|
||||
// service has explicitly committed a response, and never re-implements the
|
||||
// keepalive accounting — a second copy of that rule would drift from the handler's.
|
||||
//
|
||||
// A nil account means there is nothing to fail over from: newOpenAIStreamFailoverError
|
||||
// records ops attribution and account health against it, so those callers keep the
|
||||
// protocol-error behaviour.
|
||||
func (s *OpenAIGatewayService) nonStreamingTerminalFailureFailover(
|
||||
c *gin.Context,
|
||||
resp *http.Response,
|
||||
account *Account,
|
||||
passthrough bool,
|
||||
terminalType string,
|
||||
payload []byte,
|
||||
message string,
|
||||
) *UpstreamFailoverError {
|
||||
if account == nil || IsResponseCommitted(c) {
|
||||
return nil
|
||||
}
|
||||
shouldFailover := openAIStreamFailedEventShouldFailover(payload, message)
|
||||
if terminalType == "error" {
|
||||
shouldFailover = openAIStreamErrorEventShouldFailover(payload, message)
|
||||
}
|
||||
if !shouldFailover {
|
||||
return nil
|
||||
}
|
||||
var headers http.Header
|
||||
upstreamRequestID := ""
|
||||
if resp != nil {
|
||||
headers = resp.Header
|
||||
upstreamRequestID = strings.TrimSpace(resp.Header.Get("x-request-id"))
|
||||
}
|
||||
return s.newOpenAIStreamFailoverError(c, account, passthrough, upstreamRequestID, payload, message, headers)
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) handleStreamingResponsePassthrough(
|
||||
ctx context.Context,
|
||||
resp *http.Response,
|
||||
@@ -2171,6 +2230,9 @@ func (s *OpenAIGatewayService) handlePassthroughSSEToJSON(resp *http.Response, c
|
||||
if compactErr := newOpenAICompactFallbackSignal(c, terminalPayload, msg); compactErr != nil {
|
||||
return nil, compactErr
|
||||
}
|
||||
if failoverErr := s.nonStreamingTerminalFailureFailover(c, resp, account, true, terminalType, terminalPayload, msg); failoverErr != nil {
|
||||
return nil, failoverErr
|
||||
}
|
||||
return nil, s.writeOpenAINonStreamingProtocolError(resp, c, msg)
|
||||
}
|
||||
finalResponse, ok := extractCodexFinalResponse(bodyText)
|
||||
|
||||
@@ -1684,6 +1684,9 @@ func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Conte
|
||||
if compactErr := newOpenAICompactFallbackSignal(c, terminalPayload, msg); compactErr != nil {
|
||||
return nil, compactErr
|
||||
}
|
||||
if failoverErr := s.nonStreamingTerminalFailureFailover(c, resp, account, false, terminalType, terminalPayload, msg); failoverErr != nil {
|
||||
return nil, failoverErr
|
||||
}
|
||||
return nil, s.writeOpenAINonStreamingProtocolError(resp, c, msg)
|
||||
}
|
||||
finalResponse, ok := extractCodexFinalResponse(bodyText)
|
||||
|
||||
@@ -3806,7 +3806,10 @@ func TestHandleSSEToJSON_NoFinalResponseKeepsSSEBody(t *testing.T) {
|
||||
require.Contains(t, rec.Body.String(), `data: {"type":"response.in_progress"`)
|
||||
}
|
||||
|
||||
func TestHandleSSEToJSON_ResponseFailedReturnsProtocolError(t *testing.T) {
|
||||
// 无账号时没有可换的对象：newOpenAIStreamFailoverError 要拿 account 记录 ops 归属与
|
||||
// 账号健康，故这一支保持原有的协议错误行为。带真实账号的同一报文改为换号，
|
||||
// 由 TestNonStreamingSSEToJSON_UnclassifiedFailedEventFailsOver 钉死（issue #5281）。
|
||||
func TestHandleSSEToJSON_ResponseFailedWithoutAccountReturnsProtocolError(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
rec := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(rec)
|
||||
|
||||
@@ -0,0 +1,258 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/config"
|
||||
)
|
||||
|
||||
// issue #5281：stream=false 时上游仍可能回 SSE（其他 sub2api 实例、部分 OpenAI 兼容
|
||||
// 上游），容量/限流错误经 HTTP 200 的终止事件回传。handleSSEToJSON 与
|
||||
// handlePassthroughSSEToJSON 把所有终止事件塞进固定 502，而几百行外的流式读取器对
|
||||
// 同一帧走 openAIStreamFailedEventShouldFailover / openAIStreamErrorEventShouldFailover
|
||||
// 判定并换号——同一个上游、同一个事件，只因请求上的 stream 标志而结果相反。
|
||||
|
||||
func newNonStreamingFailoverContext(t *testing.T) (*gin.Context, *httptest.ResponseRecorder) {
|
||||
t.Helper()
|
||||
gin.SetMode(gin.TestMode)
|
||||
rec := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(rec)
|
||||
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil)
|
||||
return c, rec
|
||||
}
|
||||
|
||||
func newNonStreamingFailoverService() *OpenAIGatewayService {
|
||||
return &OpenAIGatewayService{cfg: &config.Config{}}
|
||||
}
|
||||
|
||||
func newNonStreamingFailoverAccount() *Account {
|
||||
return &Account{
|
||||
ID: 1,
|
||||
Platform: PlatformOpenAI,
|
||||
Type: AccountTypeAPIKey,
|
||||
Name: "pool-account",
|
||||
Credentials: map[string]any{
|
||||
"pool_mode": true,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func newNonStreamingSSEResponse() *http.Response {
|
||||
return &http.Response{
|
||||
StatusCode: http.StatusOK,
|
||||
Header: http.Header{
|
||||
"Content-Type": []string{"text/event-stream"},
|
||||
"X-Request-Id": []string{"rid-nonstreaming-failed"},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func sseTerminalBody(eventType, data string) []byte {
|
||||
return []byte(strings.Join([]string{
|
||||
"event: " + eventType,
|
||||
"data: " + data,
|
||||
"",
|
||||
"data: [DONE]",
|
||||
}, "\n"))
|
||||
}
|
||||
|
||||
// 主复现：issue 报告的容量错误，与流式兄弟用例
|
||||
// TestOpenAIStreamingResponseFailedBeforeOutputCapacityErrorReturnsFailover 逐项对齐。
|
||||
func TestNonStreamingSSEToJSON_CapacityFailedEventFailsOver(t *testing.T) {
|
||||
c, rec := newNonStreamingFailoverContext(t)
|
||||
svc := newNonStreamingFailoverService()
|
||||
body := sseTerminalBody("response.failed",
|
||||
`{"type":"response.failed","error":{"message":"Selected model is at capacity. Please try a different model.","type":"invalid_request_error"}}`)
|
||||
|
||||
result, err := svc.handleSSEToJSON(newNonStreamingSSEResponse(), c, newNonStreamingFailoverAccount(), body, "model", "model")
|
||||
|
||||
require.Nil(t, result)
|
||||
var failoverErr *UpstreamFailoverError
|
||||
require.ErrorAs(t, err, &failoverErr)
|
||||
require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode)
|
||||
// 容量降载是请求级信号，先在同账号有界重试——与流式路径同一套策略。
|
||||
require.True(t, failoverErr.RetryableOnSameAccount)
|
||||
require.Contains(t, string(failoverErr.ResponseBody), "Selected model is at capacity")
|
||||
// 换号的前提：一个字节都没写出去。
|
||||
require.False(t, c.Writer.Written())
|
||||
require.Empty(t, rec.Body.String())
|
||||
}
|
||||
|
||||
// 行为翻转点：未被分类为不可重试的泛化 response.failed，此前回 502，现在换号。
|
||||
// 与流式路径对齐的结果，显式钉住以免被当成回归。
|
||||
func TestNonStreamingSSEToJSON_UnclassifiedFailedEventFailsOver(t *testing.T) {
|
||||
c, rec := newNonStreamingFailoverContext(t)
|
||||
svc := newNonStreamingFailoverService()
|
||||
payload := []byte(`{"type":"response.failed","error":{"message":"upstream rejected request"}}`)
|
||||
body := sseTerminalBody("response.failed", string(payload))
|
||||
|
||||
// 前提：流式分类器对同一帧的裁决就是「换号」。翻转不是新政策，是补齐。
|
||||
require.True(t, openAIStreamFailedEventShouldFailover(payload, "upstream rejected request"))
|
||||
|
||||
result, err := svc.handleSSEToJSON(newNonStreamingSSEResponse(), c, newNonStreamingFailoverAccount(), body, "model", "model")
|
||||
|
||||
require.Nil(t, result)
|
||||
var failoverErr *UpstreamFailoverError
|
||||
require.ErrorAs(t, err, &failoverErr)
|
||||
require.False(t, c.Writer.Written())
|
||||
require.Empty(t, rec.Body.String())
|
||||
}
|
||||
|
||||
// 「response.failed 必须回写协议错误」这一契约没有丢：明确不可重试的错误仍写 502。
|
||||
func TestNonStreamingSSEToJSON_NonRetryableFailedEventStillWritesProtocolError(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
data string
|
||||
wantMsg string
|
||||
}{
|
||||
{
|
||||
name: "invalid_request",
|
||||
data: `{"type":"response.failed","error":{"type":"invalid_request_error","code":"invalid_request","message":"unknown parameter foo"}}`,
|
||||
wantMsg: "unknown parameter foo",
|
||||
},
|
||||
{
|
||||
name: "context_window",
|
||||
data: `{"type":"response.failed","response":{"id":"resp_failed","status":"failed","output":[],"error":{"code":"upstream_error","message":"input exceeds the context window"}}}`,
|
||||
wantMsg: "input exceeds the context window",
|
||||
},
|
||||
{
|
||||
name: "content_policy",
|
||||
data: `{"type":"response.failed","error":{"type":"content_policy_violation","message":"blocked by our content policy"}}`,
|
||||
wantMsg: "blocked by our content policy",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
c, rec := newNonStreamingFailoverContext(t)
|
||||
svc := newNonStreamingFailoverService()
|
||||
|
||||
result, err := svc.handleSSEToJSON(newNonStreamingSSEResponse(), c,
|
||||
newNonStreamingFailoverAccount(), sseTerminalBody("response.failed", tc.data), "model", "model")
|
||||
|
||||
require.Nil(t, result)
|
||||
require.Error(t, err)
|
||||
var failoverErr *UpstreamFailoverError
|
||||
require.False(t, errors.As(err, &failoverErr), "不可重试的上游错误不得换号")
|
||||
require.Equal(t, http.StatusBadGateway, rec.Code)
|
||||
require.Contains(t, rec.Body.String(), tc.wantMsg)
|
||||
require.Contains(t, rec.Header().Get("Content-Type"), "application/json")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 裸 error 帧走的是更保守的那个分类器：只有正向识别为瞬时才换号。
|
||||
// 这条用例证明两种终止事件确实被分派到了各自的判定，而不是共用一个。
|
||||
func TestNonStreamingSSEToJSON_BareErrorEventUsesConservativeClassifier(t *testing.T) {
|
||||
t.Run("non_transient_stays_protocol_error", func(t *testing.T) {
|
||||
c, rec := newNonStreamingFailoverContext(t)
|
||||
svc := newNonStreamingFailoverService()
|
||||
data := `{"type":"error","error":{"message":"upstream rejected request"}}`
|
||||
|
||||
// 同一条文案：failed 判换号，error 判不换号——差异就在分派。
|
||||
require.True(t, openAIStreamFailedEventShouldFailover([]byte(data), "upstream rejected request"))
|
||||
require.False(t, openAIStreamErrorEventShouldFailover([]byte(data), "upstream rejected request"))
|
||||
|
||||
result, err := svc.handleSSEToJSON(newNonStreamingSSEResponse(), c,
|
||||
newNonStreamingFailoverAccount(), sseTerminalBody("error", data), "model", "model")
|
||||
|
||||
require.Nil(t, result)
|
||||
require.Error(t, err)
|
||||
var failoverErr *UpstreamFailoverError
|
||||
require.False(t, errors.As(err, &failoverErr))
|
||||
require.Equal(t, http.StatusBadGateway, rec.Code)
|
||||
})
|
||||
|
||||
t.Run("transient_fails_over", func(t *testing.T) {
|
||||
c, _ := newNonStreamingFailoverContext(t)
|
||||
svc := newNonStreamingFailoverService()
|
||||
data := `{"type":"error","error":{"message":"Temporary upstream failure, please retry"}}`
|
||||
|
||||
result, err := svc.handleSSEToJSON(newNonStreamingSSEResponse(), c,
|
||||
newNonStreamingFailoverAccount(), sseTerminalBody("error", data), "model", "model")
|
||||
|
||||
require.Nil(t, result)
|
||||
var failoverErr *UpstreamFailoverError
|
||||
require.ErrorAs(t, err, &failoverErr)
|
||||
require.False(t, c.Writer.Written())
|
||||
})
|
||||
}
|
||||
|
||||
// 透传路径必须与合成路径同步修复，否则又造出一处新的不对称。
|
||||
func TestNonStreamingPassthroughSSEToJSON_CapacityFailedEventFailsOver(t *testing.T) {
|
||||
c, rec := newNonStreamingFailoverContext(t)
|
||||
svc := newNonStreamingFailoverService()
|
||||
body := sseTerminalBody("response.failed",
|
||||
`{"type":"response.failed","error":{"message":"Selected model is at capacity. Please try a different model.","type":"invalid_request_error"}}`)
|
||||
|
||||
result, err := svc.handlePassthroughSSEToJSON(newNonStreamingSSEResponse(), c, newNonStreamingFailoverAccount(), body, "model", "model")
|
||||
|
||||
require.Nil(t, result)
|
||||
var failoverErr *UpstreamFailoverError
|
||||
require.ErrorAs(t, err, &failoverErr)
|
||||
require.Contains(t, string(failoverErr.ResponseBody), "Selected model is at capacity")
|
||||
require.False(t, c.Writer.Written())
|
||||
require.Empty(t, rec.Body.String())
|
||||
}
|
||||
|
||||
// 不变式：非流式的裁决必须与流式分类器逐项一致。任何一边以后改了判定，这条会红。
|
||||
func TestNonStreamingSSEToJSON_MatchesStreamingClassifierVerdict(t *testing.T) {
|
||||
payloads := []string{
|
||||
`{"type":"response.failed","error":{"message":"Selected model is at capacity. Please try a different model.","type":"invalid_request_error"}}`,
|
||||
`{"type":"response.failed","error":{"message":"upstream rejected request"}}`,
|
||||
`{"type":"response.failed","error":{"type":"invalid_request_error","code":"invalid_request","message":"unknown parameter foo"}}`,
|
||||
`{"type":"response.failed","response":{"id":"r","status":"failed","output":[],"error":{"code":"upstream_error","message":"input exceeds the context window"}}}`,
|
||||
`{"type":"response.failed","error":{"type":"content_policy_violation","message":"blocked by our content policy"}}`,
|
||||
}
|
||||
|
||||
for _, data := range payloads {
|
||||
t.Run(data[:min(len(data), 60)], func(t *testing.T) {
|
||||
payload := []byte(data)
|
||||
want := openAIStreamFailedEventShouldFailover(payload, extractOpenAISSEErrorMessage(payload))
|
||||
|
||||
c, _ := newNonStreamingFailoverContext(t)
|
||||
svc := newNonStreamingFailoverService()
|
||||
_, err := svc.handleSSEToJSON(newNonStreamingSSEResponse(), c,
|
||||
newNonStreamingFailoverAccount(), sseTerminalBody("response.failed", data), "model", "model")
|
||||
|
||||
var failoverErr *UpstreamFailoverError
|
||||
require.Equal(t, want, errors.As(err, &failoverErr),
|
||||
"非流式裁决与流式分类器不一致：%s", data)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 服务侧已显式提交响应后不得再提议换号；能否真正换号仍由 handler 的
|
||||
// openAIForwardMayFailover 用 keepalive 调整后的写出量仲裁（#3887），此处不重复实现。
|
||||
func TestNonStreamingSSEToJSON_CommittedResponseKeepsProtocolError(t *testing.T) {
|
||||
c, rec := newNonStreamingFailoverContext(t)
|
||||
svc := newNonStreamingFailoverService()
|
||||
MarkResponseCommitted(c)
|
||||
body := sseTerminalBody("response.failed",
|
||||
`{"type":"response.failed","error":{"message":"Selected model is at capacity. Please try a different model.","type":"invalid_request_error"}}`)
|
||||
|
||||
result, err := svc.handleSSEToJSON(newNonStreamingSSEResponse(), c, newNonStreamingFailoverAccount(), body, "model", "model")
|
||||
|
||||
require.Nil(t, result)
|
||||
require.Error(t, err)
|
||||
var failoverErr *UpstreamFailoverError
|
||||
require.False(t, errors.As(err, &failoverErr))
|
||||
require.Equal(t, http.StatusBadGateway, rec.Code)
|
||||
}
|
||||
|
||||
func TestNonStreamingTerminalFailureFailover_NilAccountProposesNothing(t *testing.T) {
|
||||
c, _ := newNonStreamingFailoverContext(t)
|
||||
svc := newNonStreamingFailoverService()
|
||||
payload := []byte(`{"type":"response.failed","error":{"message":"Selected model is at capacity. Please try a different model."}}`)
|
||||
|
||||
require.Nil(t, svc.nonStreamingTerminalFailureFailover(
|
||||
c, newNonStreamingSSEResponse(), nil, false, "response.failed", payload,
|
||||
"Selected model is at capacity. Please try a different model."))
|
||||
}
|
||||
Reference in New Issue
Block a user