fix: 应用 xhigh 代码评审的跨端修复

来自 xhigh code review 的正确性/健壮性修复，覆盖全部五端：
- server：鉴权 fail-closed、计量交叉校验与配额扣穿处理、WS 网关并发与关闭顺序、
  billing 行锁、redis Lua 过期与设备槽刷新、config 解析
- desktop：会话 epoch 防串话、WS 重连与 401 处理、api 客户端复用、统一 usePoll 轮询
- android：握手时序、请求头封装、账户状态派生、按需重组
- ios：finalize 宽限、串行采集、错误文案服务端优先、删除死代码 CommitController

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

server/internal/gateway/gateway.go

@@ -48,9 +48,14 @@ type wsConn struct {
 	c  *websocket.Conn
 }
 
+// wsWriteTimeout 单次写超时（16B）：客户端不读时避免 WriteJSON 永久阻塞，
+// 致 pumpResults / 收尾路径卡死、goroutine 与连接泄漏。
+const wsWriteTimeout = 5 * time.Second
+
 func (w *wsConn) sendJSON(v any) error {
 	w.mu.Lock()
 	defer w.mu.Unlock()
+	_ = w.c.SetWriteDeadline(time.Now().Add(wsWriteTimeout))
 	return w.c.WriteJSON(v)
 }
 
