From 839db1c3191f0f0555093a96575c10b0c0dc37ed Mon Sep 17 00:00:00 2001 From: ddcat666 <26564566+ddcat666@users.noreply.github.com> Date: Thu, 1 Oct 2026 13:18:06 +0800 Subject: [PATCH] =?UTF-8?q?fix(agent):=20=E5=9B=BE=E7=89=87=E5=AE=A1?= =?UTF-8?q?=E6=89=B9=20-=20=E4=BF=AE=E6=94=B9=E7=94=9F=E6=88=90=E5=8F=82?= =?UTF-8?q?=E6=95=B0=E6=97=B6=E4=B8=8D=E5=86=8D=E5=9B=A0=E5=AE=B9=E9=87=8F?= =?UTF-8?q?=E6=A3=80=E6=9F=A5=E8=87=AA=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 审批落库时已经持有存储锁,干跑创建任务又申请同一把锁。互斥锁不能重入,修改图片参数会一直停住,后端测试也会在 20 分钟后超时。 --- .../app/cloud_agent_media_approval.go | 2 ++ backend/internal/app/service.go | 23 +++++++++++-------- backend/internal/app/task_creation.go | 11 +++++++-- .../internal/app/task_storage_admission.go | 11 +++++++-- 4 files changed, 33 insertions(+), 14 deletions(-) diff --git a/backend/internal/app/cloud_agent_media_approval.go b/backend/internal/app/cloud_agent_media_approval.go index acc6fca05..b61ff18b4 100644 --- a/backend/internal/app/cloud_agent_media_approval.go +++ b/backend/internal/app/cloud_agent_media_approval.go @@ -49,8 +49,10 @@ func (s *Service) updateCloudAgentMediaApproval(repo *repository.Repository, run } // Dry admission uses the same model, ownership and option checks as task // submission, without reserving credits or creating a generation task. + // DecideCloudAgentApproval already holds storageMu for this mutation. preparation := &creationTaskPreparation{} req.creationPrepare = preparation + req.callerHoldsStorageMu = true task, err := s.CreateTask(run.UserID, req) if err != nil { return err diff --git a/backend/internal/app/service.go b/backend/internal/app/service.go index 20746a456..3ea812729 100644 --- a/backend/internal/app/service.go +++ b/backend/internal/app/service.go @@ -101,16 +101,19 @@ const taskLogPayloadLimit = 4000 type CreateTaskRequest struct { creationPrepare *creationTaskPreparation admission *taskAdmission - ProjectID string `json:"projectId"` - Type string `json:"type"` - Operation string `json:"operation"` - Prompt string `json:"prompt"` - Provider string `json:"provider"` - Model string `json:"model"` - LogicalModelID string `json:"logicalModelId"` - Input map[string]any `json:"input"` - TraceID string `json:"-"` - RequestID string `json:"-"` + // callerHoldsStorageMu 只给已经持有 Service.storageMu 的内部调用设置。 + // 文件容量检查不能再锁一次,否则审批改参数会自锁。 + callerHoldsStorageMu bool + ProjectID string `json:"projectId"` + Type string `json:"type"` + Operation string `json:"operation"` + Prompt string `json:"prompt"` + Provider string `json:"provider"` + Model string `json:"model"` + LogicalModelID string `json:"logicalModelId"` + Input map[string]any `json:"input"` + TraceID string `json:"-"` + RequestID string `json:"-"` } type TaskListOptions struct { diff --git a/backend/internal/app/task_creation.go b/backend/internal/app/task_creation.go index 6614ae8a4..28c87dc8c 100644 --- a/backend/internal/app/task_creation.go +++ b/backend/internal/app/task_creation.go @@ -119,8 +119,15 @@ func (s *Service) CreateTask(userID string, req CreateTaskRequest) (*model.Task, return nil, BadAuthRequest(fmt.Sprintf("同时排队或运行的任务最多 %d 个,请等待已有任务完成", policy.Task.ActiveTaskLimit)) } // 媒体任务在扣费和调用上游前确认账号文件容量,避免容量已满仍然计费却无法回存产物。 - if err := s.requireStoredFileCapacityForTask(userID, taskType, policy); err != nil { - return nil, err + // 调用方已持锁时走不加锁的版本;这里不能再抢同一把 storageMu。 + var capacityErr error + if req.callerHoldsStorageMu { + capacityErr = s.requireStoredFileCapacityWhileLocked(userID, taskType, policy) + } else { + capacityErr = s.requireStoredFileCapacityForTask(userID, taskType, policy) + } + if capacityErr != nil { + return nil, capacityErr } task := model.Task{ID: newID(), UserID: userID, TraceID: req.TraceID, RequestID: req.RequestID, ProjectID: req.ProjectID, Type: taskType, Status: model.TaskStatusQueued, Stage: "等待队列调度", Progress: 5, Prompt: prompt, Operation: req.Operation, Provider: req.Provider, Model: req.Model} if req.admission != nil { diff --git a/backend/internal/app/task_storage_admission.go b/backend/internal/app/task_storage_admission.go index 0051a7b29..a8df5e49a 100644 --- a/backend/internal/app/task_storage_admission.go +++ b/backend/internal/app/task_storage_admission.go @@ -20,6 +20,15 @@ func taskTypeProducesStoredFile(taskType string) bool { // 账号文件容量此前只在产物回存时校验,上游调用已经发出并扣费,失败只能表现为 // 媒体无法入库。生成任务必须在扣费前确认账号还有可用容量,容量已满时直接拒绝。 func (s *Service) requireStoredFileCapacityForTask(userID string, taskType string, policy RuntimePolicySetting) error { + s.storageMu.Lock() + defer s.storageMu.Unlock() + return s.requireStoredFileCapacityWhileLocked(userID, taskType, policy) +} + +// requireStoredFileCapacityWhileLocked 与上面相同,但调用方必须已经持有 storageMu。 +// sync.Mutex 不可重入;审批改参数的干跑 admission 发生在 DecideCloudAgentApproval +// 的临界区内,再次加锁会把测试和线上审批一起卡死。 +func (s *Service) requireStoredFileCapacityWhileLocked(userID string, taskType string, policy RuntimePolicySetting) error { if !taskTypeProducesStoredFile(taskType) { return nil } @@ -27,8 +36,6 @@ func (s *Service) requireStoredFileCapacityForTask(userID string, taskType strin if storedLimit <= 0 { return nil } - s.storageMu.Lock() - defer s.storageMu.Unlock() storedBytes, err := s.repo.UserStoredFileBytes(userID) if err != nil { return err