fix(openai-live): Live 会话 finalize 与 observer 对 store 故障的容错

- 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 按时长接入计费管道）
2026-10-07 13:48:43 +08:00 · 2026-07-27 22:22:46 +08:00
parent 59ce11c780
commit 1c26dc7ad8
2 changed files with 205 additions and 12 deletions
@@ -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")
 }
@@ -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)
+}