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

fix(platform): 任务并发 - 账号活动任务上限统一跟随后台热配置

Browse Source
This commit is contained in:
ddCat committed 2026-07-27 14:23:18 +08:00
1 parent 0b064e532a
commit dd4af80b98
9 files changed
+58 -122
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/internal/service/project.go
+26 -51
View File
@@ -12,24 +12,22 @@ import (
)
type CreateProjectRequest struct {
Name string `json:"name"`
Type string `json:"type"`
AspectRatio string `json:"aspectRatio"`
SourceType string `json:"sourceType"`
Description string `json:"description"`
StylePresetID string `json:"stylePresetId"`
ActiveTaskLimit int `json:"activeTaskLimit"`
Name string `json:"name"`
Type string `json:"type"`
AspectRatio string `json:"aspectRatio"`
SourceType string `json:"sourceType"`
Description string `json:"description"`
StylePresetID string `json:"stylePresetId"`
}
type UpdateProjectRequest struct {
Name string `json:"name"`
Type string `json:"type"`
AspectRatio string `json:"aspectRatio"`
SourceType string `json:"sourceType"`
Description *string `json:"description"`
StylePresetID *string `json:"stylePresetId"`
ActiveTaskLimit *int `json:"activeTaskLimit"`
Status string `json:"status"`
Name string `json:"name"`
Type string `json:"type"`
AspectRatio string `json:"aspectRatio"`
SourceType string `json:"sourceType"`
Description *string `json:"description"`
StylePresetID *string `json:"stylePresetId"`
Status string `json:"status"`
}
type CreateProjectUnitRequest struct {
@@ -147,9 +145,6 @@ func (s *Service) ProjectDetail(userID string, id string) (ProjectDetail, error)
if err != nil {
return ProjectDetail{}, err
}
if project.ActiveTaskLimit <= 0 {
project.ActiveTaskLimit = 3
}
return ProjectDetail{Project: *project, Units: units, Canvases: canvases, CanvasUnitLinks: canvasUnitLinks, Assets: assets, Workflows: workflows, Shots: shots, ShotReferences: shotReferences, AssetCandidates: candidates}, nil
}
@@ -174,14 +169,7 @@ func (s *Service) CreateProject(userID string, req CreateProjectRequest) (model.
sourceType = "blank"
}
now := time.Now()
activeTaskLimit := req.ActiveTaskLimit
if activeTaskLimit == 0 {
activeTaskLimit = 3
}
if activeTaskLimit < 1 || activeTaskLimit > 20 {
return model.Project{}, BadAuthRequest("项目活跃任务上限必须在 1 到 20 之间")
}
project := model.Project{ID: newID(), UserID: userID, Name: name, Type: projectType, AspectRatio: aspectRatio, SourceType: sourceType, Description: strings.TrimSpace(req.Description), StylePresetID: strings.TrimSpace(req.StylePresetID), ActiveTaskLimit: activeTaskLimit, Status: model.ProjectStatusActive, Revision: 1, CreatedAt: now, UpdatedAt: now}
project := model.Project{ID: newID(), UserID: userID, Name: name, Type: projectType, AspectRatio: aspectRatio, SourceType: sourceType, Description: strings.TrimSpace(req.Description), StylePresetID: strings.TrimSpace(req.StylePresetID), Status: model.ProjectStatusActive, Revision: 1, CreatedAt: now, UpdatedAt: now}
if err := s.repo.CreateProject(&project); err != nil {
return model.Project{}, err
}
@@ -217,12 +205,6 @@ func (s *Service) UpdateProject(userID string, id string, req UpdateProjectReque
if req.StylePresetID != nil {
project.StylePresetID = strings.TrimSpace(*req.StylePresetID)
}
if req.ActiveTaskLimit != nil {
if *req.ActiveTaskLimit < 1 || *req.ActiveTaskLimit > 20 {
return model.Project{}, BadAuthRequest("项目活跃任务上限必须在 1 到 20 之间")
}
project.ActiveTaskLimit = *req.ActiveTaskLimit
}
if status := model.ProjectStatus(strings.TrimSpace(req.Status)); status != "" {
if status != model.ProjectStatusActive && status != model.ProjectStatusArchived {
return model.Project{}, BadAuthRequest("不支持的项目状态")
@@ -474,43 +456,36 @@ func IsProjectNotFound(err error) bool {
return errors.Is(err, gorm.ErrRecordNotFound)
}
// 任务仍以画布 ID 作为 projectId；这里统一解析到业务项目，避免项目限额和画布主键语义混淆。
func (s *Service) projectTaskLimit(userID string, canvasOrProjectID string) (string, int, error) {
// 任务仍以画布 ID 作为 projectId；写入前必须解析到业务项目并阻止归档项目继续生成。
func (s *Service) ensureTaskProjectActive(userID string, canvasOrProjectID string) error {
id := strings.TrimSpace(canvasOrProjectID)
if id == "" {
return "", 0, nil
return nil
}
if canvas, err := s.repo.CanvasProjectForUser(userID, id); err == nil {
if canvas.ProjectID == "" {
return "", 0, nil
return nil
}
project, projectErr := s.repo.ProjectForUser(userID, canvas.ProjectID)
if projectErr != nil {
return "", 0, projectErr
return projectErr
}
if project.Status == model.ProjectStatusArchived {
return "", 0, BadAuthRequest("项目已归档，无法创建生成任务")
return BadAuthRequest("项目已归档，无法创建生成任务")
}
return project.ID, normalizedProjectTaskLimit(project.ActiveTaskLimit), nil
return nil
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
return "", 0, err
return err
}
project, err := s.repo.ProjectForUser(userID, id)
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return "", 0, nil
return nil
}
return "", 0, err
return err
}
if project.Status == model.ProjectStatusArchived {
return "", 0, BadAuthRequest("项目已归档，无法创建生成任务")
return BadAuthRequest("项目已归档，无法创建生成任务")
}
return project.ID, normalizedProjectTaskLimit(project.ActiveTaskLimit), nil
}
func normalizedProjectTaskLimit(limit int) int {
if limit <= 0 {
return 3
}
return limit
return nil
}
backend/internal/service/service.go
+4 -12
View File
@@ -292,8 +292,7 @@ func (s *Service) CreateTask(userID string, req CreateTaskRequest) (*model.Task,
taskType = "video_image_to_video"
}
task := model.Task{ID: newID(), UserID: userID, SessionID: req.SessionID, ProjectID: req.ProjectID, Type: taskType, Status: model.TaskStatusQueued, Stage: "等待队列调度", Progress: 5, Prompt: prompt, Operation: req.Operation, Provider: req.Provider, Model: req.Model}
domainProjectID, projectTaskLimit, err := s.projectTaskLimit(userID, req.ProjectID)
if err != nil {
if err := s.ensureTaskProjectActive(userID, req.ProjectID); err != nil {
return nil, err
}
billingOrder, err := s.taskBillingOrder(userID, &task, normalizedInput)
@@ -308,13 +307,10 @@ func (s *Service) CreateTask(userID string, req CreateTaskRequest) (*model.Task,
if billingOrder != nil {
task.BillingOrderID = billingOrder.ID
}
err = s.createTaskWithinStorageQuota(&task, billingOrder, policy, domainProjectID, projectTaskLimit)
err = s.createTaskWithinStorageQuota(&task, billingOrder, policy)
if errors.Is(err, repository.ErrActiveTaskLimit) {
return nil, BadAuthRequest(fmt.Sprintf("同时排队或运行的任务最多 %d 个，请等待已有任务完成", policy.Task.ActiveTaskLimit))
}
if errors.Is(err, repository.ErrProjectTaskLimit) {
return nil, BadAuthRequest(fmt.Sprintf("当前项目同时排队或运行的任务最多 %d 个", projectTaskLimit))
}
if errors.Is(err, repository.ErrInsufficientCredits) {
return nil, BadAuthRequest("积分不足，请先使用兑换码充值")
}
@@ -419,20 +415,16 @@ func (s *Service) RetryTask(userID string, id string) (*model.Task, error) {
if err != nil {
return nil, err
}
domainProjectID, projectTaskLimit, err := s.projectTaskLimit(userID, task.ProjectID)
if err != nil {
if err := s.ensureTaskProjectActive(userID, task.ProjectID); err != nil {
return nil, err
}
task, err = s.repo.RetryTaskWithBilling(userID, task.ID, billingOrder, policy.Task.ActiveTaskLimit, domainProjectID, projectTaskLimit)
task, err = s.repo.RetryTaskWithBilling(userID, task.ID, billingOrder, policy.Task.ActiveTaskLimit)
if errors.Is(err, repository.ErrInsufficientCredits) {
return nil, BadAuthRequest("积分不足，请先使用兑换码充值")
}
if errors.Is(err, repository.ErrActiveTaskLimit) {
return nil, BadAuthRequest(fmt.Sprintf("同时排队或运行的任务最多 %d 个，请等待已有任务完成", policy.Task.ActiveTaskLimit))
}
if errors.Is(err, repository.ErrProjectTaskLimit) {
return nil, BadAuthRequest(fmt.Sprintf("当前项目同时排队或运行的任务最多 %d 个", projectTaskLimit))
}
if errors.Is(err, repository.ErrTaskNotRetryable) {
return nil, BadAuthRequest("任务已被其他请求重新入队，请勿重复重试")
}
backend/internal/service/storage_quota.go
+3 -3
View File
@@ -74,7 +74,7 @@ func validateStructuredReplacementQuotaWithPolicy(usage repository.UserStorageUs
return validateStructuredStorageQuotaWithPolicy(usage, kind, false, deltaBytes, policy)
}
func (s *Service) createTaskWithinStorageQuota(task *model.Task, billingOrder *model.BillingOrder, policy RuntimePolicySetting, domainProjectID string, projectTaskLimit int) error {
func (s *Service) createTaskWithinStorageQuota(task *model.Task, billingOrder *model.BillingOrder, policy RuntimePolicySetting) error {
s.storageMu.Lock()
defer s.storageMu.Unlock()
usage, err := s.repo.UserStorageUsage(task.UserID)
@@ -86,9 +86,9 @@ func (s *Service) createTaskWithinStorageQuota(task *model.Task, billingOrder *m
return err
}
if billingOrder != nil {
return s.repo.CreateTaskWithCreditReservation(task, billingOrder, policy.Task.ActiveTaskLimit, domainProjectID, projectTaskLimit)
return s.repo.CreateTaskWithCreditReservation(task, billingOrder, policy.Task.ActiveTaskLimit)
}
return s.repo.CreateTaskWithActiveLimit(task, policy.Task.ActiveTaskLimit, domainProjectID, projectTaskLimit)
return s.repo.CreateTaskWithActiveLimit(task, policy.Task.ActiveTaskLimit)
}
// 任务完成会同时扩张任务历史和 Agent 会话数据，必须在同一临界区核算并原子写入。