diff --git a/backend/internal/service/openai_images_responses.go b/backend/internal/service/openai_images_responses.go index 7924de4fc..4656d22c4 100644 --- a/backend/internal/service/openai_images_responses.go +++ b/backend/internal/service/openai_images_responses.go @@ -1239,6 +1239,9 @@ func (s *OpenAIGatewayService) handleOpenAIImagesOAuthNonStreamingResponse( ) (OpenAIUsage, int, []string, error) { body, err := ReadUpstreamResponseBody(resp.Body, s.cfg, c, openAITooLargeError) if err != nil { + if shouldClassifyOpenAIUpstreamStreamReadError(err, c.Request.Context()) { + err = newOpenAIUpstreamStreamReadError(err) + } return OpenAIUsage{}, 0, nil, err } @@ -1532,7 +1535,9 @@ func (s *OpenAIGatewayService) handleOpenAIImagesOAuthStreamingResponse( } else if done { return usage, imageCount, imageOutputSizes, firstTokenMs, nil } - s.tryWriteOpenAIImagesStreamEvent(c, flusher, &clientDisconnected, &lastDownstreamWriteAt, "error", buildOpenAIImagesStreamErrorBody(err.Error())) + if shouldClassifyOpenAIUpstreamStreamReadError(err, c.Request.Context()) { + err = newOpenAIUpstreamStreamReadError(err) + } return usage, imageCount, imageOutputSizes, firstTokenMs, err } } @@ -1625,7 +1630,9 @@ func (s *OpenAIGatewayService) handleOpenAIImagesOAuthStreamingResponse( } else if done { return usage, imageCount, imageOutputSizes, firstTokenMs, nil } - s.tryWriteOpenAIImagesStreamEvent(c, flusher, &clientDisconnected, &lastDownstreamWriteAt, "error", buildOpenAIImagesStreamErrorBody(ev.err.Error())) + if shouldClassifyOpenAIUpstreamStreamReadError(ev.err, c.Request.Context()) { + ev.err = newOpenAIUpstreamStreamReadError(ev.err) + } return usage, imageCount, imageOutputSizes, firstTokenMs, ev.err } done, processErr := processLine(ev.line) @@ -1846,13 +1853,41 @@ func (s *OpenAIGatewayService) handleOpenAIImagesOAuthResponseError( writerSizeBeforeResponse int, err error, ) error { + responseWritten := c != nil && c.Writer != nil && OpenAIImagesJSONKeepaliveAdjustedWrittenSize(c) != writerSizeBeforeResponse + if code, message, ok := OpenAIUpstreamStreamReadErrorDetails(err); ok { + // A body transport failure after a successful HTTP status is retryable only + // until real image output has reached the client. Keep the upstream headers + // and request ID available to the failover/error passthrough path. + headers := http.Header(nil) + requestID := "" + statusCode := http.StatusBadGateway + if resp != nil { + headers = resp.Header.Clone() + requestID = strings.TrimSpace(resp.Header.Get("x-request-id")) + } + kind := "failover" + if responseWritten { + kind = "retry_exhausted_failover" + } + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, + UpstreamStatusCode: statusCode, UpstreamRequestID: requestID, UpstreamURL: upstreamURL, + Kind: kind, Message: message, + }) + if responseWritten { + return err + } + responseBody := []byte(fmt.Sprintf(`{"error":{"type":"upstream_error","code":%q,"message":%q}}`, code, message)) + shouldDisable := s.handleOpenAIAccountUpstreamError(ctx, account, statusCode, headers, responseBody, requestedModel) + return &UpstreamFailoverError{StatusCode: statusCode, ResponseBody: responseBody, ResponseHeaders: headers, + RetryableOnSameAccount: !shouldDisable && account.IsPoolMode() && account.IsPoolModeRetryableStatus(statusCode)} + } var upstreamErr *OpenAIImagesUpstreamError if !errors.As(err, &upstreamErr) { return err } retryable := IsOpenAIImagesRetryableUpstreamError(upstreamErr) - responseWritten := c != nil && c.Writer != nil && OpenAIImagesJSONKeepaliveAdjustedWrittenSize(c) != writerSizeBeforeResponse kind := "http_error" if retryable { kind = "failover" diff --git a/backend/internal/service/openai_images_test.go b/backend/internal/service/openai_images_test.go index e3edbaa5a..a1bf5caf7 100644 --- a/backend/internal/service/openai_images_test.go +++ b/backend/internal/service/openai_images_test.go @@ -26,6 +26,13 @@ type failingOpenAIImageWriter struct { writes int } +type openAIImagesReadErrorBody struct { + err error +} + +func (b *openAIImagesReadErrorBody) Read([]byte) (int, error) { return 0, b.err } +func (b *openAIImagesReadErrorBody) Close() error { return nil } + func (w *failingOpenAIImageWriter) Write(p []byte) (int, error) { if w.writes >= w.failAfter { return 0, errors.New("write failed: client disconnected") @@ -1032,6 +1039,111 @@ func TestOpenAIGatewayServiceForwardImages_OAuthNonStreamServerErrorReturnsFailo require.Equal(t, http.StatusBadGateway, events[0].UpstreamStatusCode) } +func TestOpenAIImagesOAuthBodyReadTransportErrorFailover(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/generations", nil) + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{ + "X-Request-Id": []string{"req_h2_read_failure"}, + "X-Upstream": []string{"preserved"}, + }, + Body: &openAIImagesReadErrorBody{err: errors.New("stream error: stream ID 11; INTERNAL_ERROR; received from peer")}, + } + account := &Account{ID: 5400, Name: "openai-oauth", Platform: PlatformOpenAI, Type: AccountTypeOAuth} + svc := &OpenAIGatewayService{} + + _, _, _, readErr := svc.handleOpenAIImagesOAuthNonStreamingResponse(resp, c, "b64_json", "gpt-image-2") + require.Error(t, readErr) + err := svc.handleOpenAIImagesOAuthResponseError(context.Background(), c, account, "gpt-image-2", "https://api.openai.com/v1/responses", resp, OpenAIImagesJSONKeepaliveAdjustedWrittenSize(c), readErr) + + var failoverErr *UpstreamFailoverError + require.ErrorAs(t, err, &failoverErr) + require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode) + require.JSONEq(t, `{"error":{"type":"upstream_error","code":"upstream_http2_stream_error","message":"Upstream HTTP/2 stream failed"}}`, string(failoverErr.ResponseBody)) + require.Equal(t, "req_h2_read_failure", failoverErr.ResponseHeaders.Get("x-request-id")) + require.Equal(t, "preserved", failoverErr.ResponseHeaders.Get("x-upstream")) + resp.Header.Set("X-Upstream", "mutated") + require.Equal(t, "preserved", failoverErr.ResponseHeaders.Get("x-upstream")) + + 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, "failover", events[0].Kind) + require.Equal(t, "req_h2_read_failure", events[0].UpstreamRequestID) + require.Equal(t, "Upstream HTTP/2 stream failed", events[0].Message) +} + +func TestOpenAIImagesOAuthBodyReadErrorsNotMisclassified(t *testing.T) { + tests := []struct { + name string + err error + }{ + {name: "context canceled", err: context.Canceled}, + {name: "response too large", err: fmt.Errorf("%w: limit=1", ErrUpstreamResponseBodyTooLarge)}, + {name: "semantic error", err: &OpenAIImagesUpstreamError{StatusCode: http.StatusBadRequest, ErrorType: "invalid_request_error", Code: "invalid_value", Message: "bad image request"}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/generations", nil) + resp := &http.Response{StatusCode: http.StatusOK, Header: make(http.Header)} + err := tt.err + if tt.name != "semantic error" && shouldClassifyOpenAIUpstreamStreamReadError(err, c.Request.Context()) { + err = newOpenAIUpstreamStreamReadError(err) + } + + got := (&OpenAIGatewayService{}).handleOpenAIImagesOAuthResponseError(context.Background(), c, &Account{Platform: PlatformOpenAI}, "gpt-image-2", "", resp, OpenAIImagesJSONKeepaliveAdjustedWrittenSize(c), err) + var failoverErr *UpstreamFailoverError + require.False(t, errors.As(got, &failoverErr)) + require.ErrorIs(t, got, tt.err) + }) + } +} + +func TestOpenAIImagesOAuthTransportErrorAfterDownstreamWriteDoesNotFailover(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/generations", nil) + before := OpenAIImagesJSONKeepaliveAdjustedWrittenSize(c) + _, writeErr := c.Writer.Write([]byte("downstream image bytes")) + require.NoError(t, writeErr) + classifiedErr := newOpenAIUpstreamStreamReadError(errors.New("unexpected EOF")) + account := &Account{ID: 5401, Name: "openai-oauth", Platform: PlatformOpenAI, Type: AccountTypeOAuth} + resp := &http.Response{Header: http.Header{"X-Request-Id": []string{"req_after_write"}}} + + err := (&OpenAIGatewayService{}).handleOpenAIImagesOAuthResponseError(context.Background(), c, account, "gpt-image-2", "", resp, before, classifiedErr) + + var failoverErr *UpstreamFailoverError + require.False(t, errors.As(err, &failoverErr)) + require.ErrorIs(t, err, classifiedErr) + 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_exhausted_failover", events[0].Kind) +} + +func TestShouldClassifyOpenAIUpstreamStreamReadErrorTransportStrings(t *testing.T) { + for _, message := range []string{"unexpected EOF", "connection reset by peer", "broken pipe", "use of closed network connection"} { + t.Run(message, func(t *testing.T) { + require.True(t, shouldClassifyOpenAIUpstreamStreamReadError(errors.New(message))) + }) + } + + canceledCtx, cancel := context.WithCancel(context.Background()) + cancel() + require.False(t, shouldClassifyOpenAIUpstreamStreamReadError(errors.New("unexpected EOF"), canceledCtx)) +} + func TestOpenAIGatewayServiceForwardImages_OAuthStreamServerErrorAfterFlushDoesNotFailover(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"gpt-image-2","prompt":"draw a cat","stream":true,"response_format":"b64_json"}`) diff --git a/backend/internal/service/openai_stream_read_error.go b/backend/internal/service/openai_stream_read_error.go index b318446e0..9a008d8e0 100644 --- a/backend/internal/service/openai_stream_read_error.go +++ b/backend/internal/service/openai_stream_read_error.go @@ -1,6 +1,7 @@ package service import ( + "context" "errors" "fmt" "strings" @@ -34,6 +35,20 @@ func newOpenAIUpstreamStreamReadError(err error) error { } } +// shouldClassifyOpenAIUpstreamStreamReadError excludes cancellation and +// response-size enforcement from upstream retry. +func shouldClassifyOpenAIUpstreamStreamReadError(err error, contexts ...context.Context) bool { + if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, ErrUpstreamResponseBodyTooLarge) { + return false + } + for _, ctx := range contexts { + if ctx != nil && ctx.Err() != nil { + return false + } + } + return true +} + // OpenAIUpstreamStreamReadErrorDetails returns the stable, sanitized client // classification attached to an upstream stream read failure. func OpenAIUpstreamStreamReadErrorDetails(err error) (code, message string, ok bool) {