From 1c26dc7ad85ed25482e067ba3c81b508b31b6180 Mon Sep 17 00:00:00 2001 From: yian Date: Mon, 27 Jul 2026 22:20:59 +0800 Subject: [PATCH] =?UTF-8?q?fix(openai-live):=20Live=20=E4=BC=9A=E8=AF=9D?= =?UTF-8?q?=20finalize=20=E4=B8=8E=20observer=20=E5=AF=B9=20store=20?= =?UTF-8?q?=E6=95=85=E9=9A=9C=E7=9A=84=E5=AE=B9=E9=94=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - finalizeLiveCall 写 usage log 改走 writeUsageLogBestEffort（日志 + 同步 兜底重试）。MarkLiveCallClosed 已在 Redis 标记 first，这是该会话唯一一次 落库机会，失败不能再被 `_, _ =` 静默丢弃 - observeLiveCall / waitForLiveObserverRetry 把 store 报错与「控制权被他人 接管」拆开：store 抖动时有限次重试（liveObserverStoreRetryLimit），仍失败 则按 record.ExpiresAt 兜底 finalize，保证 usage log 与租约释放不因 Redis 故障而丢失；记录确实不存在（ErrLiveCallNotFound）才停止重试 - Live 会话 TotalCost/ActualCost 恒 0 的零计费行为保持不变，加 TODO 注明 需产品决策（免费 or 按时长接入计费管道） --- backend/internal/service/openai_live.go | 83 +++++++++-- .../service/openai_live_lifecycle_test.go | 134 ++++++++++++++++++ 2 files changed, 205 insertions(+), 12 deletions(-) diff --git a/backend/internal/service/openai_live.go b/backend/internal/service/openai_live.go index 413d288aa6..6cce50573c 100644 --- a/backend/internal/service/openai_live.go +++ b/backend/internal/service/openai_live.go @@ -28,9 +28,13 @@ const ( liveRedisOperationTimeout = 3 * time.Second liveClosedRecordTTL = 24 * time.Hour liveObserverPollInterval = 250 * time.Millisecond + liveObserverStoreRetryLimit = 5 liveUpstreamBodyLimit = 2 << 20 ) +// liveObserverStoreRetryInterval 是 var 以便测试缩短 store 报错的重试等待。 +var liveObserverStoreRetryInterval = time.Second + var ( chatGPTLiveCallsURL = "https://chatgpt.com/backend-api/codex/realtime/calls?intent=quicksilver&architecture=avas" chatGPTLiveSidebandBaseURL = "wss://chatgpt.com/backend-api/codex" @@ -229,7 +233,7 @@ func (s *OpenAIGatewayService) CreateLiveCall( return nil, fmt.Errorf("save live call mapping: %w", saveErr) } created.Account = account - go s.observeLiveCall(record.CallHash) + go s.observeLiveCall(record) return created, nil } if lastErr != nil { @@ -508,7 +512,7 @@ func (s *OpenAIGatewayService) ProxyLiveSideband( upstream, err := s.dialLiveSideband(ctx, record) if err != nil { _, _ = store.ReleaseLiveController(context.Background(), record.CallHash, owner) - go s.observeLiveCall(record.CallHash) + go s.observeLiveCall(record) return err } defer func() { _ = upstream.Close() }() @@ -558,7 +562,7 @@ func (s *OpenAIGatewayService) ProxyLiveSideband( s.finalizeLiveCall(record) return runErr } - go s.observeLiveCall(record.CallHash) + go s.observeLiveCall(record) return runErr } @@ -603,19 +607,47 @@ func (s *OpenAIGatewayService) runLiveController( } } -func (s *OpenAIGatewayService) observeLiveCall(callHash string) { +func (s *OpenAIGatewayService) observeLiveCall(record *LiveCallRecord) { + if record == nil { + return + } store, err := s.liveStore() if err != nil { return } owner := uuid.NewString() - claimed, err := store.ClaimLiveController(context.Background(), callHash, LiveControllerObserver, owner) - if err != nil || !claimed { + claimed, claimErr := store.ClaimLiveController(context.Background(), record.CallHash, LiveControllerObserver, owner) + if claimErr != nil { + // store 报错时无法确认控制权归属，不能静默退出：若 claim 实际已生效而 + // observer 消失，租约与 usage log 都会丢。兜底 finalize 是幂等的，即使 + // 控制权在他人手上也只会在到期后落一次库。 + s.finalizeLiveCallAfterExpiry(record) return } + if !claimed { + return + } + storeErrStreak := 0 for { - record, getErr := store.GetLiveCall(context.Background(), callHash) - if getErr != nil || record.Controller != LiveControllerObserver { + latest, getErr := store.GetLiveCall(context.Background(), record.CallHash) + if getErr != nil { + // 记录已被清理（closed TTL 到期）不是故障，直接退出。 + if errors.Is(getErr, ErrLiveCallNotFound) { + return + } + // store 抖动不等于控制权被接管：有限次重试；仍失败则按 + // record.ExpiresAt 兜底 finalize，保证 usage log 与租约释放不丢。 + storeErrStreak++ + if storeErrStreak >= liveObserverStoreRetryLimit { + s.finalizeLiveCallAfterExpiry(record) + return + } + time.Sleep(liveObserverStoreRetryInterval) + continue + } + storeErrStreak = 0 + record = latest + if record.Controller != LiveControllerObserver { return } if !time.Now().Before(record.ExpiresAt) { @@ -713,10 +745,29 @@ func (s *OpenAIGatewayService) waitForLiveObserverRetry(record *LiveCallRecord) if err != nil { return false } - controller, err := store.GetLiveController(context.Background(), record.CallHash) + controller, getErr := store.GetLiveController(context.Background(), record.CallHash) + if getErr != nil && !errors.Is(getErr, ErrLiveCallNotFound) { + // store 报错不等于控制权被接管：返回 true 交回 observeLiveCall 循环顶部， + // 由它对 store 故障做有限次重试与 ExpiresAt 兜底 finalize。在这里返回 + // false 会让 Redis 抖动时会话静默结束、不留记录。 + return true + } // 过期不在此处判定：返回 true 让调用方回到循环顶部的过期分支，由它 finalize // （写 usage log + 释放租约）。在这里直接返回 false 会让会话静默结束、不留记录。 - return err == nil && controller == LiveControllerObserver + return getErr == nil && controller == LiveControllerObserver +} + +// finalizeLiveCallAfterExpiry 是 store 持续报错、observer 无法继续观察时的兜底： +// 等到会话最长时限 ExpiresAt 再 finalize，保证 usage log 与租约释放最迟在会话到期 +// 时完成。MarkLiveCallClosed 的 first 语义保证与其他恢复路径不会重复落库。 +func (s *OpenAIGatewayService) finalizeLiveCallAfterExpiry(record *LiveCallRecord) { + if record == nil { + return + } + if wait := time.Until(record.ExpiresAt); wait > 0 { + time.Sleep(wait) + } + s.finalizeLiveCall(record) } func (s *OpenAIGatewayService) refreshLiveLease(record *LiveCallRecord) bool { @@ -770,7 +821,15 @@ func (s *OpenAIGatewayService) finalizeLiveCall(record *LiveCallRecord) { if record.SubscriptionID > 0 { billingType = BillingTypeSubscription } - _, _ = s.usageLogRepo.Create(context.Background(), &UsageLog{ + // TODO(billing): Live 会话目前不计费：TotalCost/ActualCost 恒为 0，完全绕过 + // recordUsageCore/applyUsageBilling，余额模式下极低余额也能反复开启最长 + // liveMaxSessionDuration 的会话。若确认按时长计费，应在此接入计费管道； + // 若确认有意免费，删除本注释即可（零值行为由 + // TestFinalizeLiveCallIsIdempotentAndWritesZeroUsage 锁定）。 + // + // 这是该会话唯一一次落库机会（MarkLiveCallClosed 已标记 first），失败即永久 + // 丢失，因此走带日志与同步兜底的 writeUsageLogBestEffort（issue #3656）。 + writeUsageLogBestEffort(context.Background(), s.usageLogRepo, &UsageLog{ UserID: record.UserID, APIKeyID: record.APIKeyID, AccountID: record.AccountID, @@ -788,5 +847,5 @@ func (s *OpenAIGatewayService) finalizeLiveCall(record *LiveCallRecord) { InboundEndpoint: &inboundEndpoint, UpstreamEndpoint: &upstreamEndpoint, CreatedAt: record.CreatedAt, - }) + }, "service.openai_live") } diff --git a/backend/internal/service/openai_live_lifecycle_test.go b/backend/internal/service/openai_live_lifecycle_test.go index 7ad2c1b0f2..456f719efa 100644 --- a/backend/internal/service/openai_live_lifecycle_test.go +++ b/backend/internal/service/openai_live_lifecycle_test.go @@ -111,6 +111,10 @@ type liveTestStore struct { GatewayCache mu sync.Mutex record *LiveCallRecord + // 注入 store 故障（模拟 Redis 抖动），区别于 ErrLiveCallNotFound。 + claimErr error + getCallErr error + getControllerErr error } func (s *liveTestStore) SaveLiveCall(_ context.Context, record *LiveCallRecord, _ time.Duration) error { @@ -124,6 +128,9 @@ func (s *liveTestStore) SaveLiveCall(_ context.Context, record *LiveCallRecord, func (s *liveTestStore) GetLiveCall(_ context.Context, callHash string) (*LiveCallRecord, error) { s.mu.Lock() defer s.mu.Unlock() + if s.getCallErr != nil { + return nil, s.getCallErr + } if s.record == nil || s.record.CallHash != callHash { return nil, ErrLiveCallNotFound } @@ -134,6 +141,9 @@ func (s *liveTestStore) GetLiveCall(_ context.Context, callHash string) (*LiveCa func (s *liveTestStore) ClaimLiveController(_ context.Context, callHash, controller, owner string) (bool, error) { s.mu.Lock() defer s.mu.Unlock() + if s.claimErr != nil { + return false, s.claimErr + } if s.record == nil || s.record.CallHash != callHash || s.record.Controller == LiveControllerClosed { return false, nil } @@ -162,6 +172,9 @@ func (s *liveTestStore) ReleaseLiveController(_ context.Context, callHash, owner func (s *liveTestStore) GetLiveController(_ context.Context, callHash string) (string, error) { s.mu.Lock() defer s.mu.Unlock() + if s.getControllerErr != nil { + return "", s.getControllerErr + } if s.record == nil || s.record.CallHash != callHash { return "", ErrLiveCallNotFound } @@ -467,3 +480,124 @@ func TestWaitForLiveObserverRetryLeavesExpiryToLoopFinalize(t *testing.T) { }, time.Hour)) require.False(t, svc.waitForLiveObserverRetry(record)) } + +// TestWaitForLiveObserverRetryTreatsStoreErrorAsRetryable 锁定：store 报错（Redis +// 抖动）不等于控制权被接管，必须返回 true 交回 observeLiveCall 循环顶部，由它做 +// 有限次重试与 ExpiresAt 兜底 finalize；记录确实不存在时才停止重试。 +func TestWaitForLiveObserverRetryTreatsStoreErrorAsRetryable(t *testing.T) { + record := &LiveCallRecord{ + CallID: "call_flaky_store", + CallHash: hashLiveCallID("call_flaky_store"), + Controller: LiveControllerObserver, + ExpiresAt: time.Now().Add(time.Hour), + } + store := &liveTestStore{getControllerErr: errors.New("redis: connection refused")} + require.NoError(t, store.SaveLiveCall(context.Background(), record, time.Hour)) + svc := &OpenAIGatewayService{cache: store} + + require.True(t, svc.waitForLiveObserverRetry(record), + "store 报错必须继续重试，否则 Redis 抖动会让会话静默结束、不留记录") + + // 记录已被清理（ErrLiveCallNotFound）不是故障，应停止重试。 + require.False(t, (&OpenAIGatewayService{cache: &liveTestStore{}}).waitForLiveObserverRetry(record)) +} + +// TestObserveLiveCallStoreOutageFallsBackToExpiryFinalize 锁定：observer 遇到持续 +// store 报错时不能静默退出，必须按 record.ExpiresAt 兜底 finalize（写 usage log + +// 释放租约）。 +func TestObserveLiveCallStoreOutageFallsBackToExpiryFinalize(t *testing.T) { + restore := liveObserverStoreRetryInterval + liveObserverStoreRetryInterval = time.Millisecond + t.Cleanup(func() { liveObserverStoreRetryInterval = restore }) + + cases := []struct { + name string + inject func(*liveTestStore) + }{ + {"GetLiveCall 持续报错", func(s *liveTestStore) { s.getCallErr = errors.New("redis: i/o timeout") }}, + {"ClaimLiveController 报错", func(s *liveTestStore) { s.claimErr = errors.New("redis: i/o timeout") }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + record := &LiveCallRecord{ + CallID: "call_store_outage", + CallHash: hashLiveCallID("call_store_outage"), + AccountID: 11, + APIKeyID: 22, + UserID: 33, + LeaseID: "lease-1", + Model: "gpt-live-test", + CreatedAt: time.Now().Add(-time.Minute), + ExpiresAt: time.Now().Add(-time.Second), // 已到期：兜底无需等待 + Controller: LiveControllerPending, + } + store := &liveTestStore{} + require.NoError(t, store.SaveLiveCall(context.Background(), record, time.Hour)) + tc.inject(store) + concurrencyCache := &liveTestConcurrencyCache{} + usageRepo := &liveTestUsageRepo{} + svc := &OpenAIGatewayService{ + cache: store, + concurrencyService: NewConcurrencyService(concurrencyCache), + usageLogRepo: usageRepo, + } + + svc.observeLiveCall(record) + + concurrencyCache.mu.Lock() + require.Equal(t, 1, concurrencyCache.releases, "store 故障时租约释放不能丢") + concurrencyCache.mu.Unlock() + usageRepo.mu.Lock() + require.Len(t, usageRepo.logs, 1, "store 故障时 usage log 不能丢") + require.Equal(t, RequestTypeLive, usageRepo.logs[0].RequestType) + usageRepo.mu.Unlock() + }) + } +} + +type liveTestBestEffortUsageRepo struct { + liveTestUsageRepo + bestEffortErr error + bestEffortCalls int +} + +func (r *liveTestBestEffortUsageRepo) CreateBestEffort(_ context.Context, _ *UsageLog) error { + r.mu.Lock() + defer r.mu.Unlock() + r.bestEffortCalls++ + return r.bestEffortErr +} + +// TestFinalizeLiveCallUsageLogFallsBackToSyncCreate 锁定：finalize 是该会话唯一一次 +// 落库机会（MarkLiveCallClosed 已标记 first），best-effort 写入失败必须走同步 Create +// 兜底，而不是丢弃错误。 +func TestFinalizeLiveCallUsageLogFallsBackToSyncCreate(t *testing.T) { + record := &LiveCallRecord{ + CallID: "call_usage_fallback", + CallHash: hashLiveCallID("call_usage_fallback"), + AccountID: 11, + APIKeyID: 22, + UserID: 33, + LeaseID: "lease-1", + Model: "gpt-live-test", + CreatedAt: time.Now().Add(-time.Second), + ExpiresAt: time.Now().Add(time.Hour), + Controller: LiveControllerPending, + } + store := &liveTestStore{} + require.NoError(t, store.SaveLiveCall(context.Background(), record, time.Hour)) + usageRepo := &liveTestBestEffortUsageRepo{bestEffortErr: errors.New("usage log queue dropped")} + svc := &OpenAIGatewayService{ + cache: store, + concurrencyService: NewConcurrencyService(&liveTestConcurrencyCache{}), + usageLogRepo: usageRepo, + } + + svc.finalizeLiveCall(record) + + usageRepo.mu.Lock() + defer usageRepo.mu.Unlock() + require.Equal(t, 1, usageRepo.bestEffortCalls) + require.Len(t, usageRepo.logs, 1, "best-effort 失败后必须同步兜底落库") + require.Equal(t, record.CallHash, usageRepo.logs[0].RequestID) +}