mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-21 05:21:20 +00:00
将 compact fallback 的流式重试改为循环，避免递归重置单次重试标志
This commit is contained in:
IanShaw027
@@ -5,6 +5,7 @@ import (
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/tidwall/gjson"
|
||||
)
|
||||
@@ -143,3 +144,39 @@ func (s *OpenAIGatewayService) prepareOpenAICompactFallbackRetry(
|
||||
}
|
||||
return retryBody, fallbackModel, true
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) applyOpenAIPassthroughCompactFallbackFromSignal(
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
requestedModel string,
|
||||
body []byte,
|
||||
err error,
|
||||
alreadyRetried bool,
|
||||
resp *http.Response,
|
||||
) ([]byte, string, bool) {
|
||||
signal, ok := asOpenAICompactFallbackSignal(err)
|
||||
if !ok {
|
||||
return body, "", false
|
||||
}
|
||||
retryBody, fallbackModel, retry := s.prepareOpenAICompactFallbackRetry(
|
||||
c, account, requestedModel, body, http.StatusBadRequest, signal.message, signal.payload, alreadyRetried,
|
||||
)
|
||||
if !retry {
|
||||
return body, "", false
|
||||
}
|
||||
if resp != nil && resp.Body != nil {
|
||||
_ = resp.Body.Close()
|
||||
}
|
||||
fromModel := strings.TrimSpace(gjson.GetBytes(body, "model").String())
|
||||
accountName := ""
|
||||
if account != nil {
|
||||
accountName = account.Name
|
||||
}
|
||||
SetOpsUpstreamModel(c, fallbackModel)
|
||||
logger.LegacyPrintf(
|
||||
"service.openai_gateway",
|
||||
"[OpenAI passthrough] Retrying explicit compact request once with fallback model (account: %s, from: %s, to: %s, upstream_code: %s)",
|
||||
accountName, fromModel, fallbackModel, extractUpstreamErrorCode(signal.payload),
|
||||
)
|
||||
return retryBody, fallbackModel, true
|
||||
}
|
||||
|
||||
@@ -272,3 +272,41 @@ func TestOpenAIGatewayForwardRetriesStreamingCompactFailureBeforeOutput(t *testi
|
||||
require.NotContains(t, recorder.Body.String(), "context_length_exceeded")
|
||||
require.Contains(t, recorder.Body.String(), "response.completed")
|
||||
}
|
||||
|
||||
func TestOpenAIGatewayForwardDoesNotRecurseWhenCompactFallbackAlsoFails(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)
|
||||
|
||||
failed := "event: response.failed\n" +
|
||||
`data: {"type":"response.failed","response":{"status":"failed","error":{"code":"model_not_found","message":"model not found"}}}` + "\n\n"
|
||||
upstream := &httpUpstreamRecorder{responses: []*http.Response{
|
||||
{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(failed))},
|
||||
{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(failed))},
|
||||
{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(failed))},
|
||||
{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(failed))},
|
||||
{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(failed))},
|
||||
}}
|
||||
svc := &OpenAIGatewayService{
|
||||
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,
|
||||
}
|
||||
|
||||
result, err := svc.Forward(context.Background(), c, account, body)
|
||||
|
||||
require.Error(t, err)
|
||||
require.Nil(t, result)
|
||||
require.Len(t, upstream.bodies, 2)
|
||||
require.Equal(t, "gpt-5.5", gjson.GetBytes(upstream.bodies[0], "model").String())
|
||||
require.Equal(t, "gpt-5.4", gjson.GetBytes(upstream.bodies[1], "model").String())
|
||||
}
|
||||
|
||||
@@ -326,6 +326,11 @@ func (s *OpenAIGatewayService) forwardOpenAIPassthrough(
|
||||
compactModelFallbackRetried := false
|
||||
rejectedFieldRetryState := openAIResponsesRejectedFieldRetryStateForRequest(c, body)
|
||||
var resp *http.Response
|
||||
var usage *OpenAIUsage
|
||||
var firstTokenMs *int
|
||||
responseID := ""
|
||||
imageCount := 0
|
||||
var imageOutputSizes []string
|
||||
for {
|
||||
actualModel := strings.TrimSpace(gjson.GetBytes(body, "model").String())
|
||||
if actualModel == "" {
|
||||
@@ -347,116 +352,110 @@ func (s *OpenAIGatewayService) forwardOpenAIPassthrough(
|
||||
// a failover so the handler switches to a healthy account.
|
||||
return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, true)
|
||||
}
|
||||
if resp.StatusCode < 400 {
|
||||
break
|
||||
}
|
||||
|
||||
// Peek only to identify an invalid task. Restore the body so the existing
|
||||
// passthrough error handling sees the same response after recovery fails.
|
||||
probeBody := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
resp.Body = io.NopCloser(bytes.NewReader(probeBody))
|
||||
if retryBody, reason, changed, retryErr := normalizeOpenAIResponsesRejectedFieldRetryBody(resp.StatusCode, body, probeBody); retryErr != nil {
|
||||
return nil, fmt.Errorf("normalize passthrough rejected Responses field retry body: %w", retryErr)
|
||||
} else if changed && rejectedFieldRetryState.Allow(retryBody) {
|
||||
body = retryBody
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Retrying passthrough request after %s (account: %s)", reason, account.Name)
|
||||
continue
|
||||
}
|
||||
if !agentTaskRecoveryTried && s.isAgentIdentityAccount(ctx, account) && isAgentIdentityTaskInvalidHTTPResponse(resp.StatusCode, probeBody) {
|
||||
agentTaskRecoveryTried = true
|
||||
expectedTaskID := account.GetCredential("task_id")
|
||||
if recoveryErr := s.recoverAgentIdentityTask(ctx, account, expectedTaskID); recoveryErr != nil {
|
||||
return nil, fmt.Errorf("agent identity task recovery failed: %w", recoveryErr)
|
||||
if resp.StatusCode >= 400 {
|
||||
// Peek only to identify an invalid task. Restore the body so the existing
|
||||
// passthrough error handling sees the same response after recovery fails.
|
||||
probeBody := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
resp.Body = io.NopCloser(bytes.NewReader(probeBody))
|
||||
if retryBody, reason, changed, retryErr := normalizeOpenAIResponsesRejectedFieldRetryBody(resp.StatusCode, body, probeBody); retryErr != nil {
|
||||
return nil, fmt.Errorf("normalize passthrough rejected Responses field retry body: %w", retryErr)
|
||||
} else if changed && rejectedFieldRetryState.Allow(retryBody) {
|
||||
body = retryBody
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Retrying passthrough request after %s (account: %s)", reason, account.Name)
|
||||
continue
|
||||
}
|
||||
continue
|
||||
}
|
||||
upstreamMsg := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(probeBody)))
|
||||
if retryBody, fallbackModel, retry := s.prepareOpenAICompactFallbackRetry(
|
||||
c, account, requestedModel, body, resp.StatusCode, upstreamMsg, probeBody, compactModelFallbackRetried,
|
||||
); retry {
|
||||
fromModel := strings.TrimSpace(gjson.GetBytes(body, "model").String())
|
||||
body = retryBody
|
||||
upstreamPassthroughModel = fallbackModel
|
||||
compactModelFallbackRetried = true
|
||||
SetOpsUpstreamModel(c, fallbackModel)
|
||||
logger.LegacyPrintf(
|
||||
"service.openai_gateway",
|
||||
"[OpenAI passthrough] Retrying explicit compact request once with fallback model (account: %s, from: %s, to: %s, upstream_code: %s)",
|
||||
account.Name, fromModel, fallbackModel, extractUpstreamErrorCode(probeBody),
|
||||
)
|
||||
continue
|
||||
if !agentTaskRecoveryTried && s.isAgentIdentityAccount(ctx, account) && isAgentIdentityTaskInvalidHTTPResponse(resp.StatusCode, probeBody) {
|
||||
agentTaskRecoveryTried = true
|
||||
expectedTaskID := account.GetCredential("task_id")
|
||||
if recoveryErr := s.recoverAgentIdentityTask(ctx, account, expectedTaskID); recoveryErr != nil {
|
||||
return nil, fmt.Errorf("agent identity task recovery failed: %w", recoveryErr)
|
||||
}
|
||||
continue
|
||||
}
|
||||
upstreamMsg := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(probeBody)))
|
||||
if retryBody, fallbackModel, retry := s.prepareOpenAICompactFallbackRetry(
|
||||
c, account, requestedModel, body, resp.StatusCode, upstreamMsg, probeBody, compactModelFallbackRetried,
|
||||
); retry {
|
||||
fromModel := strings.TrimSpace(gjson.GetBytes(body, "model").String())
|
||||
body = retryBody
|
||||
upstreamPassthroughModel = fallbackModel
|
||||
compactModelFallbackRetried = true
|
||||
SetOpsUpstreamModel(c, fallbackModel)
|
||||
logger.LegacyPrintf(
|
||||
"service.openai_gateway",
|
||||
"[OpenAI passthrough] Retrying explicit compact request once with fallback model (account: %s, from: %s, to: %s, upstream_code: %s)",
|
||||
account.Name, fromModel, fallbackModel, extractUpstreamErrorCode(probeBody),
|
||||
)
|
||||
continue
|
||||
}
|
||||
|
||||
// 透传模式默认保持原样代理；容量错误以及 API-key 上游的瞬时
|
||||
// 5xx 应先触发多账号 failover，且此时尚未写入下游响应。
|
||||
// probeBody 已在上方任务探测时读取过一次，直接复用避免重复读取。
|
||||
if shouldFailoverOpenAIPassthroughResponse(account, resp.StatusCode, probeBody) {
|
||||
return nil, s.handleFailoverErrorResponsePassthrough(ctx, resp, c, account, body, probeBody)
|
||||
}
|
||||
return nil, s.handleErrorResponsePassthrough(ctx, resp, c, account, body, probeBody)
|
||||
}
|
||||
|
||||
// 透传模式默认保持原样代理；容量错误以及 API-key 上游的瞬时
|
||||
// 5xx 应先触发多账号 failover，且此时尚未写入下游响应。
|
||||
// probeBody 已在上方任务探测时读取过一次，直接复用避免重复读取。
|
||||
if shouldFailoverOpenAIPassthroughResponse(account, resp.StatusCode, probeBody) {
|
||||
return nil, s.handleFailoverErrorResponsePassthrough(ctx, resp, c, account, body, probeBody)
|
||||
if mapping, ok := openAIResponsesClientToolMapping(c); ok && isEventStreamResponse(resp.Header) {
|
||||
maxLineSize := defaultMaxLineSize
|
||||
if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 {
|
||||
maxLineSize = s.cfg.Gateway.MaxLineSize
|
||||
}
|
||||
resp.Body = newGrokResponsesClientToolStreamBody(resp.Body, mapping, maxLineSize)
|
||||
}
|
||||
return nil, s.handleErrorResponsePassthrough(ctx, resp, c, account, body, probeBody)
|
||||
|
||||
// x-codex-turn-state 溯源：下游回传由 writeOpenAIPassthroughResponseHeaders
|
||||
// 在各 handler 的写头点强制放行，铸造账号在此统一记录，供出站守卫剥离
|
||||
// failover 换号后的跨账号回带（openai_codex_turn_state.go）。
|
||||
if extractOpenAICodexTurnState(resp.Header) != "" {
|
||||
s.noteOpenAICodexTurnStateProvenance(c, account)
|
||||
}
|
||||
|
||||
if reqStream {
|
||||
result, handleErr := s.handleStreamingResponsePassthrough(ctx, resp, c, account, startTime, reqModel, upstreamPassthroughModel)
|
||||
if handleErr != nil {
|
||||
if retryBody, fallbackModel, retry := s.applyOpenAIPassthroughCompactFallbackFromSignal(
|
||||
c, account, requestedModel, body, handleErr, compactModelFallbackRetried, resp,
|
||||
); retry {
|
||||
body = retryBody
|
||||
upstreamPassthroughModel = fallbackModel
|
||||
compactModelFallbackRetried = true
|
||||
continue
|
||||
}
|
||||
_ = resp.Body.Close()
|
||||
return nil, handleErr
|
||||
}
|
||||
usage = result.usage
|
||||
firstTokenMs = result.firstTokenMs
|
||||
responseID = strings.TrimSpace(result.responseID)
|
||||
imageCount = result.imageCount
|
||||
imageOutputSizes = result.imageOutputSizes
|
||||
} else {
|
||||
result, handleErr := s.handleNonStreamingResponsePassthrough(ctx, resp, c, account, reqModel, upstreamPassthroughModel)
|
||||
if handleErr != nil {
|
||||
if retryBody, fallbackModel, retry := s.applyOpenAIPassthroughCompactFallbackFromSignal(
|
||||
c, account, requestedModel, body, handleErr, compactModelFallbackRetried, resp,
|
||||
); retry {
|
||||
body = retryBody
|
||||
upstreamPassthroughModel = fallbackModel
|
||||
compactModelFallbackRetried = true
|
||||
continue
|
||||
}
|
||||
_ = resp.Body.Close()
|
||||
return nil, handleErr
|
||||
}
|
||||
usage = result.usage
|
||||
responseID = strings.TrimSpace(result.responseID)
|
||||
imageCount = result.imageCount
|
||||
imageOutputSizes = result.imageOutputSizes
|
||||
}
|
||||
break
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
if mapping, ok := openAIResponsesClientToolMapping(c); ok && isEventStreamResponse(resp.Header) {
|
||||
maxLineSize := defaultMaxLineSize
|
||||
if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 {
|
||||
maxLineSize = s.cfg.Gateway.MaxLineSize
|
||||
}
|
||||
resp.Body = newGrokResponsesClientToolStreamBody(resp.Body, mapping, maxLineSize)
|
||||
}
|
||||
|
||||
serviceTier := extractOpenAIServiceTierFromBody(body)
|
||||
|
||||
// x-codex-turn-state 溯源：下游回传由 writeOpenAIPassthroughResponseHeaders
|
||||
// 在各 handler 的写头点强制放行，铸造账号在此统一记录，供出站守卫剥离
|
||||
// failover 换号后的跨账号回带（openai_codex_turn_state.go）。
|
||||
if extractOpenAICodexTurnState(resp.Header) != "" {
|
||||
s.noteOpenAICodexTurnStateProvenance(c, account)
|
||||
}
|
||||
|
||||
var usage *OpenAIUsage
|
||||
var firstTokenMs *int
|
||||
responseID := ""
|
||||
imageCount := 0
|
||||
var imageOutputSizes []string
|
||||
if reqStream {
|
||||
result, err := s.handleStreamingResponsePassthrough(ctx, resp, c, account, startTime, reqModel, upstreamPassthroughModel)
|
||||
if err != nil {
|
||||
if signal, ok := asOpenAICompactFallbackSignal(err); ok {
|
||||
if retryBody, fallbackModel, retry := s.prepareOpenAICompactFallbackRetry(
|
||||
c, account, requestedModel, body, http.StatusBadRequest, signal.message, signal.payload, compactModelFallbackRetried,
|
||||
); retry {
|
||||
body = retryBody
|
||||
SetOpsUpstreamModel(c, fallbackModel)
|
||||
return s.forwardOpenAIPassthrough(ctx, c, account, body, canonicalImageIntentBody, requestedModel, true, reasoningEffort, reqStream, startTime)
|
||||
}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
usage = result.usage
|
||||
firstTokenMs = result.firstTokenMs
|
||||
responseID = strings.TrimSpace(result.responseID)
|
||||
imageCount = result.imageCount
|
||||
imageOutputSizes = result.imageOutputSizes
|
||||
} else {
|
||||
result, err := s.handleNonStreamingResponsePassthrough(ctx, resp, c, account, reqModel, upstreamPassthroughModel)
|
||||
if err != nil {
|
||||
if signal, ok := asOpenAICompactFallbackSignal(err); ok {
|
||||
if retryBody, fallbackModel, retry := s.prepareOpenAICompactFallbackRetry(
|
||||
c, account, requestedModel, body, http.StatusBadRequest, signal.message, signal.payload, compactModelFallbackRetried,
|
||||
); retry {
|
||||
body = retryBody
|
||||
SetOpsUpstreamModel(c, fallbackModel)
|
||||
return s.forwardOpenAIPassthrough(ctx, c, account, body, canonicalImageIntentBody, requestedModel, true, reasoningEffort, reqStream, startTime)
|
||||
}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
usage = result.usage
|
||||
responseID = strings.TrimSpace(result.responseID)
|
||||
imageCount = result.imageCount
|
||||
imageOutputSizes = result.imageOutputSizes
|
||||
}
|
||||
s.bindHTTPResponseAccount(ctx, c, account, responseID)
|
||||
|
||||
// 排除 spark 影子:其 codex_* 仅由 QueryUsage(/wham/usage bengalfox)更新(外审第7轮 P1)。
|
||||
|
||||
Reference in New Issue
Block a user