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

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

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 {

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

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