@@ -77,7 +82,6 @@ func (h *Handler) HandleWS(c *gin.Context) {
 	if err != nil {
 		return
 	}
-	defer conn.Close()
 	ws := &wsConn{c: conn}
 
 	conn.SetPongHandler(func(string) error {
@@ -86,7 +90,11 @@ func (h *Handler) HandleWS(c *gin.Context) {
 	_ = conn.SetReadDeadline(time.Now().Add(90 * time.Second))
 
 	var sess *session // 同连接串行多次会话
+	// 单个收尾 defer，顺序显式（16B）：先 conn.Close 拆连接——解开任何阻塞中的
+	// pumpResults 写，再 finish（其 Wait 已带超时）。避免 LIFO 让 finish 先跑、
+	// 而 Close 永不执行导致的 goroutine + 连接 + 设备槽泄漏。
 	defer func() {
+		conn.Close()
 		if sess != nil {
 			sess.finish(context.Background(), false)
 		}
@@ -165,6 +173,9 @@ func (h *Handler) startSession(ctx context.Context, ws *wsConn, uid, deviceID st
 		h: h, ws: ws, id: msg.SessionID, uid: uid, deviceID: deviceID,
 		provider: ps, started: now, done: make(chan struct{}),
 	}
+	// Add 须 happen-before finish 的 Wait（WaitGroup 文档要求），放在启动
+	// goroutine 之前，避免 Add 与 Wait 竞态（16C）。
+	s.resultsDone.Add(1)
 	go s.pumpResults()
 	go s.usageLoop()
 	return s
@@ -179,41 +190,48 @@ type session struct {
 	provider asr.Session
 	started  time.Time
 
+	// consumeMu 串行化 Consume RPC 与 finish 取快照（16D）：避免 finish 在
+	// 一笔在途 consumeDelta 已扣 Redis、尚未把 trialPart/balancePart 计入快照时
+	// 读到漏记的账。consumeMu 须在 mu 之外单独获取，且不可在持有 mu 时跨越 Redis 调用。
+	consumeMu sync.Mutex
+
 	mu            sync.Mutex
-	audioBytes    int
+	audioBytes    int   // 计费音频字节累计；截断/死亡/扣穿后在 feed 处冻结不再增长
 	providerEndMs int64 // Provider 报告的最大句尾时间戳（ms），0=未提供
 	consumedSec   int   // 已增量扣减的秒数
-	trialPart   int
-	balancePart int
-	truncated   bool
-	finished    bool
-	done        chan struct{}
-	resultsDone sync.WaitGroup
+	trialPart      int
+	balancePart    int
+	truncated      bool
+	providerDead   bool // provider 死亡（r.Err / Results 非正常关闭）后置位，停止计费（16A）
+	quotaExhausted bool // 余额+试用扣穿后置位，已下发过一次 QUOTA_EXCEEDED（16E）
+	finished       bool
+	done           chan struct{}
+	resultsDone    sync.WaitGroup
 }
 
 // feed 转发音频帧并累计实收时长；超 180s 自动截断。
+// 截断 / provider 死亡点之后到达的帧不再计入 audioBytes（不扣费）：
+// 用户为"未识别的音频"付费是 bug——先判停止条件，再决定是否累计（16A）。
 func (s *session) feed(pcm []byte) {
 	s.mu.Lock()
-	if s.finished {
+	if s.finished || s.truncated || s.providerDead || s.quotaExhausted {
+		// 已停止计费：丢弃此帧，不累计、不转发（扣穿后冻结，避免结算记录虚增秒数）
 		s.mu.Unlock()
 		return
 	}
+	// 先用"若计入本帧"的时长判是否超 180s
+	overCap := (s.audioBytes+len(pcm)+31999)/32000 >= protocol.MaxSessionSeconds
+	if overCap {
+		// 本帧触发截断：冻结记账（本帧不计入），flush 定稿
+		s.truncated = true
+		s.mu.Unlock()
+		// 截断：flush 定稿，但保持会话记账状态直至客户端 stop / 连接收尾
+		_ = s.provider.Close()
+		s.ws.sendErr(s.id, protocol.ErrSessionLimit)
+		return
+	}
 	s.audioBytes += len(pcm)
-	overCap := s.audioSeconds() >= protocol.MaxSessionSeconds
 	s.mu.Unlock()
-
-	if overCap {
-		s.mu.Lock()
-		already := s.truncated
-		s.truncated = true
-		s.mu.Unlock()
-		if !already {
-			// 截断：flush 定稿，但保持会话记账状态直至客户端 stop / 连接收尾
-			_ = s.provider.Close()
-			s.ws.sendErr(s.id, protocol.ErrSessionLimit)
-		}
-		return
-	}
 	_ = s.provider.SendAudio(pcm)
 }
 
@@ -223,11 +241,13 @@ func (s *session) audioSeconds() int {
 }
 
 // pumpResults Provider → 客户端下行泵。
+// Add(1) 已移至 startSession（16C）。provider 死亡（r.Err 或 Results 非正常关闭）
+// 时置 providerDead，停止后续计费（16A）。
 func (s *session) pumpResults() {
-	s.resultsDone.Add(1)
 	defer s.resultsDone.Done()
 	for r := range s.provider.Results() {
 		if r.Err != nil {
+			s.markProviderDead()
 			s.ws.sendErr(s.id, protocol.ErrASRUnavailable)
 			return
 		}
@@ -244,6 +264,19 @@ func (s *session) pumpResults() {
 		}
 		_ = s.ws.sendJSON(protocol.ServerMsg{Type: typ, SessionID: s.id, Text: r.Text})
 	}
+	// Results 通道关闭：若并非由 finish/截断/扣穿主动 Close 触发，则 provider 自行
+	// 死亡（上游 EOF / 异常 task-finished），标记停止计费（16A）。
+	s.markProviderDead()
+}
+
+// markProviderDead 标记 provider 已死：feed 据此冻结 audioBytes（不再计入死亡点
+// 之后的帧），consumeDelta 据此停止继续扣费；已识别部分仍由 finish 的 consumeFinal
+// 正常结算。由 finish/截断/扣穿主动 Close 触发的通道关闭也会走到这里，但此时计费已
+// 另行冻结，置位无副作用。
+func (s *session) markProviderDead() {
+	s.mu.Lock()
+	s.providerDead = true
+	s.mu.Unlock()
 }
 
 // usageLoop 每 2s 增量扣减并下发 usage 帧。
@@ -256,15 +289,28 @@ func (s *session) usageLoop() {
 			return
 		case <-t.C:
 			s.consumeDelta(context.Background())
+			// 续期设备槽（17D）：单会话墙钟可远超 4min TTL（低速发帧刷新读超时），
+			// 不续期则 slot 先于会话过期，同设备第二路 start 会被错误放行。
+			_, _ = store.RefreshDeviceSlot(context.Background(), s.h.RDB, s.deviceID, s.id)
 		}
 	}
 }
 
 // consumeDelta 将"实收音频秒数 − 已扣秒数"差额扣减并广播余额。
+// consumeMu 串行化与 finish 取快照（16D）：持锁期间 finish 不会读到漏记的在途账。
 func (s *session) consumeDelta(ctx context.Context) {
+	s.consumeMu.Lock()
+	defer s.consumeMu.Unlock()
+
 	s.mu.Lock()
+	// providerDead 后停止继续计费（16A）：audioBytes 已在 feed 处冻结，
+	// 已识别部分由 finish 的 consumeFinal 正常结算。
+	if s.finished || s.providerDead {
+		s.mu.Unlock()
+		return
+	}
 	delta := s.audioSeconds() - s.consumedSec
-	if delta <= 0 || s.finished {
+	if delta <= 0 {
 		s.mu.Unlock()
 		return
 	}
@@ -287,9 +333,30 @@ func (s *session) consumeDelta(ctx context.Context) {
 		BalanceSeconds: res.BalanceSeconds,
 		TrialRemaining: max(0, protocol.TrialDailySeconds-res.TrialUsedToday),
 	})
+
+	// 扣穿检测（16E）：余额与今日试用均已耗尽。优雅结束会话——不立刻掐断当句
+	// （Close 让 provider flush 当前句 final），下发一次 QUOTA_EXCEEDED 后由 finish 收尾。
+	if res.Exhausted() {
+		s.mu.Lock()
+		already := s.quotaExhausted
+		s.quotaExhausted = true
+		s.mu.Unlock()
+		if !already {
+			s.ws.sendErr(s.id, protocol.ErrQuotaExceeded)
+			// 触发 provider flush 当前句尾 final 并结束；finish 串行收尾。
+			_ = s.provider.Close()
+		}
+	}
 }
 
+// finishWaitTimeout 收尾等待尾部 final 下发的最长时间（16B）：客户端不读时
+// pumpResults 可能卡在写上（已由 sendJSON 写超时兜底），此处再加一层超时，
+// 超时则继续收尾不无限等。
+const finishWaitTimeout = 3 * time.Second
+
 // finish 结束会话：flush final → 补扣 → usage 帧 → 异步 settle → 记窗口 → 释放槽。
+// 注意 done/consumeMu 顺序（16D）：先停 usageLoop 并 drain 在途 consumeDelta，
+// 再读 trialPart/balancePart 快照，避免漏记在途扣费。
 func (s *session) finish(ctx context.Context, canceled bool) {
 	s.mu.Lock()
 	if s.finished {
@@ -299,11 +366,20 @@ func (s *session) finish(ctx context.Context, canceled bool) {
 	s.mu.Unlock()
 
 	_ = s.provider.Close()
-	s.resultsDone.Wait() // 尾部 final 全部下发后再收尾
+	s.waitResults() // 尾部 final 全部下发后再收尾（带超时，16B）
 
+	// 先标记 finished 并停 usageLoop，再 drain 在途 consumeDelta（16D）：
+	// 取 consumeMu 会等待任何已过 finished 检查、正等 Redis 返回的 consumeDelta
+	// 完成并把 trialPart/balancePart 计入，从而快照不漏账。
 	s.mu.Lock()
 	s.finished = true
 	close(s.done)
+	s.mu.Unlock()
+
+	s.consumeMu.Lock() // drain：等当前在途 consume（若有）完成
+	s.consumeMu.Unlock()
+
+	s.mu.Lock()
 	seconds := s.audioSeconds()
 	audioMs := int64(s.audioBytes) * 1000 / 32000
 	providerMs := s.providerEndMs
@@ -347,7 +423,23 @@ func (s *session) finish(ctx context.Context, canceled bool) {
 	}()
 }
 
+// waitResults 等 pumpResults 退出（尾部 final 下发完），最多 finishWaitTimeout（16B）。
+// 超时则放弃等待继续收尾——避免客户端不读时永久阻塞收尾路径。
+func (s *session) waitResults() {
+	done := make(chan struct{})
+	go func() {
+		s.resultsDone.Wait()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(finishWaitTimeout):
+		slog.Warn("finish: results wait timed out", "session", s.id)
+	}
+}
+
 // consumeFinal 结束时补扣差额（不足 2s 的短会话由此兜底）。
+// 由 finish 在持有 consumeMu drain 后调用，且 usageLoop 已停，天然与增量扣减串行。
 func (s *session) consumeFinal(ctx context.Context, seconds int) {
 	s.mu.Lock()
 	delta := seconds - s.consumedSec

server/internal/gateway/gateway_test.go

@@ -1,7 +1,6 @@
 package gateway
 
 import (
-	"context"
 	"encoding/json"
 	"net/http/httptest"
 	"strings"
@@ -155,35 +154,45 @@ func TestQuotaExceededOnStart(t *testing.T) {
 	}
 }
 
-// mrConsumeAll 把 u1 的当日试用直接耗尽（经 quota 通道，保证键一致）。
+// mrConsumeAll 把 u1 的当日试用耗尽（经 quota 通道，保证键一致）。
+// 单会话被截断在略低于 180s 处（截断帧不再计费，16A），故一会话不足以扣满 180s 试用；
+// 此处循环开会话推音频，直到某个 usage 帧报告 TrialRemaining<=0 为止。
 func mrConsumeAll(t *testing.T, srv *httptest.Server, db *gorm.DB) {
 	t.Helper()
-	conn := dial(t, srv)
-	start, _ := json.Marshal(protocol.ClientMsg{Type: "start", SessionID: "s0", SampleRate: 16000})
-	_ = conn.WriteMessage(websocket.TextMessage, start)
 	frame := make([]byte, protocol.FrameBytes)
-	// 180s 音频 = 1800 帧
-	for i := 0; i < protocol.TrialDailySeconds*10; i++ {
-		if err := conn.WriteMessage(websocket.BinaryMessage, frame); err != nil {
-			t.Fatal(err)
+	deadline := time.Now().Add(15 * time.Second)
+	for sess := 0; sess < 5 && time.Now().Before(deadline); sess++ {
+		conn := dial(t, srv)
+		sid := "s0_" + string(rune('a'+sess))
+		start, _ := json.Marshal(protocol.ClientMsg{Type: "start", SessionID: sid, SampleRate: 16000})
+		_ = conn.WriteMessage(websocket.TextMessage, start)
+		// 推 180s 音频（1800 帧）→ 会话在临界处截断
+		for i := 0; i < protocol.TrialDailySeconds*10; i++ {
+			if err := conn.WriteMessage(websocket.BinaryMessage, frame); err != nil {
+				break
+			}
+		}
+		stop, _ := json.Marshal(protocol.ClientMsg{Type: "stop", SessionID: sid})
+		_ = conn.WriteMessage(websocket.TextMessage, stop)
+
+		exhausted := false
+		for time.Now().Before(deadline) {
+			_ = conn.SetReadDeadline(time.Now().Add(2 * time.Second))
+			var msg protocol.ServerMsg
+			if err := conn.ReadJSON(&msg); err != nil {
+				break
+			}
+			if msg.Type == protocol.MsgUsage && msg.TrialRemaining <= 0 {
+				exhausted = true
+				break
+			}
+		}
+		conn.Close()
+		if exhausted {
+			return
 		}
 	}
-	stop, _ := json.Marshal(protocol.ClientMsg{Type: "stop", SessionID: "s0"})
-	_ = conn.WriteMessage(websocket.TextMessage, stop)
-	// 读到连接收尾的 usage 帧为止
-	deadline := time.Now().Add(5 * time.Second)
-	for time.Now().Before(deadline) {
-		_ = conn.SetReadDeadline(time.Now().Add(2 * time.Second))
-		var msg protocol.ServerMsg
-		if err := conn.ReadJSON(&msg); err != nil {
-			break
-		}
-		if msg.Type == protocol.MsgUsage && msg.SessionSeconds >= protocol.TrialDailySeconds {
-			break
-		}
-	}
-	conn.Close()
-	_ = context.Background()
+	t.Fatal("failed to exhaust trial via repeated sessions")
 }
 
 func TestSessionLimitTruncates(t *testing.T) {