seewe/sub2api
1
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-14 12:02:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(responses): emit protocol-correct stream errors once

Browse Source
This commit is contained in:
lizhaoxinlizhaoxin2-lab
2026-09-18 12:38:05 +08:00
parent efe9aab1e4
commit cf10d6d01d
3 changed files with 80 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/handler/openai_gateway_handler_test.go
+14
View File
@@ -381,6 +381,20 @@ func TestOpenAIEnsureForwardErrorResponse_AfterDeltaAppendsSingleValidResponseFa
require.Equal(t, 1, errorEvents)
}
func TestOpenAIEnsureForwardErrorResponse_PreservesCommittedStreamError(t *testing.T) {
gin.SetMode(gin.TestMode)
w := httptest.NewRecorder()
c, _ := gin.CreateTestContext(w)
c.Request = httptest.NewRequest(http.MethodPost, EndpointResponses, nil)
body := "event: error\ndata: {\"type\":\"error\",\"code\":\"upstream_stream_read_error\",\"message\":\"Upstream response stream was interrupted\",\"param\":null,\"sequence_number\":0}\n\n"
_, err := c.Writer.WriteString(body)
require.NoError(t, err)
service.MarkResponseCommitted(c)
h := &OpenAIGatewayHandler{}
require.False(t, h.ensureForwardErrorResponse(c, true))
require.Equal(t, body, w.Body.String(), "preserve the service error without a second terminal event")
}
func TestOpenAIEnsureForwardErrorResponse_CompactKeepaliveOnlyWritesResponseFailed(t *testing.T) {
gin.SetMode(gin.TestMode)
w := httptest.NewRecorder()
backend/internal/service/openai_gateway_response_handling.go
+11 -6
View File
@@ -314,17 +314,19 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context.
eventStartsTTFTOutput = false
eventShouldFlush = false
}
sendErrorEvent := func(reason string) {
sendErrorEvent := func(code, message string) {
if errorEventSent || clientDisconnected || failureDelivered {
return
}
errorEventSent = true
payload := `{"type":"error","sequence_number":0,"error":{"type":"upstream_error","message":` + strconv.Quote(reason) + `,"code":` + strconv.Quote(reason) + `}}`
// Responses error events use top-level code/message/param fields. A nested
// Chat Completions error envelope loses the classification in strict clients.
payload := `{"type":"error","sequence_number":0,"code":` + strconv.Quote(code) + `,"message":` + strconv.Quote(message) + `,"param":null}`
if err := flushBuffered(); err != nil {
clientDisconnected = true
return
}
if _, err := writePendingString("data: " + payload + "\n\n"); err != nil {
if _, err := writePendingString("event: error\ndata: " + payload + "\n\n"); err != nil {
clientDisconnected = true
return
}
@@ -334,6 +336,8 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context.
}
clientOutputStarted = true
lastDownstreamWriteAt = time.Now()
// The handler must not append its generic failure after this error event.
MarkResponseCommitted(c)
}
needModelReplace := originalModel != mappedModel
@@ -450,7 +454,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context.
}
if errors.Is(scanErr, bufio.ErrTooLong) {
logger.LegacyPrintf("service.openai_gateway", "SSE line too long: account=%d max_size=%d error=%v", account.ID, maxLineSize, scanErr)
sendErrorEvent("response_too_large")
sendErrorEvent("response_too_large", "Upstream response exceeded the size limit")
return resultWithUsage(), scanErr, true
}
if !openAIStreamClientOutputStarted(c, clientOutputStarted) && !eventShouldFlush {
@@ -465,7 +469,8 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context.
return resultWithUsage(), fmt.Errorf("stream usage incomplete after disconnect: %w", scanErr), true
}
s.recordOpenAIProxyStreamDisconnect(account, scanErr, upstreamRequestID)
sendErrorEvent("stream_read_error")
code, message := classifyOpenAIUpstreamStreamReadError(scanErr)
sendErrorEvent(code, message)
return resultWithUsage(), fmt.Errorf("stream read error: %w", scanErr), true
}
processSSELine := func(line string, queueDrained bool) {
@@ -914,7 +919,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context.
return resultWithUsage(), grokStreamIdleFailoverError(account, streamInterval)
}
}
sendErrorEvent("stream_timeout")
sendErrorEvent("stream_timeout", "Upstream response stream timed out")
return resultWithUsage(), fmt.Errorf("stream data interval timeout")
case <-firstOutputCh:
backend/internal/service/openai_gateway_service_test.go
+55
View File
@@ -1610,6 +1610,10 @@ func TestOpenAIStreamingTimeout(t *testing.T) {
if !strings.Contains(rec.Body.String(), "\"type\":\"error\"") || !strings.Contains(rec.Body.String(), "stream_timeout") {
t.Fatalf("expected OpenAI-compatible error SSE event, got %q", rec.Body.String())
}
payload := strings.TrimSpace(strings.TrimPrefix(rec.Body.String(), "event: error\ndata: "))
require.Equal(t, "stream_timeout", gjson.Get(payload, "code").String())
require.False(t, gjson.Get(payload, "error").Exists())
require.True(t, IsResponseCommitted(c))
}
func TestOpenAIStreamingContextCanceledReturnsIncompleteErrorWithoutInjectingErrorEvent(t *testing.T) {
@@ -1674,6 +1678,53 @@ func TestOpenAIStreamingReadErrorBeforeOutputReturnsFailover(t *testing.T) {
require.Empty(t, rec.Body.String())
}
func TestOpenAIStreamingReadErrorAfterOutputUsesResponsesErrorSchema(t *testing.T) {
gin.SetMode(gin.TestMode)
for _, tc := range []struct {
name string
cause error
code string
}{
{"reset", errors.New("read tcp 192.0.2.1:1234->192.0.2.2:443: connection reset by peer"), OpenAIUpstreamStreamReadErrorCode},
{"http2", errors.New("stream error: stream ID 3; INTERNAL_ERROR; received from peer"), OpenAIUpstreamHTTP2StreamErrorCode},
} {
t.Run(tc.name, func(t *testing.T) {
svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{MaxLineSize: defaultMaxLineSize}}}
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil)
resp := &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: &openAIStreamReadThenErrorCloser{
reader: strings.NewReader("event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n"),
err: tc.cause,
}}
_, err := svc.handleStreamingResponse(c.Request.Context(), resp, c, &Account{ID: 1, Platform: PlatformOpenAI}, time.Now(), "model", "model")
require.ErrorIs(t, err, tc.cause)
body := rec.Body.String()
require.Contains(t, body, "partial")
require.NotContains(t, body, "192.0.2.")
require.NotContains(t, body, "stream ID")
require.NotContains(t, body, "response.completed")
require.NotContains(t, body, "[DONE]")
require.Equal(t, 1, strings.Count(body, "event: error\n"))
require.True(t, IsResponseCommitted(c), "the handler must not append another failure")
var events []gjson.Result
for _, line := range strings.Split(body, "\n") {
if strings.HasPrefix(line, "data: ") {
event := gjson.Parse(strings.TrimPrefix(line, "data: "))
if event.Get("type").String() == "error" {
events = append(events, event)
}
}
}
require.Len(t, events, 1)
require.Equal(t, tc.code, events[0].Get("code").String())
require.NotEmpty(t, events[0].Get("message").String())
require.False(t, events[0].Get("error").Exists(), "Responses errors have top-level fields")
require.True(t, events[0].Get("param").Exists())
})
}
}
func TestOpenAIStreamingPostOutputDisconnectQuarantinesSharedProxyWithoutSameStreamFailover(t *testing.T) {
gin.SetMode(gin.TestMode)
proxyID := int64(4698)
@@ -2791,6 +2842,10 @@ func TestOpenAIStreamingTooLong(t *testing.T) {
if !strings.Contains(rec.Body.String(), "\"type\":\"error\"") || !strings.Contains(rec.Body.String(), "response_too_large") {
t.Fatalf("expected OpenAI-compatible error SSE event, got %q", rec.Body.String())
}
payload := strings.TrimSpace(strings.TrimPrefix(rec.Body.String(), "event: error\ndata: "))
require.Equal(t, "response_too_large", gjson.Get(payload, "code").String())
require.False(t, gjson.Get(payload, "error").Exists())
require.True(t, IsResponseCommitted(c))
}
func TestOpenAINonStreamingContentTypePassThrough(t *testing.T) {