fix(live): 租约丢失终止会话并补齐过期时的 usage log
三处功能修复(审计发现): 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 撞号。文件名即迁移主键,功能无害但易误读。 新增两个测试均经变异验证(去掉修复后会失败)。
This commit is contained in:
1 parent
db6fbdbf29
commit
7acc13a29b
4 files changed
+70
-8
No files matched your search
@@ -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 {
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
File renamed without changes.
File renamed without changes.
Reference in new issue
Block a user