From 7acc13a29b531aace9f5271082394974e3a064bb Mon Sep 17 00:00:00 2001 From: shaw Date: Sat, 25 Jul 2026 15:15:59 +0800 Subject: [PATCH] =?UTF-8?q?fix(live):=20=E7=A7=9F=E7=BA=A6=E4=B8=A2?= =?UTF-8?q?=E5=A4=B1=E7=BB=88=E6=AD=A2=E4=BC=9A=E8=AF=9D=E5=B9=B6=E8=A1=A5?= =?UTF-8?q?=E9=BD=90=E8=BF=87=E6=9C=9F=E6=97=B6=E7=9A=84=20usage=20log?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 三处功能修复（审计发现）： 1. 租约续租失败按会话终结处理。RefreshLiveLease 的 Lua 在 leaseID 被 GC 后不会 重新 ZADD，重连拿不回并发槽；原先把 ErrLiveUnavailable 当临时错误交给 observer 重连，会让会话以约 1 秒一轮的节奏空转到 ExpiresAt（最长 60 分钟），期间持着 上游 WS 连接却不计入账号/用户/API Key 的任何并发限制。 2. observer 因过期放弃时补写 usage log。waitForLiveObserverRetry 原先把过期判定 混在重试条件里并返回 false，直接 return 绕过了 observeLiveCall 循环顶部的过期 分支，导致该路径既不写 usage log 也不释放租约（静默结束）。改为只判控制权归属， 过期交回循环顶部 finalize。 3. 把两处重复的三项终止判据抽成 liveSessionEnded，消除 ProxyLiveSideband 与 observeLiveCall 之间的判据漂移风险。 迁移改名 186/187 -> 188/189：原编号会成为第三个 186，且 187 与已合入 main 的 #4801 的 187_add_usage_log_session_id.sql 撞号。文件名即迁移主键，功能无害但易误读。 新增两个测试均经变异验证（去掉修复后会失败）。 --- backend/internal/service/openai_live.go | 26 +++++++--- .../service/openai_live_lifecycle_test.go | 52 +++++++++++++++++++ ... => 188_allow_live_usage_request_type.sql} | 0 ..._live.sql => 189_add_group_allow_live.sql} | 0 4 files changed, 70 insertions(+), 8 deletions(-) rename backend/migrations/{186_allow_live_usage_request_type.sql => 188_allow_live_usage_request_type.sql} (100%) rename backend/migrations/{187_add_group_allow_live.sql => 189_add_group_allow_live.sql} (100%) diff --git a/backend/internal/service/openai_live.go b/backend/internal/service/openai_live.go index 3687c0c44..413d288aa 100644 --- a/backend/internal/service/openai_live.go +++ b/backend/internal/service/openai_live.go @@ -554,18 +554,26 @@ func (s *OpenAIGatewayService) ProxyLiveSideband( runErr := s.runLiveController(proxyCtx, record, upstream, errCh) cancel() _, _ = store.ReleaseLiveController(context.Background(), record.CallHash, owner) - if errors.Is(runErr, ErrLiveCallNotFound) { + if liveSessionEnded(runErr) || !time.Now().Before(record.ExpiresAt) { s.finalizeLiveCall(record) return runErr } - if !errors.Is(runErr, context.DeadlineExceeded) && time.Now().Before(record.ExpiresAt) { - go s.observeLiveCall(record.CallHash) - return runErr - } - s.finalizeLiveCall(record) + go s.observeLiveCall(record.CallHash) return runErr } +// liveSessionEnded 判断控制连接的退出原因是否意味着会话已终结（应 finalize：写 +// usage log 并释放租约），而不是可以交给 observer 重连的临时错误。 +// +// ErrLiveUnavailable 在控制循环里只会来自租约续租失败。RefreshLiveLease 的 Lua 在 +// leaseID 被 GC 后不会重新写入，重连也拿不回并发槽 —— 若按临时错误重试，会话会以 +// 约 1 秒一轮的节奏空转到 ExpiresAt，期间持着上游连接却不计入任何并发限制。 +func liveSessionEnded(err error) bool { + return errors.Is(err, ErrLiveCallNotFound) || + errors.Is(err, ErrLiveUnavailable) || + errors.Is(err, context.DeadlineExceeded) +} + func (s *OpenAIGatewayService) runLiveController( ctx context.Context, record *LiveCallRecord, @@ -626,7 +634,7 @@ func (s *OpenAIGatewayService) observeLiveCall(callHash string) { if errors.Is(runErr, ErrLiveControllerChanged) { return } - if errors.Is(runErr, context.DeadlineExceeded) || errors.Is(runErr, ErrLiveCallNotFound) { + if liveSessionEnded(runErr) { s.finalizeLiveCall(record) return } @@ -706,7 +714,9 @@ func (s *OpenAIGatewayService) waitForLiveObserverRetry(record *LiveCallRecord) return false } controller, err := store.GetLiveController(context.Background(), record.CallHash) - return err == nil && controller == LiveControllerObserver && time.Now().Before(record.ExpiresAt) + // 过期不在此处判定：返回 true 让调用方回到循环顶部的过期分支，由它 finalize + // （写 usage log + 释放租约）。在这里直接返回 false 会让会话静默结束、不留记录。 + return err == nil && controller == LiveControllerObserver } func (s *OpenAIGatewayService) refreshLiveLease(record *LiveCallRecord) bool { diff --git a/backend/internal/service/openai_live_lifecycle_test.go b/backend/internal/service/openai_live_lifecycle_test.go index 59571cf3c..7ad2c1b0f 100644 --- a/backend/internal/service/openai_live_lifecycle_test.go +++ b/backend/internal/service/openai_live_lifecycle_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "net/http" "net/http/httptest" "strings" @@ -415,3 +416,54 @@ func TestProxyLiveSidebandForwardsTextAndBinary(t *testing.T) { upstream.reads <- liveTestFrame{err: coderws.CloseError{Code: coderws.StatusNormalClosure}} require.ErrorIs(t, <-proxyResult, ErrLiveCallNotFound) } + +// TestLiveSessionEndedTreatsLeaseLossAsTerminal 锁定：租约续租失败（ErrLiveUnavailable） +// 必须判为会话终结。RefreshLiveLease 的 Lua 在 leaseID 被 GC 后不会重新写入，若把它 +// 当临时错误交给 observer 重连，会话会空转到 ExpiresAt 且不计入任何并发限制。 +func TestLiveSessionEndedTreatsLeaseLossAsTerminal(t *testing.T) { + cases := []struct { + name string + err error + want bool + }{ + {"租约丢失", ErrLiveUnavailable, true}, + {"租约丢失（被包装）", fmt.Errorf("refresh live lease: %w", ErrLiveUnavailable), true}, + {"上游报告会话已关闭", ErrLiveCallNotFound, true}, + {"到达会话时长上限", context.DeadlineExceeded, true}, + {"控制权被他人接管", ErrLiveControllerChanged, false}, + {"临时读错误", errors.New("unexpected EOF"), false}, + {"无错误", nil, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, liveSessionEnded(tc.err)) + }) + } +} + +// TestWaitForLiveObserverRetryLeavesExpiryToLoopFinalize 锁定：已过期但控制权仍在 +// observer 手上时返回 true，让调用方回到 observeLiveCall 循环顶部的过期分支去 +// finalize（写 usage log + 释放租约）。在此处直接返回 false 会让会话静默结束、不留记录。 +func TestWaitForLiveObserverRetryLeavesExpiryToLoopFinalize(t *testing.T) { + record := &LiveCallRecord{ + CallID: "call_expired", + CallHash: hashLiveCallID("call_expired"), + Controller: LiveControllerObserver, + ExpiresAt: time.Now().Add(-time.Minute), + } + store := &liveTestStore{} + require.NoError(t, store.SaveLiveCall(context.Background(), record, time.Hour)) + svc := &OpenAIGatewayService{cache: store} + + require.True(t, svc.waitForLiveObserverRetry(record), + "过期判定必须留给循环顶部，否则不会写 usage log") + + // 控制权已被他人接管时仍必须停止重试，避免与新控制者抢同一个 call。 + require.NoError(t, store.SaveLiveCall(context.Background(), &LiveCallRecord{ + CallID: record.CallID, + CallHash: record.CallHash, + Controller: LiveControllerProxy, + ExpiresAt: time.Now().Add(time.Hour), + }, time.Hour)) + require.False(t, svc.waitForLiveObserverRetry(record)) +} diff --git a/backend/migrations/186_allow_live_usage_request_type.sql b/backend/migrations/188_allow_live_usage_request_type.sql similarity index 100% rename from backend/migrations/186_allow_live_usage_request_type.sql rename to backend/migrations/188_allow_live_usage_request_type.sql diff --git a/backend/migrations/187_add_group_allow_live.sql b/backend/migrations/189_add_group_allow_live.sql similarity index 100% rename from backend/migrations/187_add_group_allow_live.sql rename to backend/migrations/189_add_group_allow_live.sql