From 7706f59549c736a4d003c5e46cf5b8d0bd0d3792 Mon Sep 17 00:00:00 2001 From: Wang Defa <1+wangdefa@noreply.gitea.bcde.io> Date: Fri, 10 Jul 2026 20:25:37 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=91=E5=B8=83=200.2.0:=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E6=B1=A0=E8=87=AA=E6=84=88=E3=80=81=E6=8E=A2=E6=B5=8B=E4=BF=AE?= =?UTF-8?q?=E6=AD=A3=E3=80=81=E4=BB=BB=E5=8A=A1=E5=BC=82=E6=AD=A5=E8=A7=A6?= =?UTF-8?q?=E5=8F=91=E4=B8=8E=E5=88=A0=E9=99=A4=E5=8A=A0=E5=9B=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitea/workflows/release.yml | 4 +- CHANGELOG.md | 18 ++ docs/docs.go | 13 +- docs/swagger.json | 13 +- docs/swagger.yaml | 11 +- internal/api/task.go | 15 +- internal/model/models.go | 7 + internal/oci/errors.go | 31 +++ internal/oci/errors_test.go | 49 ++++ internal/oci/genai.go | 42 +++- internal/oci/genai_retire_test.go | 30 +++ internal/service/aigateway.go | 242 ++++++++++++++++++-- internal/service/aigateway_chat.go | 44 ++-- internal/service/aigateway_test.go | 314 +++++++++++++++++++++++++- internal/service/ociconfig_test.go | 7 + internal/service/task.go | 63 +++++- internal/service/task_test.go | 34 +++ internal/service/tenantdelete.go | 79 +++---- internal/service/tenantdelete_test.go | 69 +++++- 19 files changed, 978 insertions(+), 107 deletions(-) diff --git a/.gitea/workflows/release.yml b/.gitea/workflows/release.yml index 7c69282..c0a04d4 100644 --- a/.gitea/workflows/release.yml +++ b/.gitea/workflows/release.yml @@ -23,7 +23,7 @@ jobs: echo "IMAGE=${REGISTRY}/${{ gitea.repository }}" >> "$GITHUB_ENV" echo "BUILD_TIME=$(date -u +%Y-%m-%dT%H:%M:%SZ)" >> "$GITHUB_ENV" - - name: 下载 DASH_VERSION 固定版本的前端 dist.zip,校验后解压进嵌入目录 + - name: 下载前端 dist.zip env: TOKEN: ${{ secrets.BUILD_TOKEN }} run: | @@ -36,7 +36,7 @@ jobs: unzip -q dist.zip -d internal/webui/dist test -f internal/webui/dist/index.html - - name: 构建双架构二进制(artifact stage 导出) + - name: 构建双架构二进制 run: | docker buildx build --pull \ --platform linux/amd64,linux/arm64 \ diff --git a/CHANGELOG.md b/CHANGELOG.md index 67fd6e8..147a1c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,24 @@ 格式遵循 [Keep a Changelog](https://keepachangelog.com/zh-CN/1.1.0/)，版本号遵循语义化版本。 +## [0.2.0] - 2026-07-10 + +### Added + +- AI 网关模型池自愈：同步与每日探测后自动逐个试调验证模型可用性（maxTokens=1），不可按需调用的模型（微调基座 / 区域未开放按需供给）自动标记并从模型列表、路由与探测候选中剔除；已标记模型每 20 小时复检，OCI 恢复供给后自动解除标记。`ai_model_caches` 新增 `unusable` / `unusable_reason` / `checked_at`（启动自动迁移），渠道详情接口保留标记行与原因便于排查 + +### Changed + +- 后台任务「立即执行」改异步触发：接口立即返回 202，执行结果经任务日志轮询呈现；同一任务在途时重复触发返回 409「任务正在执行中」，与 cron 重叠触发静默跳过（此前同步阻塞数十秒且同一任务可并发重复执行） +- 渠道探测候选跨厂商分散（上限 8，voice 等不可对话形态排除）：单一厂商在区域内全为微调基座时不再拖垮整个渠道的探测结论 +- 网关调用（对话 / 流式 / 向量化）遇「模型不可按需调用」类错误（微调基座 400 / 实体不存在 404）自动标记该模型并换渠道重试，且不计入渠道熔断；仅鉴权类错误才判定租户无配额 +- 同名模型多条目去重优先保留非微调基座条目，降低缓存到不可调用 OCID 的概率 + +### Fixed + +- 渠道「同步模型成功但探测报错不可用」：法兰克福等区域的微调基座模型占满探测候选导致的误报（探测状态与真实可用性不符的根因） +- 租户删除：告警命中清理改子查询，日志事件数万条时不再超出 SQL 绑定变量上限导致删除失败；无法解析的任务 payload 记警告跳过并保留原任务，不再永久阻断租户删除 + ## [0.1.0] - 2026-07-10 ### Added diff --git a/docs/docs.go b/docs/docs.go index 0596abf..e27f341 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -5738,7 +5738,7 @@ const docTemplate = `{ "tags": [ "任务与日志回传" ], - "summary": "立即执行任务", + "summary": "立即执行任务(异步触发,结果经任务日志轮询获取)", "parameters": [ { "type": "integer", @@ -5749,8 +5749,15 @@ const docTemplate = `{ } ], "responses": { - "200": { - "description": "OK", + "202": { + "description": "Accepted", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "409": { + "description": "任务正在执行中", "schema": { "type": "object", "additionalProperties": true diff --git a/docs/swagger.json b/docs/swagger.json index e1539c5..6027c2f 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -5731,7 +5731,7 @@ "tags": [ "任务与日志回传" ], - "summary": "立即执行任务", + "summary": "立即执行任务(异步触发,结果经任务日志轮询获取)", "parameters": [ { "type": "integer", @@ -5742,8 +5742,15 @@ } ], "responses": { - "200": { - "description": "OK", + "202": { + "description": "Accepted", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "409": { + "description": "任务正在执行中", "schema": { "type": "object", "additionalProperties": true diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 887aa21..d739168 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -4158,14 +4158,19 @@ paths: required: true type: integer responses: - "200": - description: OK + "202": + description: Accepted + schema: + additionalProperties: true + type: object + "409": + description: 任务正在执行中 schema: additionalProperties: true type: object security: - BearerAuth: [] - summary: 立即执行任务 + summary: 立即执行任务(异步触发,结果经任务日志轮询获取) tags: - 任务与日志回传 /api/v1/webhooks/oci-logs/{secret}: diff --git a/internal/api/task.go b/internal/api/task.go index c02ef84..599e265 100644 --- a/internal/api/task.go +++ b/internal/api/task.go @@ -2,6 +2,7 @@ package api import ( "encoding/json" + "errors" "net/http" "strconv" @@ -155,10 +156,11 @@ func (h *taskHandler) logs(c *gin.Context) { c.JSON(http.StatusOK, logs) } -// @Summary 立即执行任务 +// @Summary 立即执行任务(异步触发,结果经任务日志轮询获取) // @Tags 任务与日志回传 // @Param id path int true "配置 ID" -// @Success 200 {object} map[string]any +// @Success 202 {object} map[string]any +// @Failure 409 {object} map[string]any "任务正在执行中" // @Security BearerAuth // @Router /api/v1/tasks/{id}/run [post] func (h *taskHandler) run(c *gin.Context) { @@ -166,10 +168,13 @@ func (h *taskHandler) run(c *gin.Context) { if !ok { return } - entry, err := h.svc.RunTaskNow(c.Request.Context(), id) - if err != nil { + if err := h.svc.TriggerTask(c.Request.Context(), id); err != nil { + if errors.Is(err, service.ErrTaskRunning) { + c.JSON(http.StatusConflict, gin.H{"error": err.Error()}) + return + } respondError(c, err) return } - c.JSON(http.StatusOK, entry) + c.JSON(http.StatusAccepted, gin.H{"triggered": true}) } diff --git a/internal/model/models.go b/internal/model/models.go index 42bd36c..11f038e 100644 --- a/internal/model/models.go +++ b/internal/model/models.go @@ -309,6 +309,13 @@ type AiModelCache struct { // 弃用后仍可调用,直到 RetiredAt(按需推理退役,同步层已剔除过期项)。 DeprecatedAt *time.Time `json:"deprecatedAt"` RetiredAt *time.Time `json:"retiredAt"` + // Unusable 标记该模型在此渠道不可按需调用(微调基座/区域未开放按需), + // 由探测、真实调用或同步后的后台验证习得:不参与网关列表/路由/探测候选; + // 任何同步都按 OCID 结转标记,已标记模型定期复检、恢复供给自动解除。 + Unusable bool `json:"unusable"` + UnusableReason string `gorm:"size:200" json:"unusableReason,omitempty"` + // CheckedAt 是最近一次可用性验证时间;NULL 表示从未验证(同步后的后台验证以此为队列)。 + CheckedAt *time.Time `json:"checkedAt"` } // AiCallLog 是网关调用日志:仅元数据与 token 用量,绝不记录 prompt / 响应正文。 diff --git a/internal/oci/errors.go b/internal/oci/errors.go index 342e9da..ec45c41 100644 --- a/internal/oci/errors.go +++ b/internal/oci/errors.go @@ -76,3 +76,34 @@ func ServiceStatus(err error) (int, bool) { } return 0, false } + +// IsOnDemandUnsupported 识别「模型在该区域仅为微调基座、不支持按需调用」的 400: +// OCI 消息形如 "Not allowed to call finetune base model …, use Endpoint: false"。 +// ListModels 无字段可事先区分,只能在调用报错时识别并换渠道。 +func IsOnDemandUnsupported(err error) bool { + var svcErr common.ServiceError + if !errors.As(err, &svcErr) { + return false + } + return svcErr.GetHTTPStatusCode() == 400 && + strings.Contains(strings.ToLower(svcErr.GetMessage()), "finetune base model") +} + +// IsEntityNotFound 识别「实体不存在」404(消息 "Entity with key … not found"): +// GenAI 对区域内无按需供给的模型 OCID 返回此类 404,属模型级错误; +// 鉴权失败的 404 是 NotAuthorizedOrNotFound(消息为 Authorization failed…),不在此列。 +func IsEntityNotFound(err error) bool { + var svcErr common.ServiceError + if !errors.As(err, &svcErr) { + return false + } + msg := strings.ToLower(svcErr.GetMessage()) + return svcErr.GetHTTPStatusCode() == 404 && + strings.Contains(msg, "entity with key") && strings.Contains(msg, "not found") +} + +// IsModelUnavailable 归并「模型×区域不可按需调用」两类报错:微调基座 400 与实体不存在 404; +// 命中即应把该 (渠道, 模型) 从池中剔除并换候选/换渠道,而非定论租户配额问题。 +func IsModelUnavailable(err error) bool { + return IsOnDemandUnsupported(err) || IsEntityNotFound(err) +} diff --git a/internal/oci/errors_test.go b/internal/oci/errors_test.go index c261621..739a1aa 100644 --- a/internal/oci/errors_test.go +++ b/internal/oci/errors_test.go @@ -109,3 +109,52 @@ func TestErrorHint(t *testing.T) { }) } } + +func TestIsOnDemandUnsupported(t *testing.T) { + ft := fakeServiceError{status: 400, code: "InvalidParameter", + message: "Not allowed to call finetune base model ocid1.generativeaimodel.oc1.eu-frankfurt-1.tpel5q, use Endpoint: false"} + tests := []struct { + name string + err error + want bool + }{ + {"微调基座 400 命中(含包装链)", fmt.Errorf("genai chat: %w", ft), true}, + {"同消息但非 400 不命中", fakeServiceError{status: 500, code: "InternalError", message: ft.message}, false}, + {"普通 400 不命中", fakeServiceError{status: 400, code: "InvalidParameter", message: "bad request"}, false}, + {"非服务端错误不命中", errors.New("finetune base model"), false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := IsOnDemandUnsupported(tt.err); got != tt.want { + t.Errorf("IsOnDemandUnsupported() = %v, want %v", got, tt.want) + } + }) + } +} + +func TestIsModelUnavailable(t *testing.T) { + entity404 := fakeServiceError{status: 404, code: "NotFound", + message: "Entity with key ocid1.generativeaimodel.oc1.eu-frankfurt-1.2flsfq not found"} + auth404 := fakeServiceError{status: 404, code: "NotAuthorizedOrNotFound", + message: "Authorization failed or requested resource not found."} + ft400 := fakeServiceError{status: 400, code: "InvalidParameter", + message: "Not allowed to call finetune base model ocid1.generativeaimodel.oc1..x, use Endpoint: false"} + tests := []struct { + name string + err error + want bool + }{ + {"实体不存在 404 命中(含包装链)", fmt.Errorf("genai chat: %w", entity404), true}, + {"鉴权类 404 不命中(仍属租户级)", auth404, false}, + {"微调基座 400 命中", ft400, true}, + {"其他 404 消息不命中", fakeServiceError{status: 404, code: "NotFound", message: "route not found"}, false}, + {"非服务端错误不命中", errors.New("entity with key x not found"), false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := IsModelUnavailable(tt.err); got != tt.want { + t.Errorf("IsModelUnavailable() = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/internal/oci/genai.go b/internal/oci/genai.go index 60788a6..81d77e2 100644 --- a/internal/oci/genai.go +++ b/internal/oci/genai.go @@ -17,11 +17,10 @@ import ( // GenAiModel 是区域可用基础模型的摘要(管理面 ListModels)。 type GenAiModel struct { - Ocid string `json:"ocid"` - Name string `json:"name"` - Vendor string `json:"vendor"` - ChatOnly bool `json:"-"` - Caps []string `json:"capabilities"` + Ocid string `json:"ocid"` + Name string `json:"name"` + Vendor string `json:"vendor"` + Caps []string `json:"capabilities"` // Capability 是网关侧归一能力:CHAT / EMBEDDING(兼具时算 CHAT) Capability string `json:"capability"` // Deprecated 是 OCI 宣布的弃用时间(TimeDeprecated),nil 表示未宣布; @@ -71,18 +70,39 @@ func (c *RealClient) ListGenAiModels(ctx context.Context, cred Credentials, regi if err != nil { return nil, fmt.Errorf("list genai models: %w", err) } - seen := map[string]bool{} - now := time.Now() + return dedupGenAiModels(resp.Items, time.Now()), nil +} + +// dedupGenAiModels 压平并按名称去重;同名多条目时优先保留不含 FINE_TUNE 能力的条目 +// (微调基座条目在部分区域不支持按需调用,缓存其 OCID 会导致调用 400)。 +func dedupGenAiModels(items []generativeai.ModelSummary, now time.Time) []GenAiModel { + seen := map[string]int{} var out []GenAiModel - for _, m := range resp.Items { + for _, m := range items { gm, ok := toGenAiModel(m, now) - if !ok || seen[gm.Name] { + if !ok { continue } - seen[gm.Name] = true + if i, dup := seen[gm.Name]; dup { + if hasFineTune(out[i].Caps) && !hasFineTune(gm.Caps) { + out[i] = gm + } + continue + } + seen[gm.Name] = len(out) out = append(out, gm) } - return out, nil + return out +} + +// hasFineTune 判断能力列表是否含 FINE_TUNE(微调基座条目)。 +func hasFineTune(caps []string) bool { + for _, c := range caps { + if c == string(generativeai.ModelCapabilityFineTune) { + return true + } + } + return false } // toGenAiModel 压平模型摘要;无归一能力、无名称或按需推理已退役 diff --git a/internal/oci/genai_retire_test.go b/internal/oci/genai_retire_test.go index 0ec4739..92ae05c 100644 --- a/internal/oci/genai_retire_test.go +++ b/internal/oci/genai_retire_test.go @@ -38,3 +38,33 @@ func TestToGenAiModelRetiredFilter(t *testing.T) { t.Error("正常模型应入池") } } + +func TestDedupGenAiModelsFineTunePreference(t *testing.T) { + now := time.Date(2026, 7, 10, 0, 0, 0, 0, time.UTC) + chat := []generativeai.ModelCapabilityEnum{generativeai.ModelCapabilityChat} + chatFT := []generativeai.ModelCapabilityEnum{generativeai.ModelCapabilityChat, generativeai.ModelCapabilityFineTune} + mk := func(id, name string, caps []generativeai.ModelCapabilityEnum) generativeai.ModelSummary { + return generativeai.ModelSummary{ + Id: common.String(id), DisplayName: common.String(name), Vendor: common.String("meta"), + Capabilities: caps, LifecycleState: generativeai.ModelLifecycleStateActive, + } + } + const name = "meta.llama-3-70b-instruct" + tests := []struct { + name string + items []generativeai.ModelSummary + want string // 期望保留的 OCID + }{ + {"基座在前、纯对话在后:保留纯对话条目", []generativeai.ModelSummary{mk("o-ft", name, chatFT), mk("o-od", name, chat)}, "o-od"}, + {"纯对话在前、基座在后:保留纯对话条目", []generativeai.ModelSummary{mk("o-od", name, chat), mk("o-ft", name, chatFT)}, "o-od"}, + {"仅基座条目:保留不误删", []generativeai.ModelSummary{mk("o-ft", name, chatFT)}, "o-ft"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := dedupGenAiModels(tt.items, now) + if len(got) != 1 || got[0].Ocid != tt.want { + t.Errorf("dedupGenAiModels() = %+v, want 仅保留 %s", got, tt.want) + } + }) + } +} diff --git a/internal/service/aigateway.go b/internal/service/aigateway.go index beacd98..6ea075c 100644 --- a/internal/service/aigateway.go +++ b/internal/service/aigateway.go @@ -30,6 +30,10 @@ const ( aiBackoffCap = 30 * time.Minute // aiKeyTouchGap 是 LastUsedAt 的最小写库间隔,避免高频调用刷库 aiKeyTouchGap = time.Minute + // modelRecheckGap 是已标记不可用模型的复检间隔(恢复供给自动解除标记); + // validateBatchCap 限制单渠道单轮验证的试调次数 + modelRecheckGap = 20 * time.Hour + validateBatchCap = 32 // 内容日志(红线例外)约束:开启必须限时(上限 7 天),正文截断,短保留 aiContentLogMaxHours = 168 aiContentLogRetention = 7 * 24 * time.Hour @@ -55,13 +59,17 @@ type AiGatewayService struct { // touchMu 保护各密钥的最近触达时间(内存节流,不追求跨实例精确) touchMu sync.Mutex lastTouch map[uint]time.Time + // validateMu 保护 validating:同一渠道的模型验证不并发 + validateMu sync.Mutex + validating map[uint]bool // onChannelsChanged 在渠道增删后触发,由 main 装配为探测任务同步钩子 onChannelsChanged func(context.Context) } // NewAiGatewayService 组装依赖;调用 StartCleanup 后开始调用日志周期清理。 func NewAiGatewayService(db *gorm.DB, configs *OciConfigService, client oci.Client) *AiGatewayService { - return &AiGatewayService{db: db, configs: configs, client: client, lastTouch: map[uint]time.Time{}} + return &AiGatewayService{db: db, configs: configs, client: client, + lastTouch: map[uint]time.Time{}, validating: map[uint]bool{}} } // SetOnChannelsChanged 注册渠道数量变化钩子(渠道创建/删除成功后调用)。 @@ -290,7 +298,7 @@ func (s *AiGatewayService) ProbeChannel(ctx context.Context, id uint) (*model.Ai return &fresh, nil } -// probe 执行探测并返回 (状态, 错误摘要);同时完成模型缓存同步。 +// probe 执行探测并返回 (状态, 错误摘要);同时完成模型缓存同步(保留不可用标记)。 func (s *AiGatewayService) probe(ctx context.Context, cred oci.Credentials, ch *model.AiChannel) (string, string) { models, err := s.client.ListGenAiModels(ctx, cred, ch.Region) if err != nil { @@ -303,43 +311,209 @@ func (s *AiGatewayService) probe(ctx context.Context, cred oci.Credentials, ch * if err := s.replaceModels(ctx, ch.ID, models); err != nil { return "error", truncateErr(err.Error()) } - return s.probeChat(ctx, cred, ch, models) + usable, err := s.usableModels(ctx, ch.ID, models) + if err != nil { + return "error", truncateErr(err.Error()) + } + return s.probeChat(ctx, cred, ch, usable) } -// probeChat 按偏好挑选至多 3 个模型依次试调:部分模型元数据标 CHAT 但实际 -// 不可对话(如 voice agent),遇 400/5xx 换下一个;401/403/404 属租户级直接定论。 +// usableModels 过滤掉缓存中已标记不可按需调用的模型,探测候选不再反复踩坑。 +func (s *AiGatewayService) usableModels(ctx context.Context, channelID uint, models []oci.GenAiModel) ([]oci.GenAiModel, error) { + marks, err := s.loadModelMarks(ctx, channelID) + if err != nil { + return nil, err + } + out := make([]oci.GenAiModel, 0, len(models)) + for _, m := range models { + if !marks[m.Ocid].Unusable { + out = append(out, m) + } + } + return out, nil +} + +// probeChat 按候选顺序试调(上限 8):遇「模型不可按需调用」标记剔除并换下一个; +// 401/403 与鉴权类 404 属租户级直接定论 no_quota;其余错误(部分模型元数据标 CHAT +// 但实际不可对话,如 voice agent)累计 3 次止损。 func (s *AiGatewayService) probeChat(ctx context.Context, cred oci.Credentials, ch *model.AiChannel, models []oci.GenAiModel) (string, string) { status, detail := "error", "无可试调对话模型" + errBudget := 3 for _, m := range probeCandidates(models) { code, err := s.client.GenAiProbeChat(ctx, cred, ch.Region, m.Ocid, m.Name) switch { case code == 200 || code == 429: return "ok", "" + case oci.IsModelUnavailable(err): + s.markModelUnusable(ctx, ch.ID, m.Ocid, oci.CompactError(err)) + status, detail = "error", truncateErr(fmt.Sprintf("%s: 不可按需调用,已从模型池剔除", m.Name)) case code == 401 || code == 403 || code == 404: return "no_quota", truncateErr(oci.CompactError(err)) default: status, detail = "error", truncateErr(fmt.Sprintf("%s: %s", m.Name, oci.CompactError(err))) + if errBudget--; errBudget == 0 { + return status, detail + } } } return status, detail } -// probeCandidates 只取对话模型并按可靠度排序取前 3:主流文本模型优先, -// voice 等非常规形态殿后(embedding / rerank 已被能力筛选排除)。 +// markModelUnusable 把 (渠道, 模型OCID) 标记为不可按需调用;失败仅记日志不阻断主流程。 +func (s *AiGatewayService) markModelUnusable(ctx context.Context, channelID uint, ocid, reason string) { + err := s.db.WithContext(ctx).Model(&model.AiModelCache{}). + Where("channel_id = ? AND model_ocid = ?", channelID, ocid). + Updates(map[string]any{"unusable": true, "unusable_reason": shortReason(reason), "checked_at": time.Now()}).Error + if err != nil { + log.Printf("mark model unusable: %v", err) + } +} + +func shortReason(reason string) string { + if len(reason) > 200 { + return reason[:200] + } + return reason +} + +// beginValidate 抢占渠道的验证执行权,同渠道同时只跑一轮。 +func (s *AiGatewayService) beginValidate(id uint) bool { + s.validateMu.Lock() + defer s.validateMu.Unlock() + if s.validating[id] { + return false + } + s.validating[id] = true + return true +} + +func (s *AiGatewayService) endValidate(id uint) { + s.validateMu.Lock() + delete(s.validating, id) + s.validateMu.Unlock() +} + +// validateTargets 取待验证行:从未验证的新模型,以及标记超过复检间隔的模型(仅对话能力, +// 兼容存量空串);单轮上限 validateBatchCap 控制调用量。 +func (s *AiGatewayService) validateTargets(ctx context.Context, channelID uint) ([]model.AiModelCache, error) { + stale := time.Now().Add(-modelRecheckGap) + var rows []model.AiModelCache + err := s.db.WithContext(ctx). + Where("channel_id = ? AND capability IN ?", channelID, []string{"CHAT", ""}). + Where("checked_at IS NULL OR (unusable = ? AND checked_at < ?)", true, stale). + Order("id ASC").Limit(validateBatchCap).Find(&rows).Error + return rows, err +} + +// validateChannelModels 逐个试调渠道内待验证模型并落结论,把不可按需调用的模型 +// 从池中剔除、把恢复供给的解除标记;ListModels 无字段可事先判别,只能试调习得。 +func (s *AiGatewayService) validateChannelModels(ctx context.Context, channelID uint) { + if !s.beginValidate(channelID) { + return + } + defer s.endValidate(channelID) + var ch model.AiChannel + if err := s.db.WithContext(ctx).First(&ch, channelID).Error; err != nil { + return + } + cred, err := s.configs.credentialsByID(ctx, ch.OciConfigID) + if err != nil { + return + } + rows, err := s.validateTargets(ctx, channelID) + if err != nil { + log.Printf("validate channel %d models: %v", channelID, err) + return + } + for _, r := range rows { + s.validateOne(ctx, cred, &ch, r) + } +} + +// validateOne 试调单个模型并落库:可用(200/429)清除标记;模型级不可用标记剔除; +// 其余 4xx 记录已检不改状态;5xx/网络类瞬态不落 checked_at,下轮再验。 +func (s *AiGatewayService) validateOne(ctx context.Context, cred oci.Credentials, ch *model.AiChannel, r model.AiModelCache) { + code, err := s.client.GenAiProbeChat(ctx, cred, ch.Region, r.ModelOcid, r.Name) + updates := map[string]any{"checked_at": time.Now()} + switch { + case code == 200 || code == 429: + updates["unusable"], updates["unusable_reason"] = false, "" + case oci.IsModelUnavailable(err): + updates["unusable"], updates["unusable_reason"] = true, shortReason(oci.CompactError(err)) + case code == 0 || code >= 500: + return + } + s.db.WithContext(ctx).Model(&model.AiModelCache{}).Where("id = ?", r.ID).Updates(updates) +} + +// validateModelsAsync 后台验证渠道模型池(手动同步后触发,不阻塞接口响应)。 +func (s *AiGatewayService) validateModelsAsync(ctx context.Context, channelID uint) { + s.wg.Add(1) + go func() { + defer s.wg.Done() + vctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Minute) + defer cancel() + s.validateChannelModels(vctx, channelID) + }() +} + +// probeCandidateCap 是单次探测的候选上限:模型级不可用会被标记剔除不再重试, +// 放宽到 8 让一次探测有机会越过整批坏模型找到可用者;其他错误另有 3 次止损预算。 +const probeCandidateCap = 8 + +// probeCandidates 只取对话模型,按可靠度排序后跨厂商取候选:主流文本模型优先; +// voice 等负分形态(元数据标 CHAT 但实际不可对话)直接排除,不浪费试调预算; +// 每厂商先取最高分再按分数补位——部分区域某厂商全为微调基座(调用必失败), +// 不能让单一厂商占满候选名额拖垮整个渠道的探测结论。 func probeCandidates(models []oci.GenAiModel) []oci.GenAiModel { var sorted []oci.GenAiModel for _, m := range models { - if m.Capability == "" || m.Capability == "CHAT" { + if (m.Capability == "" || m.Capability == "CHAT") && probeScore(m.Name) >= 0 { sorted = append(sorted, m) } } sort.SliceStable(sorted, func(i, j int) bool { return probeScore(sorted[i].Name) > probeScore(sorted[j].Name) }) - if len(sorted) > 3 { - sorted = sorted[:3] + return diversifyByVendor(sorted, probeCandidateCap) +} + +// diversifyByVendor 从已排序列表先每厂商各取一个,不足 limit 再按原序补位。 +func diversifyByVendor(sorted []oci.GenAiModel, limit int) []oci.GenAiModel { + picked := make([]oci.GenAiModel, 0, limit) + used := make(map[int]bool) + seenVendor := make(map[string]bool) + for i, m := range sorted { + if len(picked) >= limit { + break + } + if v := modelVendor(m); !seenVendor[v] { + seenVendor[v] = true + used[i] = true + picked = append(picked, m) + } } - return sorted + for i, m := range sorted { + if len(picked) >= limit { + break + } + if !used[i] { + picked = append(picked, m) + } + } + return picked +} + +// modelVendor 取厂商标识;OCI 未回填 vendor 时退化为模型名「.」前缀。 +func modelVendor(m oci.GenAiModel) string { + if m.Vendor != "" { + return strings.ToLower(m.Vendor) + } + name := strings.ToLower(m.Name) + if i := strings.Index(name, "."); i > 0 { + return name[:i] + } + return name } func probeScore(name string) int { @@ -377,7 +551,8 @@ func truncateErr(msg string) string { return msg } -// SyncModels 重新拉取渠道区域的模型列表并覆盖缓存。 +// SyncModels 重新拉取渠道区域的模型列表并覆盖缓存(标记按 OCID 结转), +// 随后触发后台验证:新模型逐个试调,不可按需调用的数十秒内从池中剔除。 func (s *AiGatewayService) SyncModels(ctx context.Context, id uint) ([]model.AiModelCache, error) { var ch model.AiChannel if err := s.db.WithContext(ctx).First(&ch, id).Error; err != nil { @@ -394,16 +569,26 @@ func (s *AiGatewayService) SyncModels(ctx context.Context, id uint) ([]model.AiM if err := s.replaceModels(ctx, id, models); err != nil { return nil, err } + s.validateModelsAsync(ctx, id) return s.channelModels(ctx, id) } -// replaceModels 以事务整组覆盖渠道模型缓存。 +// replaceModels 以事务整组覆盖渠道模型缓存,按 OCID 结转不可用标记与验证时间 +// (OCID 变化视为新条目,自然回到待验证状态)。 func (s *AiGatewayService) replaceModels(ctx context.Context, channelID uint, models []oci.GenAiModel) error { + marks, err := s.loadModelMarks(ctx, channelID) + if err != nil { + return err + } rows := make([]model.AiModelCache, 0, len(models)) now := time.Now() for _, m := range models { - rows = append(rows, model.AiModelCache{ChannelID: channelID, ModelOcid: m.Ocid, Name: m.Name, Vendor: m.Vendor, - Capability: m.Capability, SyncedAt: now, DeprecatedAt: m.Deprecated, RetiredAt: m.Retired}) + row := model.AiModelCache{ChannelID: channelID, ModelOcid: m.Ocid, Name: m.Name, Vendor: m.Vendor, + Capability: m.Capability, SyncedAt: now, DeprecatedAt: m.Deprecated, RetiredAt: m.Retired} + if prev, ok := marks[m.Ocid]; ok { + row.Unusable, row.UnusableReason, row.CheckedAt = prev.Unusable, prev.UnusableReason, prev.CheckedAt + } + rows = append(rows, row) } return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { if err := tx.Where("channel_id = ?", channelID).Delete(&model.AiModelCache{}).Error; err != nil { @@ -416,17 +601,33 @@ func (s *AiGatewayService) replaceModels(ctx context.Context, channelID uint, mo }) } +// loadModelMarks 取渠道内带标记或已验证过的行(OCID → 行),供同步结转与候选过滤。 +func (s *AiGatewayService) loadModelMarks(ctx context.Context, channelID uint) (map[string]model.AiModelCache, error) { + var rows []model.AiModelCache + err := s.db.WithContext(ctx).Select("model_ocid", "unusable", "unusable_reason", "checked_at"). + Where("channel_id = ? AND (unusable = ? OR checked_at IS NOT NULL)", channelID, true).Find(&rows).Error + if err != nil { + return nil, fmt.Errorf("load model marks: %w", err) + } + marks := make(map[string]model.AiModelCache, len(rows)) + for _, r := range rows { + marks[r.ModelOcid] = r + } + return marks, nil +} + func (s *AiGatewayService) channelModels(ctx context.Context, channelID uint) ([]model.AiModelCache, error) { var rows []model.AiModelCache err := s.db.WithContext(ctx).Where("channel_id = ?", channelID).Order("name ASC").Find(&rows).Error return rows, err } -// GatewayModels 聚合启用渠道的模型(按名称去重),供 /ai/v1/models; -// group 非空时仅聚合该分组渠道(与密钥分组路由口径一致)。 +// GatewayModels 聚合启用渠道的可用模型(按名称去重,剔除不可按需调用标记), +// 供 /ai/v1/models;group 非空时仅聚合该分组渠道(与密钥分组路由口径一致)。 func (s *AiGatewayService) GatewayModels(ctx context.Context, group string) (aiwire.ModelList, error) { q := s.db.WithContext(ctx). - Joins("JOIN ai_channels ON ai_channels.id = ai_model_caches.channel_id AND ai_channels.enabled = ?", true) + Joins("JOIN ai_channels ON ai_channels.id = ai_model_caches.channel_id AND ai_channels.enabled = ?", true). + Where("(ai_model_caches.unusable = ? OR ai_model_caches.unusable IS NULL)", false) if group != "" { q = q.Where("ai_channels.channel_group = ?", group) } @@ -457,6 +658,7 @@ func (s *AiGatewayService) DeprecatingModels(ctx context.Context, within time.Du err := s.db.WithContext(ctx). Where("(retired_at IS NOT NULL AND retired_at > ? AND retired_at <= ?) OR (deprecated_at IS NOT NULL AND deprecated_at >= ? AND deprecated_at <= ?)", now, deadline, now, deadline). + Where("(unusable = ? OR unusable IS NULL)", false). Order("name ASC").Find(&rows).Error if err != nil { return nil, err @@ -477,7 +679,7 @@ func (s *AiGatewayService) DeprecatingModels(ctx context.Context, within time.Du return out, nil } -// ProbeAll 逐个探测全部渠道,返回状态汇总;供 AI 探测后台任务调用。 +// ProbeAll 逐个探测全部渠道并顺带验证模型池,返回状态汇总;供 AI 探测后台任务调用。 func (s *AiGatewayService) ProbeAll(ctx context.Context) (string, error) { chs, err := s.Channels(ctx) if err != nil { @@ -493,6 +695,8 @@ func (s *AiGatewayService) ProbeAll(ctx context.Context) (string, error) { continue } counts[fresh.ProbeStatus]++ + // 后台任务里同步执行:新模型验证入池、坏模型定期复检,零人工收敛 + s.validateChannelModels(ctx, ch.ID) } msg := fmt.Sprintf("probed %d: %d ok, %d no_service, %d no_quota, %d error", len(chs), counts["ok"], counts["no_service"], counts["no_quota"], counts["error"]) diff --git a/internal/service/aigateway_chat.go b/internal/service/aigateway_chat.go index b3179cb..28aea9c 100644 --- a/internal/service/aigateway_chat.go +++ b/internal/service/aigateway_chat.go @@ -43,10 +43,13 @@ func (s *AiGatewayService) Chat(ctx context.Context, ir aiwire.ChatRequest, grou s.markSuccess(ctx, cand.ch.ID) return resp, meta, nil } - if !retryable(err) { + retry, penalize := s.noteCallErr(ctx, cand, err) + if !retry { return nil, meta, err } - s.markFailure(ctx, cand.ch.ID) + if penalize { + s.markFailure(ctx, cand.ch.ID) + } excluded[cand.ch.ID] = true meta.Retries++ lastErr = err @@ -83,10 +86,13 @@ func (s *AiGatewayService) OpenStream(ctx context.Context, ir aiwire.ChatRequest s.markSuccess(ctx, cand.ch.ID) return stream, meta, nil } - if !retryable(err) { + retry, penalize := s.noteCallErr(ctx, cand, err) + if !retry { return nil, meta, err } - s.markFailure(ctx, cand.ch.ID) + if penalize { + s.markFailure(ctx, cand.ch.ID) + } excluded[cand.ch.ID] = true meta.Retries++ lastErr = err @@ -102,12 +108,18 @@ func firstErr(lastErr, pickErr error) error { return pickErr } -// retryable 判定是否换渠道重试:429 / 5xx / 网络错误可重试,其余 4xx 直接透传。 -func retryable(err error) bool { - if status, ok := oci.ServiceStatus(err); ok { - return status == 429 || status >= 500 +// noteCallErr 汇总一次调用失败:模型级不可用(微调基座 400 / 实体不存在 404)先标记 +// 剔除该 (渠道, 模型),换渠道重试且不计熔断——这是模型×区域供给问题而非渠道健康问题; +// 429 / 5xx / 网络错误换渠道并计熔断;其余 4xx 直接透传。 +func (s *AiGatewayService) noteCallErr(ctx context.Context, cand *aiCandidate, err error) (retry, penalize bool) { + if oci.IsModelUnavailable(err) { + s.markModelUnusable(ctx, cand.ch.ID, cand.modelOcid, oci.CompactError(err)) + return true, false } - return true + if status, ok := oci.ServiceStatus(err); ok { + return status == 429 || status >= 500, true + } + return true, true } // pick 选出支持该模型的最优渠道:能力匹配 → 启用 → 分组匹配 → 未熔断 → 最小优先级组 → 加权随机。 @@ -142,10 +154,11 @@ func (s *AiGatewayService) pick(ctx context.Context, modelName, group, capabilit return &aiCandidate{ch: chosen, modelOcid: ocids[chosen.ID]}, nil } -// modelChannels 查出提供该模型的渠道 ID 及各自的模型 OCID; -// capability=CHAT 时兼容存量空串(加列前只同步对话模型)。 +// modelChannels 查出提供该模型的渠道 ID 及各自的模型 OCID(剔除不可按需调用标记); +// capability=CHAT 时兼容存量空串(加列前只同步对话模型);unusable 需兼容 NULL +// (AutoMigrate 加列后、首次重同步前的存量行)。 func (s *AiGatewayService) modelChannels(ctx context.Context, modelName, capability string) (map[uint]string, []uint, error) { - q := s.db.WithContext(ctx).Where("name = ?", modelName) + q := s.db.WithContext(ctx).Where("name = ? AND (unusable = ? OR unusable IS NULL)", modelName, false) if capability == "CHAT" { q = q.Where("capability IN ?", []string{"CHAT", ""}) } else { @@ -221,10 +234,13 @@ func (s *AiGatewayService) Embeddings(ctx context.Context, req aiwire.Embeddings s.markSuccess(ctx, cand.ch.ID) return resp, meta, nil } - if !retryable(err) { + retry, penalize := s.noteCallErr(ctx, cand, err) + if !retry { return nil, meta, err } - s.markFailure(ctx, cand.ch.ID) + if penalize { + s.markFailure(ctx, cand.ch.ID) + } excluded[cand.ch.ID] = true meta.Retries++ lastErr = err diff --git a/internal/service/aigateway_test.go b/internal/service/aigateway_test.go index 65e70f1..ba98d74 100644 --- a/internal/service/aigateway_test.go +++ b/internal/service/aigateway_test.go @@ -3,6 +3,7 @@ package service import ( "context" "errors" + "fmt" "strings" "testing" "time" @@ -13,14 +14,28 @@ import ( ) // stubServiceError 实现 common.ServiceError,用于模拟带状态码的 OCI 服务端错误。 -type stubServiceError struct{ status int } +type stubServiceError struct { + status int + msg string +} -func (e stubServiceError) Error() string { return "stub service error" } -func (e stubServiceError) GetHTTPStatusCode() int { return e.status } -func (e stubServiceError) GetMessage() string { return "stub" } +func (e stubServiceError) Error() string { return "stub service error" } +func (e stubServiceError) GetHTTPStatusCode() int { return e.status } +func (e stubServiceError) GetMessage() string { + if e.msg != "" { + return e.msg + } + return "stub" +} func (e stubServiceError) GetCode() string { return "Stub" } func (e stubServiceError) GetOpcRequestID() string { return "req-1" } +// finetuneBaseErr 模拟「微调基座模型不可按需调用」的 OCI 400。 +func finetuneBaseErr() stubServiceError { + return stubServiceError{status: 400, + msg: "Not allowed to call finetune base model ocid1.generativeaimodel.oc1.eu-frankfurt-1.tpel5q, use Endpoint: false"} +} + // gatewayStubClient 覆写 GenAI 四方法;chatErrs 逐次弹出以模拟先失败后成功。 type gatewayStubClient struct { *fakeClient @@ -29,6 +44,8 @@ type gatewayStubClient struct { modelsErr error probeCode int probeErr error + // probeSeq 非空时逐次弹出,模拟按候选依次试调;弹尽后回落 probeCode/probeErr + probeSeq []probeResult chatResp *aiwire.ChatResponse chatErrs []error chatCalls int @@ -47,7 +64,18 @@ func (f *gatewayStubClient) ListGenAiModels(ctx context.Context, cred oci.Creden return f.models, f.modelsErr } +// probeResult 是 gatewayStubClient.probeSeq 的单次探测结果。 +type probeResult struct { + code int + err error +} + func (f *gatewayStubClient) GenAiProbeChat(ctx context.Context, cred oci.Credentials, region, modelOcid, modelName string) (int, error) { + if len(f.probeSeq) > 0 { + r := f.probeSeq[0] + f.probeSeq = f.probeSeq[1:] + return r.code, r.err + } return f.probeCode, f.probeErr } @@ -400,6 +428,263 @@ func TestProbeCandidates(t *testing.T) { } } +func TestProbeCandidatesVendorDiversity(t *testing.T) { + // 部分区域单一厂商全为微调基座:候选须跨厂商分散,不能被 3 个 llama 占满前排 + models := []oci.GenAiModel{ + {Ocid: "l1", Name: "meta.llama-3-70b-instruct"}, + {Ocid: "l2", Name: "meta.llama-3.1-405b-instruct"}, + {Ocid: "l3", Name: "meta.llama-3.3-70b-instruct"}, + {Ocid: "c1", Name: "cohere.command-a-03-2025"}, + {Ocid: "g1", Name: "xai.grok-4"}, + } + got := probeCandidates(models) + if len(got) != 5 || got[0].Name != "meta.llama-3-70b-instruct" { + t.Fatalf("上限内全量返回且最高分居首: %+v", got) + } + vendors := map[string]bool{} + for _, m := range got[:3] { + vendors[modelVendor(m)] = true + } + if len(vendors) != 3 { + t.Errorf("前 3 个候选应覆盖 3 个厂商: %+v", got) + } + // 超过上限时截断到 probeCandidateCap + var many []oci.GenAiModel + for i := 0; i < 12; i++ { + many = append(many, oci.GenAiModel{Ocid: fmt.Sprintf("m%d", i), Name: fmt.Sprintf("meta.llama-%d", i)}) + } + if capped := probeCandidates(many); len(capped) != probeCandidateCap { + t.Errorf("候选应截断到 %d: got %d", probeCandidateCap, len(capped)) + } +} + +// entityNotFoundErr 模拟「实体不存在」404(模型在区域内无按需供给)。 +func entityNotFoundErr() stubServiceError { + return stubServiceError{status: 404, + msg: "Entity with key ocid1.generativeaimodel.oc1.eu-frankfurt-1.2flsfq not found"} +} + +func TestProbeSkipsUnavailableModels(t *testing.T) { + // 首选 llama 不可按需调用(基座 400 / 实体 404)→ 标记剔除 → 换候选成功 → 渠道判可用 + tests := []struct { + name string + bad probeResult + }{ + {"微调基座 400", probeResult{400, finetuneBaseErr()}}, + {"实体不存在 404", probeResult{404, entityNotFoundErr()}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client := &gatewayStubClient{ + fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}, + models: []oci.GenAiModel{ + {Ocid: "m1", Name: "meta.llama-3-70b-instruct", Vendor: "meta"}, + {Ocid: "m2", Name: "meta.llama-3.1-70b-instruct", Vendor: "meta"}, + {Ocid: "m3", Name: "cohere.command-a-03-2025", Vendor: "cohere"}, + }, + probeSeq: []probeResult{tt.bad, {200, nil}}, + } + gw, svc := newTestGateway(t, client) + cfg := importAliveConfig(t, svc) + ctx := context.Background() + + ch, err := gw.CreateChannel(ctx, ChannelInput{OciConfigID: cfg.ID, Region: "eu-frankfurt-1"}) + if err != nil { + t.Fatalf("CreateChannel: %v", err) + } + probed, err := gw.ProbeChannel(ctx, ch.ID) + if err != nil || probed.ProbeStatus != "ok" { + t.Fatalf("坏候选后应换候选并判可用: %+v, %v", probed, err) + } + var row model.AiModelCache + if err := gw.db.Where("channel_id = ? AND model_ocid = ?", ch.ID, "m1").First(&row).Error; err != nil || !row.Unusable { + t.Fatalf("首个坏候选应被标记不可用: %+v, %v", row, err) + } + // 再次探测:同步保留标记,m1 不再进候选(probeSeq 只需一次 200) + client.probeSeq = []probeResult{{200, nil}} + probed, err = gw.ProbeChannel(ctx, ch.ID) + if err != nil || probed.ProbeStatus != "ok" { + t.Fatalf("复测应跳过已标记模型: %+v, %v", probed, err) + } + var again model.AiModelCache + gw.db.Where("channel_id = ? AND model_ocid = ?", ch.ID, "m1").First(&again) + if !again.Unusable { + t.Error("探测触发的同步应保留不可用标记") + } + // 手动同步同样保留标记(坏模型不随重新同步复活);近期已检的标记不被后台验证翻转 + client.probeCode = 200 + if _, err := gw.SyncModels(ctx, ch.ID); err != nil { + t.Fatalf("SyncModels: %v", err) + } + gw.Wait() + var kept model.AiModelCache + gw.db.Where("channel_id = ? AND model_ocid = ?", ch.ID, "m1").First(&kept) + if !kept.Unusable { + t.Error("手动同步不应清除不可用标记") + } + }) + } +} + +func TestValidateModelsAfterSync(t *testing.T) { + // 同步后后台验证:坏模型标记剔除、好模型记录已检、其他 4xx 不改状态、非对话模型不试调 + client := &gatewayStubClient{ + fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}, + models: []oci.GenAiModel{ + {Ocid: "v1", Name: "meta.llama-4-maverick", Vendor: "meta", Capability: "CHAT"}, + {Ocid: "v2", Name: "xai.grok-4", Vendor: "xai", Capability: "CHAT"}, + {Ocid: "v3", Name: "xai.grok-voice-agent", Vendor: "xai", Capability: "CHAT"}, + {Ocid: "v4", Name: "cohere.embed-v4.0", Vendor: "cohere", Capability: "EMBEDDING"}, + }, + probeSeq: []probeResult{{404, entityNotFoundErr()}, {200, nil}, {400, stubServiceError{status: 400}}}, + } + gw, svc := newTestGateway(t, client) + cfg := importAliveConfig(t, svc) + ctx := context.Background() + ch, err := gw.CreateChannel(ctx, ChannelInput{OciConfigID: cfg.ID, Region: "us-ashburn-1"}) + if err != nil { + t.Fatalf("CreateChannel: %v", err) + } + if _, err := gw.SyncModels(ctx, ch.ID); err != nil { + t.Fatalf("SyncModels: %v", err) + } + gw.Wait() + + want := map[string]struct { + unusable bool + checked bool + }{ + "v1": {true, true}, // 实体 404 → 标记剔除 + "v2": {false, true}, // 200 → 可用已检 + "v3": {false, true}, // 普通 400 → 已检不标记 + "v4": {false, false}, // EMBEDDING 不试调 + } + rows, _ := gw.channelModels(ctx, ch.ID) + for _, r := range rows { + w := want[r.ModelOcid] + if r.Unusable != w.unusable || (r.CheckedAt != nil) != w.checked { + t.Errorf("%s: unusable=%v checked=%v, want %+v", r.ModelOcid, r.Unusable, r.CheckedAt != nil, w) + } + } + list, err := gw.GatewayModels(ctx, "") + if err != nil || len(list.Data) != 3 { + t.Errorf("网关列表应只剔除被标记的坏模型(余 v2/v3/v4), got %+v, %v", list.Data, err) + } +} + +func TestValidateRecheckUnmarksRecovered(t *testing.T) { + // 已标记模型超过复检间隔后重验:恢复供给(200)自动解除标记 + client := &gatewayStubClient{fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}, probeCode: 200} + gw, svc := newTestGateway(t, client) + cfg := importAliveConfig(t, svc) + ch := seedChannel(t, gw, cfg.ID, "eu-frankfurt-1", 1, 1) + ctx := context.Background() + old := time.Now().Add(-25 * time.Hour) + gw.db.Model(&model.AiModelCache{}).Where("channel_id = ?", ch.ID). + Updates(map[string]any{"unusable": true, "unusable_reason": "x", "checked_at": old}) + + gw.validateChannelModels(ctx, ch.ID) + var row model.AiModelCache + gw.db.Where("channel_id = ?", ch.ID).First(&row) + if row.Unusable || row.UnusableReason != "" || row.CheckedAt == nil || !row.CheckedAt.After(old) { + t.Errorf("超期复检应解除标记并刷新已检时间: %+v", row) + } + // 未超期的标记不复检(probeSeq 为空、fallback 200 也不会被消费) + fresh := time.Now() + gw.db.Model(&model.AiModelCache{}).Where("channel_id = ?", ch.ID). + Updates(map[string]any{"unusable": true, "checked_at": fresh}) + gw.validateChannelModels(ctx, ch.ID) + gw.db.Where("channel_id = ?", ch.ID).First(&row) + if !row.Unusable { + t.Error("未超期的标记不应被复检翻转") + } +} + +func TestProbeAuth404StillNoQuota(t *testing.T) { + // 鉴权类 404(NotAuthorizedOrNotFound)仍属租户级,直接定论 no_quota 且不标记模型 + client := &gatewayStubClient{ + fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}, + models: []oci.GenAiModel{{Ocid: "m1", Name: "meta.llama-3.3-70b-instruct", Vendor: "meta"}}, + probeCode: 404, + probeErr: stubServiceError{status: 404, msg: "Authorization failed or requested resource not found."}, + } + gw, svc := newTestGateway(t, client) + cfg := importAliveConfig(t, svc) + ctx := context.Background() + + ch, _ := gw.CreateChannel(ctx, ChannelInput{OciConfigID: cfg.ID, Region: "eu-frankfurt-1"}) + probed, _ := gw.ProbeChannel(ctx, ch.ID) + if probed.ProbeStatus != "no_quota" { + t.Errorf("鉴权 404 status = %q, want no_quota", probed.ProbeStatus) + } + var marked int64 + gw.db.Model(&model.AiModelCache{}).Where("unusable = ?", true).Count(&marked) + if marked != 0 { + t.Errorf("鉴权 404 不应标记模型, marked=%d", marked) + } +} + +func TestAiChatFinetuneSwitchesChannelWithoutPenalty(t *testing.T) { + // 微调基座 400 换渠道重试成功,且不计入熔断失败 + client := &gatewayStubClient{ + fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}, + chatResp: &aiwire.ChatResponse{Model: "meta.llama-3.3-70b-instruct", Choices: []aiwire.Choice{{Message: aiwire.ChatMessage{Role: "assistant", Content: aiwire.NewTextContent("hi")}, FinishReason: "stop"}}}, + chatErrs: []error{finetuneBaseErr()}, + } + gw, svc := newTestGateway(t, client) + cfg := importAliveConfig(t, svc) + seedChannel(t, gw, cfg.ID, "eu-frankfurt-1", 1, 1) + seedChannel(t, gw, cfg.ID, "us-chicago-1", 1, 1) + + resp, meta, err := gw.Chat(context.Background(), aiwire.ChatRequest{Model: "meta.llama-3.3-70b-instruct", Messages: []aiwire.ChatMessage{{Role: "user", Content: aiwire.NewTextContent("你好")}}}, "") + if err != nil || resp == nil { + t.Fatalf("Chat = %v, %v", resp, err) + } + if meta.Retries != 1 || client.chatCalls != 2 { + t.Errorf("应换渠道重试一次: retries=%d calls=%d", meta.Retries, client.chatCalls) + } + var chs []model.AiChannel + gw.db.Find(&chs) + for _, ch := range chs { + if ch.FailCount != 0 { + t.Errorf("微调基座 400 不应计入熔断: 渠道 %s failCount=%d", ch.Name, ch.FailCount) + } + } + // 失败渠道的该模型被标记,不再参与路由;成功渠道不受影响 + region := client.regions[0] + var row model.AiModelCache + gw.db.Where("model_ocid = ?", "ocid1..m-"+region).First(&row) + if !row.Unusable { + t.Errorf("失败渠道的模型应被标记不可用: %+v", row) + } + var usable int64 + gw.db.Model(&model.AiModelCache{}).Where("unusable = ?", false).Count(&usable) + if usable != 1 { + t.Errorf("成功渠道模型应保持可用, usable=%d", usable) + } +} + +func TestUnusableModelExcludedFromPoolAndRouting(t *testing.T) { + // 唯一渠道的模型被标记后:网关列表不再展示,路由按未知模型拒绝,渠道详情仍可见标记 + gw, svc := newTestGateway(t, &gatewayStubClient{fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}}) + cfg := importAliveConfig(t, svc) + ch := seedChannel(t, gw, cfg.ID, "eu-frankfurt-1", 1, 1) + ctx := context.Background() + + gw.markModelUnusable(ctx, ch.ID, "ocid1..m-eu-frankfurt-1", "Entity with key … not found") + list, err := gw.GatewayModels(ctx, "") + if err != nil || len(list.Data) != 0 { + t.Errorf("已标记模型不应出现在网关列表: %+v, %v", list.Data, err) + } + if _, _, err := gw.Chat(ctx, aiwire.ChatRequest{Model: "meta.llama-3.3-70b-instruct"}, ""); !errors.Is(err, ErrAiUnknownModel) { + t.Errorf("已标记模型路由应拒绝: %v", err) + } + rows, err := gw.channelModels(ctx, ch.ID) + if err != nil || len(rows) != 1 || !rows[0].Unusable || rows[0].UnusableReason == "" { + t.Errorf("渠道详情应保留标记行便于排查: %+v, %v", rows, err) + } +} + func TestDeprecatingModels(t *testing.T) { gw, _ := newTestGateway(t, &gatewayStubClient{fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}}) ctx := context.Background() @@ -507,3 +792,24 @@ func TestAiContentLogSwitch(t *testing.T) { t.Fatalf("关闭失败: %+v, %v", fresh.ContentLogUntil, err) } } + +func TestLegacyNullUnusableRowsStayVisible(t *testing.T) { + // 升级路径回归:AutoMigrate 加列后、首次重同步前,存量行 unusable 为 NULL, + // 网关列表与路由必须照常包含这些行,不能因 unusable = false 过滤而整池消失 + gw, svc := newTestGateway(t, &gatewayStubClient{fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}}) + cfg := importAliveConfig(t, svc) + ch := seedChannel(t, gw, cfg.ID, "eu-frankfurt-1", 1, 1) + ctx := context.Background() + if err := gw.db.Exec("UPDATE ai_model_caches SET unusable = NULL WHERE channel_id = ?", ch.ID).Error; err != nil { + t.Fatalf("set legacy null: %v", err) + } + + list, err := gw.GatewayModels(ctx, "") + if err != nil || len(list.Data) != 1 { + t.Errorf("存量 NULL 行应仍在网关列表: %+v, %v", list.Data, err) + } + _, ids, err := gw.modelChannels(ctx, "meta.llama-3.3-70b-instruct", "CHAT") + if err != nil || len(ids) != 1 { + t.Errorf("存量 NULL 行应仍参与路由: %v, %v", ids, err) + } +} diff --git a/internal/service/ociconfig_test.go b/internal/service/ociconfig_test.go index 6e39017..8daeb03 100644 --- a/internal/service/ociconfig_test.go +++ b/internal/service/ociconfig_test.go @@ -137,6 +137,13 @@ func newTestService(t *testing.T, client oci.Client) *OciConfigService { if err != nil { t.Fatalf("open in-memory sqlite: %v", err) } + // :memory: 库每条连接各自独立,后台 goroutine(模型验证等)触发第二条连接会看到空库; + // 收敛到单连接,与 tenantdelete 测试环境一致 + sqlDB, err := db.DB() + if err != nil { + t.Fatalf("database handle: %v", err) + } + sqlDB.SetMaxOpenConns(1) if err := db.AutoMigrate( &model.OciConfig{}, &model.RegionCache{}, &model.CompartmentCache{}, &model.Proxy{}, ); err != nil { diff --git a/internal/service/task.go b/internal/service/task.go index 0c99b31..69e542a 100644 --- a/internal/service/task.go +++ b/internal/service/task.go @@ -3,6 +3,7 @@ package service import ( "context" "encoding/json" + "errors" "fmt" "log" "strings" @@ -46,6 +47,12 @@ type TaskService struct { entries map[uint]cron.EntryID // runMu 让租户删除等待在途执行结束,并阻止新执行读取删除前的任务快照。 runMu sync.RWMutex + // runningMu 保护 running:同一任务不允许并发重复执行 + // (cron 重叠触发静默跳过,手动触发返回 ErrTaskRunning) + runningMu sync.Mutex + running map[uint]bool + // runWG 追踪手动触发的后台执行,Stop 时等待收尾 + runWG sync.WaitGroup // snatchVarsMu 保护 snatchVars:runSnatch 抢满时写入成功通知变量, // execute 组装执行后快照时取走(一次性) @@ -63,6 +70,7 @@ func NewTaskService(db *gorm.DB, configs *OciConfigService, notifier *Notifier, settings: settings, cron: cron.New(), entries: map[uint]cron.EntryID{}, + running: map[uint]bool{}, snatchIPWait: defaultSnatchIPWait, snatchVars: map[uint]map[string]string{}, } @@ -86,10 +94,11 @@ func (s *TaskService) Start() error { return nil } -// Stop 停止调度并等待执行中的任务收尾， +// Stop 停止调度并等待执行中的任务收尾(含手动触发的后台执行), // 保证任务产生的异步通知都已进入 Notifier 的等待队列。 func (s *TaskService) Stop() { <-s.cron.Stop().Done() + s.runWG.Wait() } // healthCheckPayload 是测活任务参数；ociConfigIds 为空表示全部配置。 @@ -356,7 +365,7 @@ func (s *TaskService) TaskLogs(ctx context.Context, id uint, limit int) ([]model return logs, nil } -// RunTaskNow 立即执行一次任务并返回本次日志。 +// RunTaskNow 同步执行一次任务并返回本次日志(内部与测试使用;API 走 TriggerTask)。 func (s *TaskService) RunTaskNow(ctx context.Context, id uint) (*model.TaskLog, error) { if _, err := s.GetTask(ctx, id); err != nil { return nil, err @@ -364,6 +373,43 @@ func (s *TaskService) RunTaskNow(ctx context.Context, id uint) (*model.TaskLog, return s.execute(id), nil } +// ErrTaskRunning 表示任务已有一次执行在途,拒绝重复触发。 +var ErrTaskRunning = errors.New("任务正在执行中,请稍候") + +// TriggerTask 异步触发一次任务执行并立即返回;执行结果照常落任务日志与通知, +// 由前端轮询呈现。同一任务在途时返回 ErrTaskRunning。 +func (s *TaskService) TriggerTask(ctx context.Context, id uint) error { + if _, err := s.GetTask(ctx, id); err != nil { + return err + } + if !s.beginRun(id) { + return ErrTaskRunning + } + s.runWG.Add(1) + go func() { + defer s.runWG.Done() + s.runHeld(id) + }() + return nil +} + +// beginRun 抢占任务的执行权;在途时返回 false。 +func (s *TaskService) beginRun(id uint) bool { + s.runningMu.Lock() + defer s.runningMu.Unlock() + if s.running[id] { + return false + } + s.running[id] = true + return true +} + +func (s *TaskService) endRun(id uint) { + s.runningMu.Lock() + delete(s.running, id) + s.runningMu.Unlock() +} + // schedule 把任务注册进 cron 调度。 func (s *TaskService) schedule(task *model.Task) error { s.mu.Lock() @@ -412,9 +458,18 @@ func (s *TaskService) lockTenantCleanup() func() { return s.runMu.Unlock } -// execute 执行一次任务：加载 → 分派 → 更新任务状态并写日志， -// 前后状态交给通知判定，只在状态变化时推送。 +// execute 执行一次任务:同一任务已有执行在途时静默跳过(cron 重叠触发防抖)。 func (s *TaskService) execute(taskID uint) *model.TaskLog { + if !s.beginRun(taskID) { + return nil + } + return s.runHeld(taskID) +} + +// runHeld 在已持有执行权的前提下完成一次执行:加载 → 分派 → 更新任务状态并写日志, +// 前后状态交给通知判定,只在状态变化时推送。 +func (s *TaskService) runHeld(taskID uint) *model.TaskLog { + defer s.endRun(taskID) s.runMu.RLock() defer s.runMu.RUnlock() return s.executeLocked(taskID) diff --git a/internal/service/task_test.go b/internal/service/task_test.go index 8a63cd4..4fa04d9 100644 --- a/internal/service/task_test.go +++ b/internal/service/task_test.go @@ -3,6 +3,7 @@ package service import ( "context" "encoding/json" + "errors" "fmt" "strings" "sync" @@ -801,3 +802,36 @@ func TestNotifyEventsSnatchVarsMerged(t *testing.T) { } } } + +func TestTriggerTaskAsyncAndDedup(t *testing.T) { + client := &blockingClient{fakeClient: &fakeClient{}, started: make(chan struct{}, 1), release: make(chan struct{})} + configs, tasks, db := newTenantDeleteEnv(t, client) + target, _ := seedDeleteTenants(t, db) + setTenantPrivateKey(t, configs, target.ID) + task := createHealthTask(t, tasks, target.ID) + ctx := context.Background() + + // 触发立即返回,执行在后台开始 + if err := tasks.TriggerTask(ctx, task.ID); err != nil { + t.Fatalf("TriggerTask: %v", err) + } + <-client.started + // 在途重复触发被拒;cron 重叠执行静默跳过 + if err := tasks.TriggerTask(ctx, task.ID); !errors.Is(err, ErrTaskRunning) { + t.Errorf("在途重复触发应返回 ErrTaskRunning: %v", err) + } + if entry := tasks.execute(task.ID); entry != nil { + t.Error("在途时 cron 重叠执行应静默跳过") + } + close(client.release) + tasks.Stop() // 等待后台执行收尾 + assertCount(t, db, &model.TaskLog{}, "task_id = ?", []any{task.ID}, 1) + // 执行权已释放,可再次触发;不存在的任务报错 + if !tasks.beginRun(task.ID) { + t.Error("执行结束后应可重新获得执行权") + } + tasks.endRun(task.ID) + if err := tasks.TriggerTask(ctx, 9999); err == nil { + t.Error("不存在的任务应报错") + } +} diff --git a/internal/service/tenantdelete.go b/internal/service/tenantdelete.go index 4b5053d..75f112a 100644 --- a/internal/service/tenantdelete.go +++ b/internal/service/tenantdelete.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "log" "strings" "gorm.io/gorm" @@ -104,7 +105,8 @@ func planTenantTask(task model.Task, id uint) (tenantTaskAction, bool, error) { func planSnatchTask(task model.Task, id uint) (tenantTaskAction, bool, error) { var payload snatchPayload if err := json.Unmarshal([]byte(task.Payload), &payload); err != nil { - return tenantTaskAction{}, false, fmt.Errorf("parse task %d payload: %w", task.ID, err) + logSkippedTask(task, err) + return tenantTaskAction{}, false, nil } if payload.OciConfigID != id { return tenantTaskAction{}, false, nil @@ -112,10 +114,17 @@ func planSnatchTask(task model.Task, id uint) (tenantTaskAction, bool, error) { return tenantTaskAction{task: task, deleteTask: true}, true, nil } +// logSkippedTask 记录 payload 无法解析而被跳过的任务:坏数据无法归属租户, +// fail-closed 会永久阻断删除且只能手工修库;改为保留原任务(不删不改)并放行删除。 +func logSkippedTask(task model.Task, err error) { + log.Printf("tenant delete: 任务 %d(%s)payload 无法解析,跳过处理: %v", task.ID, task.Type, err) +} + func planMultiTenantTask(task model.Task, id uint) (tenantTaskAction, bool, error) { payload, err := decodeConfigIDs(task) if err != nil { - return tenantTaskAction{}, false, err + logSkippedTask(task, err) + return tenantTaskAction{}, false, nil } if len(payload.OciConfigIDs) == 0 { return tenantTaskAction{task: task, payload: task.Payload}, true, nil @@ -197,11 +206,11 @@ func resetTenantTask(tx *gorm.DB, action tenantTaskAction) error { } func deleteTenantEvents(tx *gorm.DB, id uint, result *tenantDeleteResult) error { - ruleIDs, eventIDs, affectedRules, err := loadTenantEventRefs(tx, id) + ruleIDs, affectedRules, err := loadTenantEventRefs(tx, id) if err != nil { return err } - if err := deleteAlertHits(tx, ruleIDs, eventIDs); err != nil { + if err := deleteAlertHits(tx, id); err != nil { return err } result.alertRuleIDs = mergeIDs(ruleIDs, affectedRules) @@ -214,20 +223,19 @@ func deleteTenantEvents(tx *gorm.DB, id uint, result *tenantDeleteResult) error return nil } -func loadTenantEventRefs(tx *gorm.DB, id uint) ([]uint, []uint, []uint, error) { +func loadTenantEventRefs(tx *gorm.DB, id uint) ([]uint, []uint, error) { ruleIDs, err := lockedTenantRuleIDs(tx, id) if err != nil { - return nil, nil, nil, fmt.Errorf("load tenant alert rules: %w", err) + return nil, nil, fmt.Errorf("load tenant alert rules: %w", err) } - eventIDs, err := lockedTenantEventIDs(tx, id) + if err := lockTenantEventRows(tx, id); err != nil { + return nil, nil, fmt.Errorf("load tenant log events: %w", err) + } + affected, err := alertHitRuleIDs(tx, id) if err != nil { - return nil, nil, nil, fmt.Errorf("load tenant log events: %w", err) + return nil, nil, err } - affected, err := alertHitRuleIDs(tx, eventIDs) - if err != nil { - return nil, nil, nil, err - } - return ruleIDs, eventIDs, affected, nil + return ruleIDs, affected, nil } func lockedTenantRuleIDs(tx *gorm.DB, id uint) ([]uint, error) { @@ -241,23 +249,24 @@ func lockedTenantRuleIDs(tx *gorm.DB, id uint) ([]uint, error) { return ids, err } -func lockedTenantEventIDs(tx *gorm.DB, id uint) ([]uint, error) { +// lockTenantEventRows 对租户全部日志事件行加 FOR UPDATE 锁(SQLite 忽略, +// MySQL/PG 阻塞并发命中写入);事件可达数万,ID 不再回传拼接 SQL, +// 后续删除与关联查询一律用子查询,避免绑定变量上限。 +func lockTenantEventRows(tx *gorm.DB, id uint) error { var rows []model.LogEvent - err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id"). + return tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id"). Where("oci_config_id = ?", id).Order("id").Find(&rows).Error - ids := make([]uint, 0, len(rows)) - for _, row := range rows { - ids = append(ids, row.ID) - } - return ids, err } -func alertHitRuleIDs(tx *gorm.DB, eventIDs []uint) ([]uint, error) { - if len(eventIDs) == 0 { - return nil, nil - } +// tenantEventIDs 构造「本租户日志事件 ID」子查询,供 IN (?) 内联。 +func tenantEventIDs(tx *gorm.DB, id uint) *gorm.DB { + return tx.Model(&model.LogEvent{}).Select("id").Where("oci_config_id = ?", id) +} + +// alertHitRuleIDs 找出命中引用了本租户事件的规则 ID(去重,可能含他租户/全局规则)。 +func alertHitRuleIDs(tx *gorm.DB, id uint) ([]uint, error) { ids := make([]uint, 0) - err := tx.Model(&model.AlertRuleHit{}).Where("log_event_id IN ?", eventIDs). + err := tx.Model(&model.AlertRuleHit{}).Where("log_event_id IN (?)", tenantEventIDs(tx, id)). Distinct().Pluck("rule_id", &ids).Error if err != nil { return nil, fmt.Errorf("load affected alert rules: %w", err) @@ -280,19 +289,13 @@ func mergeIDs(groups ...[]uint) []uint { return out } -func deleteAlertHits(tx *gorm.DB, ruleIDs, eventIDs []uint) error { - query := tx.Model(&model.AlertRuleHit{}) - switch { - case len(ruleIDs) > 0 && len(eventIDs) > 0: - query = query.Where("rule_id IN ? OR log_event_id IN ?", ruleIDs, eventIDs) - case len(ruleIDs) > 0: - query = query.Where("rule_id IN ?", ruleIDs) - case len(eventIDs) > 0: - query = query.Where("log_event_id IN ?", eventIDs) - default: - return nil - } - if err := query.Delete(&model.AlertRuleHit{}).Error; err != nil { +// deleteAlertHits 删除本租户相关的全部命中:规则属于本租户,或命中引用了 +// 本租户的日志事件(他租户/全局规则命中本租户事件的行一并清)。 +func deleteAlertHits(tx *gorm.DB, id uint) error { + rules := tx.Model(&model.AlertRule{}).Select("id").Where("oci_config_id = ?", id) + err := tx.Where("rule_id IN (?) OR log_event_id IN (?)", rules, tenantEventIDs(tx, id)). + Delete(&model.AlertRuleHit{}).Error + if err != nil { return fmt.Errorf("delete tenant alert hits: %w", err) } return nil diff --git a/internal/service/tenantdelete_test.go b/internal/service/tenantdelete_test.go index 25eb58b..70faa58 100644 --- a/internal/service/tenantdelete_test.go +++ b/internal/service/tenantdelete_test.go @@ -108,7 +108,10 @@ func tenantTaskCases() []tenantTaskCase { {name: "测活多租户", task: taskOf(model.TaskTypeHealthCheck, `{"ociConfigIds":[1,2]}`), wantOK: true, wantPayload: `{"ociConfigIds":[2]}`}, {name: "成本去重命中", task: taskOf(model.TaskTypeCost, `{"ociConfigIds":[1,1,2]}`), wantOK: true, wantPayload: `{"ociConfigIds":[2]}`}, {name: "成本未命中", task: taskOf(model.TaskTypeCost, `{"ociConfigIds":[2]}`)}, - {name: "非法 JSON", task: taskOf(model.TaskTypeCost, `{`), wantErr: true}, + // 坏 payload 记警告跳过(保留原任务),不再 fail-closed 阻断整个租户删除 + {name: "非法 JSON 跳过", task: taskOf(model.TaskTypeCost, `{`)}, + {name: "抢机坏 payload 跳过", task: taskOf(model.TaskTypeSnatch, `not-json`)}, + {name: "抢机空 payload 跳过", task: taskOf(model.TaskTypeSnatch, ``)}, } } @@ -607,3 +610,67 @@ func assertCount(t *testing.T, db *gorm.DB, value any, query string, args []any, t.Errorf("count %T = %d, want %d", value, got, want) } } + +func TestDeleteTenantSkipsCorruptTaskPayload(t *testing.T) { + configs, _, db := newTenantDeleteEnv(t, &fakeClient{}) + target, _ := seedDeleteTenants(t, db) + corrupt := model.Task{Name: "corrupt-snatch", Type: model.TaskTypeSnatch, Payload: `{broken`} + mustCreate(t, db, &corrupt) + mustCreate(t, db, &model.TaskLog{TaskID: corrupt.ID, Message: "keep"}) + + if err := configs.Delete(context.Background(), target.ID); err != nil { + t.Fatalf("坏 payload 不应阻断租户删除: %v", err) + } + var kept model.Task + if err := db.First(&kept, corrupt.ID).Error; err != nil || kept.Payload != `{broken` { + t.Errorf("坏任务应原样保留: %+v, %v", kept, err) + } + assertCount(t, db, &model.TaskLog{}, "task_id = ?", []any{corrupt.ID}, 1) + assertCount(t, db, &model.OciConfig{}, "id = ?", []any{target.ID}, 0) +} + +func TestDeleteTenantEventsAffectedRules(t *testing.T) { + _, _, db := newTenantDeleteEnv(t, &fakeClient{}) + target, other := seedDeleteTenants(t, db) + seedTenantEvents(t, db, target.ID, other.ID) + + // 规则 ID 须在删除前取:target-rule 会随租户一并删除 + var targetRule, globalRule model.AlertRule + db.Where("name = ?", "target-rule").First(&targetRule) + db.Where("name = ?", "global-rule").First(&globalRule) + + result := &tenantDeleteResult{} + err := db.Transaction(func(tx *gorm.DB) error { + return deleteTenantEvents(tx, target.ID, result) + }) + if err != nil { + t.Fatalf("deleteTenantEvents: %v", err) + } + // target-rule 属本租户,global-rule 的命中引用了本租户事件:都应回收冷却 + got := map[uint]bool{} + for _, id := range result.alertRuleIDs { + got[id] = true + } + if len(got) != 2 || !got[targetRule.ID] || !got[globalRule.ID] { + t.Errorf("alertRuleIDs = %v, want 含 target-rule(%d) 与 global-rule(%d)", result.alertRuleIDs, targetRule.ID, globalRule.ID) + } + // 仅保留 global-rule × other-event 一条命中 + assertCount(t, db, &model.AlertRuleHit{}, "", nil, 1) +} + +func TestDeleteTenantManyEventsNoVarLimit(t *testing.T) { + // 回归:事件数超 SQLite 绑定变量上限(32766)时删除仍成功(旧实现 IN 展开必失败) + configs, _, db := newTenantDeleteEnv(t, &fakeClient{}) + target, _ := seedDeleteTenants(t, db) + events := make([]model.LogEvent, 0, 33000) + for i := 0; i < 33000; i++ { + events = append(events, model.LogEvent{OciConfigID: target.ID, MessageID: fmt.Sprintf("bulk-%d", i)}) + } + if err := db.CreateInBatches(&events, 500).Error; err != nil { + t.Fatalf("seed events: %v", err) + } + if err := configs.Delete(context.Background(), target.ID); err != nil { + t.Fatalf("数万事件时删除不应受绑定变量上限影响: %v", err) + } + assertCount(t, db, &model.LogEvent{}, "oci_config_id = ?", []any{target.ID}, 0) +}