tt/ttstd-infinite-canvas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(backend): PR 合并治理 - 限制轮询并收敛模型目录事务

Browse Source
This commit is contained in:
ddCat
2026-09-07 15:41:37 +08:00
parent f749bab2e5
commit 6327c08f16
6 changed files with 91 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
backend/internal/repository/finance.go
View File

@@ -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

27
backend/internal/service/channel_models.go
View File

@@ -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 {

14
backend/internal/service/runtime_policy_test.go
View File

@@ -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")
}

18
backend/internal/service/task_worker.go
View File

@@ -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 "视频生成等待超时，请稍后到任务中心查看或重试。"

4
docs/content/docs/backend/newapi-channel-2.mdx
View File

@@ -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 秒时，只要查询接口返回有效的处理中状态，本地会持续以低频查询等待完成，不会把正常排队误报为“模型服务拒绝了请求”。视频真正返回失败状态时，仍按供应商失败处理。
## 兼容边界

8
docs/content/docs/progress/pending-test.mdx
View File

@@ -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、画幅和生成方式差异定价收进可选的“按规格定价”。