From 6327c08f16ac33a0a1326287df6c3ddf967be145 Mon Sep 17 00:00:00 2001 From: ddCat <2695092045@qq.com> Date: Mon, 7 Sep 2026 15:41:37 +0800 Subject: [PATCH] =?UTF-8?q?fix(backend):=20PR=20=E5=90=88=E5=B9=B6?= =?UTF-8?q?=E6=B2=BB=E7=90=86=20-=20=E9=99=90=E5=88=B6=E8=BD=AE=E8=AF=A2?= =?UTF-8?q?=E5=B9=B6=E6=94=B6=E6=95=9B=E6=A8=A1=E5=9E=8B=E7=9B=AE=E5=BD=95?= =?UTF-8?q?=E4=BA=8B=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/repository/finance.go | 59 +++++++++++++++---- backend/internal/service/channel_models.go | 27 +-------- .../internal/service/runtime_policy_test.go | 14 ++++- backend/internal/service/task_worker.go | 18 +++++- .../content/docs/backend/newapi-channel-2.mdx | 4 +- docs/content/docs/progress/pending-test.mdx | 8 +++ 6 files changed, 91 insertions(+), 39 deletions(-) diff --git a/backend/internal/repository/finance.go b/backend/internal/repository/finance.go index eb2de0e6c..0d80734ad 100644 --- a/backend/internal/repository/finance.go +++ b/backend/internal/repository/finance.go @@ -3,6 +3,7 @@ package repository import ( "crypto/rand" "encoding/hex" + "encoding/json" "errors" "strconv" "strings" @@ -230,8 +231,8 @@ func (r *Repository) PopulateChannelModelPriceTier(item *model.ChannelModel) err return r.attachChannelModelPriceTiers([]*model.ChannelModel{item}) } -func (r *Repository) DeleteChannelModel(channelID string, id string, modelsJSON string, now time.Time) error { - deleted, err := r.DeleteChannelModels(channelID, []string{id}, modelsJSON, now) +func (r *Repository) DeleteChannelModel(channelID string, id string, now time.Time) error { + deleted, err := r.DeleteChannelModels(channelID, []string{id}, now) if err != nil { return err } @@ -244,13 +245,16 @@ func (r *Repository) DeleteChannelModel(channelID string, id string, modelsJSON // DeleteChannelModels atomically removes a selection and refreshes the channel's // compatibility model list. Any active route or task reference aborts the whole // transaction so a bulk action cannot leave the administrator with a partial result. -func (r *Repository) DeleteChannelModels(channelID string, ids []string, modelsJSON string, now time.Time) (int64, error) { +func (r *Repository) DeleteChannelModels(channelID string, ids []string, now time.Time) (int64, error) { ids = uniqueStrings(ids) if len(ids) == 0 { return 0, gorm.ErrRecordNotFound } var deleted int64 err := r.db.Transaction(func(tx *gorm.DB) error { + if err := r.lockSystemChannelForModelMutation(tx, channelID); err != nil { + return err + } var existing int64 if err := tx.Model(&model.ChannelModel{}). Where("channel_id = ? AND id IN ?", channelID, ids). @@ -290,14 +294,8 @@ func (r *Repository) DeleteChannelModels(channelID string, ids []string, modelsJ if err := tx.Where("id IN ? AND channel_id = ?", ids, channelID).Delete(&model.ChannelModel{}).Error; err != nil { return err } - channelResult := tx.Model(&model.ModelChannel{}). - Where("id = ? AND scope = ?", channelID, model.ChannelScopeSystem). - Updates(map[string]any{"models_json": modelsJSON, "updated_at": now}) - if channelResult.Error != nil { - return channelResult.Error - } - if channelResult.RowsAffected != 1 { - return gorm.ErrRecordNotFound + if err := refreshChannelModelNames(tx, channelID, now); err != nil { + return err } deleted = result.RowsAffected return nil @@ -305,6 +303,45 @@ func (r *Repository) DeleteChannelModels(channelID string, ids []string, modelsJ return deleted, err } +// SyncChannelModelNames serializes compatibility-list refreshes for a system +// channel and derives the list from the committed channel_models rows. Callers +// must not pass a previously-read snapshot because concurrent administrator +// writes could otherwise overwrite a newer catalog. +func (r *Repository) SyncChannelModelNames(channelID string, now time.Time) error { + return r.db.Transaction(func(tx *gorm.DB) error { + if err := r.lockSystemChannelForModelMutation(tx, channelID); err != nil { + return err + } + return refreshChannelModelNames(tx, channelID, now) + }) +} + +func (r *Repository) lockSystemChannelForModelMutation(tx *gorm.DB, channelID string) error { + query := tx.Select("id").Where("id = ? AND scope = ?", channelID, model.ChannelScopeSystem) + if r.Dialect() == "postgres" { + query = query.Clauses(clause.Locking{Strength: "UPDATE"}) + } + var channel model.ModelChannel + return query.First(&channel).Error +} + +func refreshChannelModelNames(tx *gorm.DB, channelID string, now time.Time) error { + var names []string + if err := tx.Model(&model.ChannelModel{}). + Where("channel_id = ? AND enabled = ?", channelID, true). + Order("created_at asc"). + Pluck("model_key", &names).Error; err != nil { + return err + } + encoded, err := json.Marshal(names) + if err != nil { + return err + } + return tx.Model(&model.ModelChannel{}). + Where("id = ? AND scope = ?", channelID, model.ChannelScopeSystem). + Updates(map[string]any{"models_json": string(encoded), "updated_at": now}).Error +} + func (r *Repository) CreateMissingChannelModels(items []model.ChannelModel) (int64, error) { if len(items) == 0 { return 0, nil diff --git a/backend/internal/service/channel_models.go b/backend/internal/service/channel_models.go index 811965d62..02cdc4a1b 100644 --- a/backend/internal/service/channel_models.go +++ b/backend/internal/service/channel_models.go @@ -866,18 +866,8 @@ func (s *Service) DeleteAdminChannelModels(actor *model.User, channelID string, if found != len(modelIDs) { return 0, BadAuthRequest("所选渠道模型中存在已删除或不属于当前渠道的记录，请刷新后重试") } - names := make([]string, 0, len(items)) - for _, item := range items { - if item.Enabled && !selected[item.ID] { - names = append(names, item.ModelKey) - } - } - encoded, err := json.Marshal(names) - if err != nil { - return 0, err - } // 删除模型与渠道的兼容模型清单必须同事务提交，避免接口报错但列表已部分变化。 - deleted, err := s.repo.DeleteChannelModels(channelID, modelIDs, string(encoded), time.Now()) + deleted, err := s.repo.DeleteChannelModels(channelID, modelIDs, time.Now()) if errors.Is(err, repository.ErrChannelModelInUse) { return 0, BadAuthRequest("所选渠道模型中有模型仍被前台模型供应线路或进行中任务使用，本次未删除任何模型") } @@ -986,20 +976,7 @@ func (s *Service) ensureChannelModels(channelID string, includeDisabled bool) ([ } func (s *Service) syncChannelModelNames(channel *model.ModelChannel) error { - items, err := s.repo.ChannelModels(channel.ID, false) - if err != nil { - return err - } - names := make([]string, 0, len(items)) - for _, item := range items { - names = append(names, item.ModelKey) - } - encoded, err := json.Marshal(names) - if err != nil { - return err - } - channel.ModelsJSON = string(encoded) - return s.repo.Save(channel) + return s.repo.SyncChannelModelNames(channel.ID, time.Now()) } func (s *Service) capabilityForProtocol(protocol model.ChannelInterfaceType) string { diff --git a/backend/internal/service/runtime_policy_test.go b/backend/internal/service/runtime_policy_test.go index 96bdab14c..d918039e8 100644 --- a/backend/internal/service/runtime_policy_test.go +++ b/backend/internal/service/runtime_policy_test.go @@ -31,7 +31,8 @@ func TestOnlyResumableNewAPIChannel2VideoDeadlinesStayRunning(t *testing.T) { if err != nil { t.Fatal(err) } - base := model.Task{ID: "task-1", Type: "canvas_video", ProviderRequestID: "provider-task-1"} + startedAt := time.Now() + base := model.Task{ID: "task-1", Type: "canvas_video", ProviderRequestID: "provider-task-1", StartedAt: &startedAt} if !svc.shouldDeferVideoProviderTask(base, string(input), context.DeadlineExceeded) { t.Fatal("resumable NewAPI Channel 2 deadline should remain running") } @@ -39,6 +40,17 @@ func TestOnlyResumableNewAPIChannel2VideoDeadlinesStayRunning(t *testing.T) { if !svc.shouldDeferVideoProviderTask(base, string(input), pendingErr) { t.Fatal("resumable NewAPI Channel 2 provider sync delay should remain running") } + expiredStartedAt := startedAt.Add(-newAPIChannel2TaskSyncMaxAge - time.Second) + expiredTask := model.Task{ID: base.ID, Type: base.Type, ProviderRequestID: base.ProviderRequestID, StartedAt: &expiredStartedAt} + if svc.shouldDeferVideoProviderTask(expiredTask, string(input), pendingErr) { + t.Fatal("provider sync delay must stop after the bounded synchronization window") + } + if !svc.shouldDeferVideoProviderTask(expiredTask, string(input), context.DeadlineExceeded) { + t.Fatal("a normal long-running provider task must remain resumable after the synchronization window") + } + if svc.shouldDeferVideoProviderTask(model.Task{ID: base.ID, Type: base.Type, ProviderRequestID: base.ProviderRequestID}, string(input), pendingErr) { + t.Fatal("provider sync delay without a persisted start time must fail closed") + } if svc.shouldDeferVideoProviderTask(model.Task{ID: "task-2", Type: "canvas_video"}, string(input), context.DeadlineExceeded) { t.Fatal("missing provider task id must not be deferred") } diff --git a/backend/internal/service/task_worker.go b/backend/internal/service/task_worker.go index 0e60e9e42..13d8f0b12 100644 --- a/backend/internal/service/task_worker.go +++ b/backend/internal/service/task_worker.go @@ -12,6 +12,8 @@ import ( "infinite-canvas/backend/internal/model" ) +const newAPIChannel2TaskSyncMaxAge = 5 * time.Minute + // taskWorkerCoordinator 收敛任务领取、租约维护和执行结果落库，避免 Service 同时承担 worker 生命周期与业务命令。 type taskWorkerCoordinator struct { service *Service @@ -226,6 +228,9 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot _ = s.log(task.UserID, task.ID, "info", message, task.PollStage) return nil } + if newAPIChannel2TaskSyncExpired(*task, err, time.Now()) { + err = errors.New("上游任务长时间未同步，已停止自动查询，请确认渠道任务状态后重试。") + } if errors.Is(err, context.DeadlineExceeded) { err = errors.New(taskTimeoutMessage(task.Type)) } @@ -297,7 +302,7 @@ func (s *Service) shouldDeferVideoProviderTask(task model.Task, decryptedInput s deferSignal := errors.Is(err, context.DeadlineExceeded) var pendingErr providerStatePendingError if errors.As(err, &pendingErr) { - deferSignal = strings.TrimSpace(pendingErr.TaskID) == providerRequestID + deferSignal = strings.TrimSpace(pendingErr.TaskID) == providerRequestID && !newAPIChannel2TaskSyncExpired(task, err, time.Now()) } if !deferSignal { return false @@ -310,6 +315,17 @@ func (s *Service) shouldDeferVideoProviderTask(task model.Task, decryptedInput s return resolveErr == nil && resolved.InterfaceType == string(model.ChannelInterfaceNewAPIChannel2) } +func newAPIChannel2TaskSyncExpired(task model.Task, err error, now time.Time) bool { + var pendingErr providerStatePendingError + if !errors.As(err, &pendingErr) || strings.TrimSpace(pendingErr.TaskID) == "" || strings.TrimSpace(pendingErr.TaskID) != strings.TrimSpace(task.ProviderRequestID) { + return false + } + if task.StartedAt == nil { + return true + } + return !now.Before(task.StartedAt.Add(newAPIChannel2TaskSyncMaxAge)) +} + func taskTimeoutMessage(taskType string) string { if strings.HasPrefix(taskType, "canvas_video") || strings.HasPrefix(taskType, "video_") { return "视频生成等待超时，请稍后到任务中心查看或重试。" diff --git a/docs/content/docs/backend/newapi-channel-2.mdx b/docs/content/docs/backend/newapi-channel-2.mdx index 9b32b4d6b..ff3eb2f2c 100644 --- a/docs/content/docs/backend/newapi-channel-2.mdx +++ b/docs/content/docs/backend/newapi-channel-2.mdx @@ -42,7 +42,9 @@ description: Channel 2 视频任务创建、轮询与上游任务同步延迟处 - 释放当前租约并把 `next_poll_at` 延后 15 秒； - 下一轮从已保存的 Provider Task ID 恢复，只发送查询请求。 -因此，上游生成耗时 500 至 600 秒时，本地会持续以低频查询等待完成，不会把正常排队误报为“模型服务拒绝了请求”。视频真正返回失败状态时，仍按供应商失败处理。 +`task_not_exist` 的跨轮次同步等待从本地任务首次开始执行起最多持续 5 分钟。超过该上限仍未得到有效任务状态时，Worker 将任务置为失败并释放活跃任务与计费预留，避免无效 Provider Task ID 永久占用用户任务名额。该限制只针对持续的 `task_not_exist`；查询接口已经返回正常 pending/running 状态后的长视频生成仍按视频任务超时和延期策略恢复。 + +因此，上游生成耗时 500 至 600 秒时，只要查询接口返回有效的处理中状态，本地会持续以低频查询等待完成，不会把正常排队误报为“模型服务拒绝了请求”。视频真正返回失败状态时，仍按供应商失败处理。 ## 兼容边界 diff --git a/docs/content/docs/progress/pending-test.mdx b/docs/content/docs/progress/pending-test.mdx index 99c02e4b7..24d796608 100644 --- a/docs/content/docs/progress/pending-test.mdx +++ b/docs/content/docs/progress/pending-test.mdx @@ -153,6 +153,7 @@ description: 当前项目已实现功能的手工验证清单与未确认边界 - [x] 官方插件兼容从 `data.task_id` 和 `data.taskId` 读取上游任务 ID；Channel 2 创建成功后首次查询等待 2 秒，仅将结构化 `400 task_not_exist` 识别为上游同步延迟，并按 2 秒、3 秒、5 秒、5 秒退避复查同一任务。 - [x] 短期重试耗尽后，Worker 保留本地任务 `running`、原 `provider_request_id` 和计费状态，延后 15 秒继续查询；恢复过程不重复发送创建请求、不切换渠道，也不立即失败或退款。其他接口类型、HTTP 状态和错误码仍按原失败语义处理。 +- [x] 合并补修为跨轮次 `task_not_exist` 增加 5 分钟总同步上限；超过上限进入明确失败态并释放任务与计费资源，正常 pending/running 长视频仍可跨执行窗口恢复。 - [x] Channel 2 终态响应兼容 `data.result_url`、`data.video_url`、`data.output_url`、`data.url`、`data.metadata.url` 及对应的 `data.data.*` 嵌套路径；API 调用日志在该接口下优先记录嵌套真实任务 ID。协议与日志富化专项回归测试通过。 - [ ] 使用真实 Channel 2 分别验证上游任务 ID 短暂不可见和生成耗时 500 至 600 秒的任务：本地不再提示“模型服务拒绝了请求”，创建请求只出现一次，后续始终查询同一 Provider Task ID，最终视频正常回写；真实参数错误仍应及时失败。 - [ ] 使用返回 `SUCCESS` 且视频地址位于 `data.result_url` 或 `data.data.video_url` 的真实任务，确认不再提示“声明式协议已完成但没有返回结果”，并在当前 OSS 配置下成功下载并持久化为本地资源。 @@ -523,6 +524,13 @@ description: 当前项目已实现功能的手工验证清单与未确认边界 - [ ] 在画布中分别选择按次图片模型、按秒视频模型和前台逻辑模型，确认修改数量、时长、分辨率和参考素材后价格立即更新且与实际账单一致。 - [ ] 在浅色/深色主题及生成中、禁用状态下确认胶囊价格可读、圆形操作区清晰且不挤压模型与参数控件。 +## 系统渠道模型批量删除 + +- [x] 管理员可勾选最多 100 个渠道模型并一次删除；任一模型仍被活动线路或进行中任务引用时整批拒绝，不产生部分删除。 +- [x] 合并补修将兼容 `models_json` 的重建收进删除事务，并由 repository 在取得渠道写锁后读取当前启用模型；模型保存后的目录同步也复用同一锁定入口，避免并发管理员操作用事务外旧快照覆盖较新的模型目录。 +- [x] 后端批量删除与引用拒绝专项测试、前端管理页回归测试和格式检查通过。 +- [ ] 在 PostgreSQL 测试环境由两个管理员并发执行模型新增、启停和批量删除，确认 `channel_models` 与渠道 `models_json` 最终一致，删除失败时列表不发生部分变化。 + ## 系统渠道默认积分价格 - [x] 新建渠道模型默认使用“所有规格统一价格”，管理员只需填写一次积分；1K/2K/4K、画幅和生成方式差异定价收进可选的“按规格定价”。