修复跨区间缓存串数据与会话回收泄漏,收敛网关重试

2026-07-17 12:19:21 +08:00
parent 7019d4c5a6
commit 882eeade1e
6 changed files with 121 additions and 68 deletions
@@ -27,27 +27,27 @@ type aiCandidate struct {
 	modelOcid string
 }

-// RespPassthrough 编排一次非流式直通调用:选渠道(priority→加权随机)→ 调用 →
-// 可重试错误换渠道(整请求上限 3 次)并维护熔断;group 非空时只在同分组渠道内路由。
-// 上游为 OpenAI-compatible /actions/v1/responses(实测可用,无 Oracle 文档合同)。
-func (s *AiGatewayService) RespPassthrough(ctx context.Context, raw []byte, modelName, group string) ([]byte, ChatMeta, error) {
+// routeRetry 统一编排「选渠道(priority→加权随机)→ 调用 → 可重试错误换渠道」,
+// 整请求上限 3 次并维护熔断;group 非空时只在同分组渠道内路由。
+func routeRetry[T any](ctx context.Context, s *AiGatewayService, modelName, group, capability string, once func(*aiCandidate) (T, error)) (T, ChatMeta, error) {
+	var zero T
 	meta := ChatMeta{}
 	excluded := map[uint]bool{}
 	var lastErr error
 	for attempt := 0; attempt < 3; attempt++ {
-		cand, err := s.pick(ctx, modelName, group, "CHAT", excluded)
+		cand, err := s.pick(ctx, modelName, group, capability, excluded)
 		if err != nil {
-			return nil, meta, firstErr(lastErr, err)
+			return zero, meta, firstErr(lastErr, err)
 		}
 		meta.ChannelID, meta.ChannelName = cand.ch.ID, cand.ch.Name
-		payload, err := s.passthroughOnce(ctx, cand, raw)
+		out, err := once(cand)
 		if err == nil {
 			s.markSuccess(ctx, cand.ch.ID)
-			return payload, meta, nil
+			return out, meta, nil
 		}
 		retry, penalize := switchable(err)
 		if !retry {
-			return nil, meta, err
+			return zero, meta, err
 		}
 		if penalize {
 			s.markFailure(ctx, cand.ch.ID)
@@ -56,7 +56,15 @@ func (s *AiGatewayService) RespPassthrough(ctx context.Context, raw []byte, mode
 		meta.Retries++
 		lastErr = err
 	}
-	return nil, meta, lastErr
+	return zero, meta, lastErr
+}
+
+// RespPassthrough 编排一次非流式直通调用。
+// 上游为 OpenAI-compatible /actions/v1/responses(实测可用,无 Oracle 文档合同)。
+func (s *AiGatewayService) RespPassthrough(ctx context.Context, raw []byte, modelName, group string) ([]byte, ChatMeta, error) {
+	return routeRetry(ctx, s, modelName, group, "CHAT", func(cand *aiCandidate) ([]byte, error) {
+		return s.passthroughOnce(ctx, cand, raw)
+	})
 }

 func (s *AiGatewayService) passthroughOnce(ctx context.Context, cand *aiCandidate, raw []byte) ([]byte, error) {
@@ -70,36 +78,13 @@ func (s *AiGatewayService) passthroughOnce(ctx context.Context, cand *aiCandidat
 // RespPassthroughStream 编排流式直通:流建立成功即绑定渠道,建立失败按 switchable
 // 换渠道重试;建立后的中断不重试、不计熔断(与 OpenStream 语义一致)。
 func (s *AiGatewayService) RespPassthroughStream(ctx context.Context, raw []byte, modelName, group string) (io.ReadCloser, ChatMeta, error) {
-	meta := ChatMeta{}
-	excluded := map[uint]bool{}
-	var lastErr error
-	for attempt := 0; attempt < 3; attempt++ {
-		cand, err := s.pick(ctx, modelName, group, "CHAT", excluded)
-		if err != nil {
-			return nil, meta, firstErr(lastErr, err)
-		}
-		meta.ChannelID, meta.ChannelName = cand.ch.ID, cand.ch.Name
+	return routeRetry(ctx, s, modelName, group, "CHAT", func(cand *aiCandidate) (io.ReadCloser, error) {
 		cred, err := s.configs.credentialsByID(ctx, cand.ch.OciConfigID)
 		if err != nil {
-			return nil, meta, err
+			return nil, err
 		}
-		stream, err := s.client.GenAiCompatResponsesStream(ctx, cred, cand.ch.Region, raw, s.UpstreamWait())
-		if err == nil {
-			s.markSuccess(ctx, cand.ch.ID)
-			return stream, meta, nil
-		}
-		retry, penalize := switchable(err)
-		if !retry {
-			return nil, meta, err
-		}
-		if penalize {
-			s.markFailure(ctx, cand.ch.ID)
-		}
-		excluded[cand.ch.ID] = true
-		meta.Retries++
-		lastErr = err
-	}
-	return nil, meta, lastErr
+		return s.client.GenAiCompatResponsesStream(ctx, cred, cand.ch.Region, raw, s.UpstreamWait())
+	})
 }

 // firstErr 在换渠道后仍失败时优先返回上游错误(而非「无渠道」)。
@@ -221,34 +206,11 @@ func weightedPick(chs []model.AiChannel) model.AiChannel {
 	return chs[len(chs)-1]
 }

-// Embeddings 编排向量化调用:按 EMBEDDING 能力选渠道,可重试错误换渠道(整请求上限 3 次)。
+// Embeddings 编排向量化调用:按 EMBEDDING 能力选渠道,可重试错误换渠道。
 func (s *AiGatewayService) Embeddings(ctx context.Context, req aiwire.EmbeddingsRequest, group string) (*aiwire.EmbeddingsResponse, ChatMeta, error) {
-	meta := ChatMeta{}
-	excluded := map[uint]bool{}
-	var lastErr error
-	for attempt := 0; attempt < 3; attempt++ {
-		cand, err := s.pick(ctx, req.Model, group, "EMBEDDING", excluded)
-		if err != nil {
-			return nil, meta, firstErr(lastErr, err)
-		}
-		meta.ChannelID, meta.ChannelName = cand.ch.ID, cand.ch.Name
-		resp, err := s.embedOnce(ctx, cand, req)
-		if err == nil {
-			s.markSuccess(ctx, cand.ch.ID)
-			return resp, meta, nil
-		}
-		retry, penalize := switchable(err)
-		if !retry {
-			return nil, meta, err
-		}
-		if penalize {
-			s.markFailure(ctx, cand.ch.ID)
-		}
-		excluded[cand.ch.ID] = true
-		meta.Retries++
-		lastErr = err
-	}
-	return nil, meta, lastErr
+	return routeRetry(ctx, s, req.Model, group, "EMBEDDING", func(cand *aiCandidate) (*aiwire.EmbeddingsResponse, error) {
+		return s.embedOnce(ctx, cand, req)
+	})
 }

 // embedOnce 调用渠道向量化并装配 OpenAI 形态响应。
@@ -46,6 +46,7 @@ type ConsoleSession struct {
 type ConsoleService struct {
 	configs      *OciConfigService
 	pollInterval time.Duration // 清理残留连接的轮询间隔，测试注入缩短
+	sessionTTL   time.Duration // 会话回收检查周期，测试注入缩短

 	mu       sync.Mutex
 	sessions map[string]*ConsoleSession
@@ -55,6 +56,7 @@ func NewConsoleService(configs *OciConfigService) *ConsoleService {
 	return &ConsoleService{
 		configs:      configs,
 		pollInterval: 2 * time.Second,
+		sessionTTL:   consoleSessionTTL,
 		sessions:     map[string]*ConsoleSession{},
 	}
 }
@@ -161,11 +163,12 @@ func (s *ConsoleService) storeSession(cfgID uint, instanceID, region, typ, connI
 	s.mu.Lock()
 	s.sessions[sess.ID] = sess
 	s.mu.Unlock()
-	time.AfterFunc(consoleSessionTTL, func() { s.expire(sess.ID) })
+	time.AfterFunc(s.sessionTTL, func() { s.expire(sess.ID) })
 	return sess
 }

-// expire TTL 到期回收：正在使用的会话跳过（连接断开后自然停止，无续期）。
+// expire TTL 到期回收：正在使用的会话跳过并重挂下一轮检查，
+// 连接断开后由后续轮次回收，避免会话与云端连接常驻到进程退出。
 func (s *ConsoleService) expire(id string) {
 	s.mu.Lock()
 	sess, ok := s.sessions[id]
@@ -173,6 +176,7 @@ func (s *ConsoleService) expire(id string) {
 		sess.mu.Lock()
 		if sess.inUse {
 			ok = false
+			time.AfterFunc(s.sessionTTL, func() { s.expire(id) })
 		} else {
 			delete(s.sessions, id)
 		}
@@ -116,3 +116,44 @@ func TestCreateConsoleSession(t *testing.T) {
 		})
 	}
 }
+
+// TestConsoleSessionExpire 验证过期回收:在用会话跳过本轮,断开后下一轮回收。
+func TestConsoleSessionExpire(t *testing.T) {
+	tests := []struct {
+		name      string
+		inUse     bool
+		wantAlive bool
+	}{
+		{name: "在用会话跳过回收", inUse: true, wantAlive: true},
+		{name: "空闲会话回收并删云端连接", inUse: false, wantAlive: false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			client := &consoleFakeClient{}
+			console, cfgID := newConsoleTestService(t, client)
+			sess, err := console.CreateSession(context.Background(), cfgID, "ocid1.instance.i", "", ConsoleTypeSerial)
+			if err != nil {
+				t.Fatalf("CreateSession: %v", err)
+			}
+			sess.MarkUse(tt.inUse)
+			console.expire(sess.ID)
+			if alive := console.Get(sess.ID) != nil; alive != tt.wantAlive {
+				t.Fatalf("session alive = %v, want %v", alive, tt.wantAlive)
+			}
+			if wantDel := !tt.wantAlive; (len(client.deleted) == 1) != wantDel {
+				t.Errorf("cloud connection deleted %v, want %v", client.deleted, wantDel)
+			}
+			if tt.inUse {
+				// 断开后下一轮回收(此前的缺陷:跳过后不再检查,会话常驻)
+				sess.MarkUse(false)
+				console.expire(sess.ID)
+				if console.Get(sess.ID) != nil {
+					t.Fatal("session still alive after idle expire round")
+				}
+				if len(client.deleted) != 1 {
+					t.Errorf("cloud connection not deleted after idle expire: %v", client.deleted)
+				}
+			}
+		})
+	}
+}