diff --git a/backend/internal/handler/routes.go b/backend/internal/handler/routes.go index 2c01aa0f0..731d2847a 100644 --- a/backend/internal/handler/routes.go +++ b/backend/internal/handler/routes.go @@ -144,7 +144,7 @@ func RegisterTaskRoutes(r *gin.RouterGroup, svc *service.Service) { failService(c, err) return } - task, err := svc.CancelTask(user.ID, c.Param("id")) + task, err := svc.CancelTask(c.Request.Context(), user.ID, c.Param("id")) if err != nil { fail(c, http.StatusBadRequest, err) return diff --git a/backend/internal/model/models.go b/backend/internal/model/models.go index 787825af5..1a90d6c29 100644 --- a/backend/internal/model/models.go +++ b/backend/internal/model/models.go @@ -7,6 +7,7 @@ import ( ) type TaskStatus string +type ProviderCancelStatus string type SessionStatus string type UserRole string type UserStatus string @@ -88,6 +89,10 @@ const ( BillingStatusRefunded BillingStatus = "refunded" BillingStatusUncertain BillingStatus = "uncertain" + ProviderCancelStatusRequested ProviderCancelStatus = "requested" + ProviderCancelStatusConfirmed ProviderCancelStatus = "confirmed" + ProviderCancelStatusUncertain ProviderCancelStatus = "uncertain" + CreditLedgerRedeem CreditLedgerType = "redeem" CreditLedgerAdminGrant CreditLedgerType = "admin_grant" CreditLedgerReserve CreditLedgerType = "reserve" @@ -748,33 +753,39 @@ type UserAnnouncementRead struct { } type Task struct { - ID string `json:"id" gorm:"primaryKey;size:36"` - UserID string `json:"userId" gorm:"index;size:36;index:idx_tasks_user_created,priority:1"` - SessionID string `json:"sessionId" gorm:"index;size:36"` - ProjectID string `json:"projectId" gorm:"index;size:80"` - Type string `json:"type" gorm:"index;size:64"` - Status TaskStatus `json:"status" gorm:"index;size:24;index:idx_tasks_status_created,priority:1;index:idx_tasks_claim,priority:1"` - Stage string `json:"stage" gorm:"size:80"` - Progress int `json:"progress"` - Prompt string `json:"prompt"` - Operation string `json:"operation" gorm:"size:64"` - Provider string `json:"provider" gorm:"size:64"` - Model string `json:"model" gorm:"size:120"` - BillingOrderID string `json:"billingOrderId,omitempty" gorm:"index;size:36"` - ProviderRequestID string `json:"providerRequestId,omitempty" gorm:"index;size:160"` - PollStage string `json:"pollStage,omitempty" gorm:"size:32"` - NextPollAt *time.Time `json:"nextPollAt,omitempty" gorm:"index"` - LeaseOwner string `json:"-" gorm:"index;size:120"` - LeaseExpiresAt *time.Time `json:"-" gorm:"index;index:idx_tasks_claim,priority:2"` - InputJSON string `json:"inputJson" gorm:"type:text"` - ResultJSON string `json:"resultJson" gorm:"type:text"` - TextDraft string `json:"textDraft,omitempty" gorm:"type:text"` - Error string `json:"error"` - Attempts int `json:"attempts"` - StartedAt *time.Time `json:"startedAt"` - CompletedAt *time.Time `json:"completedAt"` - CreatedAt time.Time `json:"createdAt" gorm:"index:idx_tasks_user_created,priority:2;index:idx_tasks_status_created,priority:2;index:idx_tasks_claim,priority:3"` - UpdatedAt time.Time `json:"updatedAt"` + ID string `json:"id" gorm:"primaryKey;size:36"` + UserID string `json:"userId" gorm:"index;size:36;index:idx_tasks_user_created,priority:1"` + SessionID string `json:"sessionId" gorm:"index;size:36"` + ProjectID string `json:"projectId" gorm:"index;size:80"` + Type string `json:"type" gorm:"index;size:64"` + Status TaskStatus `json:"status" gorm:"index;size:24;index:idx_tasks_status_created,priority:1;index:idx_tasks_claim,priority:1;index:idx_tasks_provider_cancel,priority:1"` + Stage string `json:"stage" gorm:"size:80"` + Progress int `json:"progress"` + Prompt string `json:"prompt"` + Operation string `json:"operation" gorm:"size:64"` + Provider string `json:"provider" gorm:"size:64"` + Model string `json:"model" gorm:"size:120"` + BillingOrderID string `json:"billingOrderId,omitempty" gorm:"index;size:36"` + ProviderRequestID string `json:"providerRequestId,omitempty" gorm:"index;size:160"` + ProviderCancelStatus ProviderCancelStatus `json:"providerCancelStatus,omitempty" gorm:"index;size:24;index:idx_tasks_provider_cancel,priority:2"` + ProviderCancelError string `json:"providerCancelError,omitempty" gorm:"type:text"` + ProviderCancelAttempts int `json:"providerCancelAttempts,omitempty"` + ProviderCancelRequestedAt *time.Time `json:"providerCancelRequestedAt,omitempty"` + ProviderCancelledAt *time.Time `json:"providerCancelledAt,omitempty"` + ProviderCancelNextCheckAt *time.Time `json:"providerCancelNextCheckAt,omitempty" gorm:"index:idx_tasks_provider_cancel,priority:3"` + PollStage string `json:"pollStage,omitempty" gorm:"size:32"` + NextPollAt *time.Time `json:"nextPollAt,omitempty" gorm:"index"` + LeaseOwner string `json:"-" gorm:"index;size:120"` + LeaseExpiresAt *time.Time `json:"-" gorm:"index;index:idx_tasks_claim,priority:2"` + InputJSON string `json:"inputJson" gorm:"type:text"` + ResultJSON string `json:"resultJson" gorm:"type:text"` + TextDraft string `json:"textDraft,omitempty" gorm:"type:text"` + Error string `json:"error"` + Attempts int `json:"attempts"` + StartedAt *time.Time `json:"startedAt"` + CompletedAt *time.Time `json:"completedAt"` + CreatedAt time.Time `json:"createdAt" gorm:"index:idx_tasks_user_created,priority:2;index:idx_tasks_status_created,priority:2;index:idx_tasks_claim,priority:3"` + UpdatedAt time.Time `json:"updatedAt"` } // TaskTextDelta 只保存可回放窗口内的文本增量；最终正文和失败草稿分别归并到 Task.ResultJSON 与 Task.TextDraft。 diff --git a/backend/internal/repository/finance.go b/backend/internal/repository/finance.go index cccaa0dc1..a85d542dc 100644 --- a/backend/internal/repository/finance.go +++ b/backend/internal/repository/finance.go @@ -224,7 +224,11 @@ func (r *Repository) RetryTaskWithBilling(userID string, taskID string, order *m } updates := map[string]any{ "status": model.TaskStatusQueued, "stage": "等待队列调度", "progress": 5, "error": "", "result_json": "", - "text_draft": "", "started_at": nil, "completed_at": nil, "updated_at": time.Now(), + "text_draft": "", "started_at": nil, "completed_at": nil, + "provider_request_id": "", "poll_stage": "", "next_poll_at": nil, + "provider_cancel_status": "", "provider_cancel_error": "", "provider_cancel_attempts": 0, + "provider_cancel_requested_at": nil, "provider_cancelled_at": nil, "provider_cancel_next_check_at": nil, + "lease_owner": "", "lease_expires_at": nil, "updated_at": time.Now(), } if order != nil { updates["billing_order_id"] = order.ID diff --git a/backend/internal/repository/provider_task_cancellation_test.go b/backend/internal/repository/provider_task_cancellation_test.go new file mode 100644 index 000000000..ec646b07c --- /dev/null +++ b/backend/internal/repository/provider_task_cancellation_test.go @@ -0,0 +1,41 @@ +package repository + +import ( + "errors" + "testing" + "time" + + "infinite-canvas/backend/internal/model" + + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +func TestClaimTaskProviderCancellationOnlySucceedsOnce(t *testing.T) { + db, err := gorm.Open(sqlite.Open("file:provider-cancel-claim?mode=memory&cache=shared"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.Task{}); err != nil { + t.Fatal(err) + } + task := model.Task{ID: "task-1", UserID: "user-1", Status: model.TaskStatusCancelled} + if err := db.Create(&task).Error; err != nil { + t.Fatal(err) + } + + repo := New(db) + if err := repo.ClaimTaskProviderCancellation(task.UserID, task.ID, time.Now()); err != nil { + t.Fatal(err) + } + if err := repo.ClaimTaskProviderCancellation(task.UserID, task.ID, time.Now()); !errors.Is(err, ErrTaskProviderCancellationConflict) { + t.Fatalf("expected cancellation claim conflict, got %v", err) + } + stored, err := repo.Task(task.ID) + if err != nil { + t.Fatal(err) + } + if stored.ProviderCancelStatus != model.ProviderCancelStatusRequested || stored.ProviderCancelAttempts != 1 { + t.Fatalf("unexpected cancellation state: status=%s attempts=%d", stored.ProviderCancelStatus, stored.ProviderCancelAttempts) + } +} diff --git a/backend/internal/repository/repository.go b/backend/internal/repository/repository.go index cddc02147..ba19e1c1c 100644 --- a/backend/internal/repository/repository.go +++ b/backend/internal/repository/repository.go @@ -16,6 +16,10 @@ var ErrDailyUploadLimitExceeded = errors.New("daily upload limit exceeded") var ErrTaskProviderRecoveryConflict = errors.New("task provider recovery is already running") +var ErrTaskProviderCancellationConflict = errors.New("task provider cancellation is already claimed") + +var ErrTaskStateConflict = errors.New("task state changed concurrently") + var ErrTextReplayQuotaExceeded = errors.New("text replay quota exceeded") var ErrTextReplayClosed = errors.New("text replay task is closed") @@ -382,15 +386,21 @@ func (r *Repository) ReleaseTaskProviderRecovery(id string, owner string) error } func (r *Repository) UpdateTaskProgress(id string, stage string, progress int) error { - return r.db.Model(&model.Task{}).Where("id = ?", id).Updates(map[string]any{ + return r.db.Model(&model.Task{}).Where("id = ? AND status = ?", id, model.TaskStatusRunning).Updates(map[string]any{ "stage": stage, "progress": progress, "updated_at": time.Now(), }).Error } -func (r *Repository) SaveTaskCompletion(task *model.Task, session *model.Session, message *model.Message, results []model.Result) error { +func (r *Repository) SaveTaskCompletion(task *model.Task, expected model.TaskStatus, session *model.Session, message *model.Message, results []model.Result) error { return r.db.Transaction(func(tx *gorm.DB) error { - if err := tx.Save(task).Error; err != nil { - return err + updated := tx.Model(&model.Task{}). + Where("id = ? AND status = ?", task.ID, expected). + Select("*").Omit("id", "created_at").Updates(task) + if updated.Error != nil { + return updated.Error + } + if updated.RowsAffected != 1 { + return ErrTaskStateConflict } if session != nil { if err := tx.Save(session).Error; err != nil { @@ -411,21 +421,112 @@ func (r *Repository) SaveTaskCompletion(task *model.Task, session *model.Session }) } +func (r *Repository) UpdateTaskTerminalState(id string, expected model.TaskStatus, status model.TaskStatus, stage string, errorText string, completedAt time.Time) (bool, error) { + result := r.db.Model(&model.Task{}). + Where("id = ? AND status = ?", id, expected). + Updates(map[string]any{ + "status": status, "stage": stage, "error": errorText, "completed_at": &completedAt, + "lease_owner": "", "lease_expires_at": nil, "updated_at": completedAt, + }) + return result.RowsAffected == 1, result.Error +} + func (r *Repository) CancelTaskIfStatus(userID string, id string, expected model.TaskStatus, now time.Time) (bool, error) { result := r.db.Model(&model.Task{}). Where("id = ? AND user_id = ? AND status = ?", id, userID, expected). Updates(map[string]any{ - "status": model.TaskStatusCancelled, "stage": "任务已取消", "completed_at": &now, "updated_at": now, + "status": model.TaskStatusCancelled, "stage": "任务已取消", "completed_at": &now, + "lease_owner": "", "lease_expires_at": nil, "updated_at": now, }) return result.RowsAffected == 1, result.Error } +// 上游取消先落库再发请求；条件更新保证并发和重复取消只有一个调用方取得发送权。 +func (r *Repository) ClaimTaskProviderCancellation(userID string, id string, now time.Time) error { + result := r.db.Model(&model.Task{}). + Where("id = ? AND user_id = ? AND status = ? AND provider_cancel_status = ''", id, userID, model.TaskStatusCancelled). + Updates(map[string]any{ + "provider_cancel_status": model.ProviderCancelStatusRequested, + "provider_cancel_attempts": 1, + "provider_cancel_requested_at": &now, + "provider_cancel_next_check_at": now.Add(15 * time.Second), + "provider_cancel_error": "", + "updated_at": now, + }) + if result.Error != nil { + return result.Error + } + if result.RowsAffected != 1 { + return ErrTaskProviderCancellationConflict + } + return nil +} + +func (r *Repository) UpdateTaskProviderCancellation(id string, expected model.ProviderCancelStatus, status model.ProviderCancelStatus, errorText string, nextCheckAt *time.Time, cancelledAt *time.Time) error { + updates := map[string]any{ + "provider_cancel_status": status, + "provider_cancel_error": errorText, + "provider_cancel_next_check_at": nextCheckAt, + "lease_owner": "", + "lease_expires_at": nil, + "updated_at": time.Now(), + } + if cancelledAt != nil { + updates["provider_cancelled_at"] = cancelledAt + } + result := r.db.Model(&model.Task{}).Where("id = ? AND status = ? AND provider_cancel_status = ?", id, model.TaskStatusCancelled, expected).Updates(updates) + if result.Error != nil { + return result.Error + } + if result.RowsAffected != 1 { + return ErrTaskProviderCancellationConflict + } + return nil +} + +// 对账任务同样使用数据库租约，多实例和服务重启后只会有一个 worker 查询同一上游任务。 +func (r *Repository) ClaimNextTaskProviderCancellation(owner string, leaseDuration time.Duration) (*model.Task, error) { + var task model.Task + now := time.Now() + err := r.db.Transaction(func(tx *gorm.DB) error { + query := tx.Where( + "status = ? AND provider_cancel_status = ? AND (provider_cancel_next_check_at IS NULL OR provider_cancel_next_check_at <= ?) AND (lease_expires_at IS NULL OR lease_expires_at <= ?)", + model.TaskStatusCancelled, model.ProviderCancelStatusRequested, now, now, + ).Order("provider_cancel_requested_at asc").Limit(1) + if r.Dialect() == "postgres" { + query = query.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}) + } + if result := query.Find(&task); result.Error != nil || result.RowsAffected == 0 { + task = model.Task{} + return result.Error + } + claim := tx.Model(&model.Task{}).Where( + "id = ? AND status = ? AND provider_cancel_status = ? AND (lease_expires_at IS NULL OR lease_expires_at <= ?)", + task.ID, model.TaskStatusCancelled, model.ProviderCancelStatusRequested, now, + ).Updates(map[string]any{ + "lease_owner": owner, + "lease_expires_at": now.Add(leaseDuration), + "provider_cancel_attempts": gorm.Expr("provider_cancel_attempts + 1"), + "updated_at": now, + }) + if claim.Error != nil || claim.RowsAffected == 0 { + task = model.Task{} + return claim.Error + } + return tx.First(&task, "id = ?", task.ID).Error + }) + if err != nil || task.ID == "" { + return nil, err + } + return &task, nil +} + func (r *Repository) Tasks(userID string, limit int, projectID string, activeOnly bool) ([]model.Task, error) { var tasks []model.Task if limit <= 0 || limit > 100 { limit = 50 } - query := r.db.Select("id", "session_id", "project_id", "type", "status", "stage", "progress", "prompt", "operation", "provider", "model", "input_json", "result_json", "billing_order_id", "attempts", "started_at", "completed_at", "created_at", "updated_at"). + query := r.db.Select("id", "session_id", "project_id", "type", "status", "stage", "progress", "prompt", "operation", "provider", "model", "input_json", "result_json", "billing_order_id", "provider_request_id", "provider_cancel_status", "provider_cancel_error", "provider_cancel_attempts", "provider_cancel_requested_at", "provider_cancelled_at", "provider_cancel_next_check_at", "attempts", "started_at", "completed_at", "created_at", "updated_at"). Where("user_id = ?", userID) if strings.TrimSpace(projectID) != "" { query = query.Where("project_id = ?", strings.TrimSpace(projectID)) diff --git a/backend/internal/service/analytics.go b/backend/internal/service/analytics.go index d3a662e4b..85f05b14e 100644 --- a/backend/internal/service/analytics.go +++ b/backend/internal/service/analytics.go @@ -953,7 +953,7 @@ func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) { log.OutputTokens = firstInt64(usageMetadata, "candidatesTokenCount") log.CachedTokens = firstInt64(usageMetadata, "cachedContentTokenCount") } - log.ProviderRequestID = firstNonEmpty(stringField(payload, "task_id"), stringField(payload, "id"), stringField(payload, "request_id"), log.ProviderRequestID) + log.ProviderRequestID = firstNonEmpty(stringField(payload, "task_id"), stringField(payload, "id"), stringField(payload, "request_id"), stringField(payload, "name"), log.ProviderRequestID) log.ProviderStatus = strings.ToLower(firstNonEmpty(stringField(payload, "status"), log.ProviderStatus)) if log.ProviderStatus == "failed" || log.ProviderStatus == "cancelled" || log.ProviderStatus == "expired" { log.Status = model.ApiCallStatusFailed diff --git a/backend/internal/service/provider.go b/backend/internal/service/provider.go index 168f63025..f65bed745 100644 --- a/backend/internal/service/provider.go +++ b/backend/internal/service/provider.go @@ -2021,7 +2021,7 @@ func recordProviderRequest(req *http.Request, startedAt time.Time, statusCode in log := model.ApiCallLog{ UserID: metadata.UserID, ChannelID: metadata.ChannelID, TaskID: metadata.TaskID, BillingOrderID: metadata.BillingOrderID, Source: "backend-task", Capability: metadata.Capability, Operation: metadata.Operation, - RequestKind: requestKind, Billable: req.Method == http.MethodPost, + RequestKind: requestKind, Billable: req.Method == http.MethodPost && requestKind != "cancel", APIFormat: apiFormat, Method: req.Method, Path: req.URL.Path, Model: metadata.Model, Status: status, StatusCode: statusCode, DurationMs: time.Since(startedAt).Milliseconds(), Error: errorText, ConcurrencyLimit: metadata.ConcurrencyLimit, UpstreamURL: req.URL.Scheme + "://" + req.URL.Host + req.URL.Path, diff --git a/backend/internal/service/provider_task_cancellation.go b/backend/internal/service/provider_task_cancellation.go new file mode 100644 index 000000000..40fd0618b --- /dev/null +++ b/backend/internal/service/provider_task_cancellation.go @@ -0,0 +1,261 @@ +package service + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/url" + "strings" + "time" + + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/repository" +) + +const ( + providerCancellationLeaseDuration = time.Minute + providerCancellationMaxAttempts = 41 +) + +type providerCancellationOutcome string + +const ( + providerCancellationPending providerCancellationOutcome = "pending" + providerCancellationConfirmed providerCancellationOutcome = "confirmed" + providerCancellationSucceeded providerCancellationOutcome = "succeeded" + providerCancellationFailed providerCancellationOutcome = "failed" +) + +func (s *Service) startProviderCancellationReconciliation() { + go func() { + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + for range ticker.C { + task, err := s.repo.ClaimNextTaskProviderCancellation("provider-cancel:"+s.workerID, providerCancellationLeaseDuration) + if err != nil || task == nil { + continue + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + _ = s.reconcileProviderCancellation(ctx, task) + cancel() + } + }() +} + +func (s *Service) requestProviderCancellation(ctx context.Context, task *model.Task) error { + if task == nil || task.ID == "" { + return errors.New("任务取消状态无效") + } + s.hydrateTaskProviderRequestID(task) + if task.ProviderRequestID != "" { + if err := s.repo.UpdateTaskProviderState(task.ID, task.ProviderRequestID, task.PollStage, task.NextPollAt); err != nil { + return err + } + } + if err := s.repo.ClaimTaskProviderCancellation(task.UserID, task.ID, time.Now()); err != nil { + if errors.Is(err, repository.ErrTaskProviderCancellationConflict) { + return nil + } + return err + } + if task.ProviderRequestID == "" { + return s.markProviderCancellationUncertain(task, "上游未返回任务 ID，无法发送取消请求，费用待核对") + } + + input, err := s.providerCancellationInput(task) + if err != nil { + return s.markProviderCancellationUncertain(task, "读取上游取消配置失败，费用待核对："+err.Error()) + } + if !supportsProviderCancellation(input.Config.InterfaceType) { + return s.markProviderCancellationUncertain(task, "当前上游协议不支持取消，费用待核对") + } + + requestCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) + defer cancel() + requestTask := *task + requestTask.InputJSON = mustJSON(input) + requestCtx = withProviderRequestKind(withProviderAnalytics(requestCtx, s, requestTask), "cancel") + if err := cancelProviderTask(requestCtx, input.Config, task.ProviderRequestID); err != nil { + return s.markProviderCancellationUncertain(task, "上游取消请求结果不明确，费用待核对："+safeProviderLogError(err)) + } + _ = s.log(task.UserID, task.ID, "info", "已请求上游取消，等待供应商确认", task.ProviderRequestID) + return nil +} + +func (s *Service) reconcileProviderCancellation(ctx context.Context, task *model.Task) error { + if task.ProviderCancelAttempts >= providerCancellationMaxAttempts { + return s.markProviderCancellationUncertain(task, "上游取消状态长时间未确认，费用待核对") + } + input, err := s.providerCancellationInput(task) + if err != nil { + return s.markProviderCancellationUncertain(task, "读取上游对账配置失败，费用待核对："+err.Error()) + } + queryTask := *task + queryTask.InputJSON = mustJSON(input) + queryCtx := withProviderRequestKind(withProviderAnalytics(ctx, s, queryTask), "cancel-query") + outcome, providerStatus, err := queryProviderCancellation(queryCtx, input.Config, task.ProviderRequestID) + if err != nil { + if task.ProviderCancelAttempts >= providerCancellationMaxAttempts-1 { + return s.markProviderCancellationUncertain(task, "连续查询上游取消状态失败，费用待核对："+safeProviderLogError(err)) + } + next := time.Now().Add(30 * time.Second) + return s.repo.UpdateTaskProviderCancellation(task.ID, model.ProviderCancelStatusRequested, model.ProviderCancelStatusRequested, safeProviderLogError(err), &next, nil) + } + + switch outcome { + case providerCancellationConfirmed: + if err := s.RefundBilling(task.BillingOrderID, "上游已确认取消"); err != nil { + return s.deferProviderCancellation(task, "上游已取消，但积分退回失败："+err.Error()) + } + now := time.Now() + if err := s.repo.UpdateTaskProviderCancellation(task.ID, model.ProviderCancelStatusRequested, model.ProviderCancelStatusConfirmed, "", nil, &now); err != nil { + return err + } + _ = s.log(task.UserID, task.ID, "info", "上游已确认取消，积分已退回", providerStatus) + case providerCancellationSucceeded: + errorText := "取消请求发出前上游已完成，费用已结算" + if err := s.SettleBilling(task.BillingOrderID, task.ProviderRequestID); err != nil { + errorText = "取消请求发出前上游已完成，但费用结算失败，需人工核对：" + err.Error() + _ = s.MarkBillingUncertain(task.BillingOrderID, errorText) + } + if err := s.repo.UpdateTaskProviderCancellation(task.ID, model.ProviderCancelStatusRequested, model.ProviderCancelStatusUncertain, errorText, nil, nil); err != nil { + return err + } + _ = s.log(task.UserID, task.ID, "warn", "上游任务在取消确认前已完成", providerStatus) + case providerCancellationFailed: + errorText := "上游任务已失败，取消状态无法确认，积分已退回" + if err := s.RefundBilling(task.BillingOrderID, errorText); err != nil { + return s.deferProviderCancellation(task, "上游任务已失败，但积分退回失败："+err.Error()) + } + if err := s.repo.UpdateTaskProviderCancellation(task.ID, model.ProviderCancelStatusRequested, model.ProviderCancelStatusUncertain, errorText, nil, nil); err != nil { + return err + } + _ = s.log(task.UserID, task.ID, "warn", errorText, providerStatus) + default: + next := time.Now().Add(15 * time.Second) + return s.repo.UpdateTaskProviderCancellation(task.ID, model.ProviderCancelStatusRequested, model.ProviderCancelStatusRequested, "", &next, nil) + } + return nil +} + +func (s *Service) deferProviderCancellation(task *model.Task, errorText string) error { + next := time.Now().Add(30 * time.Second) + return s.repo.UpdateTaskProviderCancellation(task.ID, model.ProviderCancelStatusRequested, model.ProviderCancelStatusRequested, truncateRunes(errorText, 1000), &next, nil) +} + +func (s *Service) markProviderCancellationUncertain(task *model.Task, errorText string) error { + message := truncateRunes(errorText, 1000) + if err := s.repo.UpdateTaskProviderCancellation(task.ID, model.ProviderCancelStatusRequested, model.ProviderCancelStatusUncertain, message, nil, nil); err != nil { + return err + } + if err := s.MarkBillingUncertain(task.BillingOrderID, message); err != nil { + _ = s.log(task.UserID, task.ID, "error", "上游取消已转入人工核对，但计费状态更新失败", err.Error()) + return err + } + _ = s.log(task.UserID, task.ID, "warn", "上游取消无法确认", message) + return nil +} + +func (s *Service) providerCancellationInput(task *model.Task) (canvasGenerationInput, error) { + decrypted, err := s.decryptTaskInputJSON(task.InputJSON) + if err != nil { + return canvasGenerationInput{}, err + } + var input canvasGenerationInput + if err := json.Unmarshal([]byte(decrypted), &input); err != nil { + return canvasGenerationInput{}, err + } + config, err := s.resolveProviderConfig(input.Config) + if err != nil { + return canvasGenerationInput{}, err + } + input.Config = config + return input, nil +} + +func supportsProviderCancellation(interfaceType string) bool { + return interfaceType == string(model.ChannelInterfaceGeminiVeo) || interfaceType == string(model.ChannelInterfaceVolcengineArkVideo) +} + +func cancelProviderTask(ctx context.Context, config providerConfig, providerRequestID string) error { + switch config.InterfaceType { + case string(model.ChannelInterfaceGeminiVeo): + path := "/" + strings.TrimLeft(providerRequestID, "/") + ":cancel" + return postGeminiJSON(ctx, config, path, map[string]any{}, &map[string]any{}) + case string(model.ChannelInterfaceVolcengineArkVideo): + path := "/contents/generations/tasks/" + url.PathEscape(providerRequestID) + return deleteProviderTask(ctx, config, path) + default: + return errors.New("当前上游协议不支持取消") + } +} + +func deleteProviderTask(ctx context.Context, config providerConfig, path string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodDelete, apiURL(config.BaseURL, path), nil) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+config.APIKey) + ApplyOutboundHeaders(req, config.Headers) + data, _, err := doBinary(req) + if err != nil { + return err + } + if len(bytes.TrimSpace(data)) > 0 && !json.Valid(data) { + return errors.New("上游取消接口返回了无法识别的内容") + } + return nil +} + +func queryProviderCancellation(ctx context.Context, config providerConfig, providerRequestID string) (providerCancellationOutcome, string, error) { + switch config.InterfaceType { + case string(model.ChannelInterfaceGeminiVeo): + var operation map[string]any + if err := getGeminiJSON(ctx, config, "/"+strings.TrimLeft(providerRequestID, "/"), &operation); err != nil { + return "", "", err + } + if done, _ := operation["done"].(bool); !done { + return providerCancellationPending, "running", nil + } + if errorValue, ok := operation["error"].(map[string]any); ok { + code := strings.TrimSpace(fmt.Sprint(errorValue["code"])) + message := strings.ToLower(stringField(errorValue, "message")) + if code == "1" || code == "1.0" || strings.Contains(message, "cancel") { + return providerCancellationConfirmed, "cancelled", nil + } + return providerCancellationFailed, firstNonEmpty(message, "failed"), nil + } + return providerCancellationSucceeded, "succeeded", nil + case string(model.ChannelInterfaceVolcengineArkVideo): + var state map[string]any + if err := getJSON(ctx, config, "/contents/generations/tasks/"+url.PathEscape(providerRequestID), &state); err != nil { + return "", "", err + } + if data, ok := state["data"].(map[string]any); ok { + state = data + } + status := strings.ToLower(strings.TrimSpace(stringField(state, "status"))) + switch status { + case "cancelled", "canceled": + return providerCancellationConfirmed, status, nil + case "succeeded", "completed": + return providerCancellationSucceeded, status, nil + case "failed", "expired": + return providerCancellationFailed, status, nil + case "queued", "running", "processing", "pending": + return providerCancellationPending, status, nil + default: + return providerCancellationPending, firstNonEmpty(status, "unknown"), nil + } + default: + return "", "", errors.New("当前上游协议不支持取消状态查询") + } +} + +func mustJSON(value any) string { + data, _ := json.Marshal(value) + return string(data) +} diff --git a/backend/internal/service/provider_task_cancellation_test.go b/backend/internal/service/provider_task_cancellation_test.go new file mode 100644 index 000000000..f6ecf6568 --- /dev/null +++ b/backend/internal/service/provider_task_cancellation_test.go @@ -0,0 +1,63 @@ +package service + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestGeminiProviderCancellationAndConfirmation(t *testing.T) { + requests := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("x-goog-api-key") != "test-key" { + t.Errorf("missing Gemini API key") + w.WriteHeader(http.StatusUnauthorized) + return + } + switch { + case r.Method == http.MethodPost && r.URL.Path == "/v1beta/operations/video-1:cancel": + requests++ + _ = json.NewEncoder(w).Encode(map[string]any{}) + case r.Method == http.MethodGet && r.URL.Path == "/v1beta/operations/video-1": + _ = json.NewEncoder(w).Encode(map[string]any{"done": true, "error": map[string]any{"code": 1, "message": "Operation cancelled"}}) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + config := providerConfig{InterfaceType: "gemini-veo", BaseURL: server.URL, APIKey: "test-key"} + if err := cancelProviderTask(context.Background(), config, "operations/video-1"); err != nil { + t.Fatal(err) + } + outcome, _, err := queryProviderCancellation(context.Background(), config, "operations/video-1") + if err != nil { + t.Fatal(err) + } + if requests != 1 || outcome != providerCancellationConfirmed { + t.Fatalf("unexpected Gemini cancellation result: requests=%d outcome=%s", requests, outcome) + } +} + +func TestVolcengineArkProviderCancellationUsesDelete(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodDelete || r.URL.Path != "/api/v3/contents/generations/tasks/task-1" { + http.NotFound(w, r) + return + } + if r.Header.Get("Authorization") != "Bearer test-key" { + t.Errorf("missing Ark bearer token") + w.WriteHeader(http.StatusUnauthorized) + return + } + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + config := providerConfig{InterfaceType: "volcengine-ark-video", BaseURL: server.URL + "/api/v3", APIKey: "test-key"} + if err := cancelProviderTask(context.Background(), config, "task-1"); err != nil { + t.Fatal(err) + } +} diff --git a/backend/internal/service/service.go b/backend/internal/service/service.go index 4c91a7566..c3483042b 100644 --- a/backend/internal/service/service.go +++ b/backend/internal/service/service.go @@ -71,28 +71,33 @@ type SessionDetail struct { } type TaskSummary struct { - ID string `json:"id"` - SessionID string `json:"sessionId,omitempty"` - ProjectID string `json:"projectId,omitempty"` - Type string `json:"type"` - Status model.TaskStatus `json:"status"` - Stage string `json:"stage"` - Progress int `json:"progress"` - Prompt string `json:"prompt"` - Operation string `json:"operation,omitempty"` - Provider string `json:"provider,omitempty"` - Model string `json:"model,omitempty"` - ProviderRequestID string `json:"providerRequestId,omitempty"` - ErrorCode string `json:"errorCode,omitempty"` - PreviewURL string `json:"previewUrl,omitempty"` - PreviewKind string `json:"previewKind,omitempty"` - Attempts int `json:"attempts"` - StartedAt *time.Time `json:"startedAt"` - CompletedAt *time.Time `json:"completedAt"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` - Billing *TaskBillingSummary `json:"billing,omitempty"` - ClientContext *TaskClientContext `json:"clientContext,omitempty"` + ID string `json:"id"` + SessionID string `json:"sessionId,omitempty"` + ProjectID string `json:"projectId,omitempty"` + Type string `json:"type"` + Status model.TaskStatus `json:"status"` + Stage string `json:"stage"` + Progress int `json:"progress"` + Prompt string `json:"prompt"` + Operation string `json:"operation,omitempty"` + Provider string `json:"provider,omitempty"` + Model string `json:"model,omitempty"` + ProviderRequestID string `json:"providerRequestId,omitempty"` + ProviderCancelStatus model.ProviderCancelStatus `json:"providerCancelStatus,omitempty"` + ProviderCancelError string `json:"providerCancelError,omitempty"` + ProviderCancelAttempts int `json:"providerCancelAttempts,omitempty"` + ProviderCancelRequestedAt *time.Time `json:"providerCancelRequestedAt,omitempty"` + ProviderCancelledAt *time.Time `json:"providerCancelledAt,omitempty"` + ErrorCode string `json:"errorCode,omitempty"` + PreviewURL string `json:"previewUrl,omitempty"` + PreviewKind string `json:"previewKind,omitempty"` + Attempts int `json:"attempts"` + StartedAt *time.Time `json:"startedAt"` + CompletedAt *time.Time `json:"completedAt"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` + Billing *TaskBillingSummary `json:"billing,omitempty"` + ClientContext *TaskClientContext `json:"clientContext,omitempty"` } type TaskClientContext struct { @@ -190,6 +195,7 @@ func New(repo *repository.Repository, dataDir string) *Service { func (s *Service) StartWorker() { s.startTextReplayCleanup() + s.startProviderCancellationReconciliation() go func() { slots := make(chan struct{}, maxChannelConcurrencyLimit) dispatch := func() { @@ -460,6 +466,12 @@ func (s *Service) refreshTaskProviderState(task *model.Task) error { } task.PollStage = latest.PollStage task.NextPollAt = latest.NextPollAt + task.ProviderCancelStatus = latest.ProviderCancelStatus + task.ProviderCancelError = latest.ProviderCancelError + task.ProviderCancelAttempts = latest.ProviderCancelAttempts + task.ProviderCancelRequestedAt = latest.ProviderCancelRequestedAt + task.ProviderCancelledAt = latest.ProviderCancelledAt + task.ProviderCancelNextCheckAt = latest.ProviderCancelNextCheckAt return nil } @@ -471,6 +483,9 @@ func (s *Service) RetryTask(userID string, id string) (*model.Task, error) { if task.Status != model.TaskStatusFailed && task.Status != model.TaskStatusCancelled { return nil, errors.New("only failed or cancelled tasks can be retried") } + if task.ProviderCancelStatus == model.ProviderCancelStatusRequested { + return nil, BadAuthRequest("上游取消状态仍在确认中，请确认费用结果后再重试") + } if isContentModerationFailure(task.Error) { return nil, BadAuthRequest(contentModerationRetryMessage) } @@ -517,7 +532,7 @@ func (s *Service) RetryTask(userID string, id string) (*model.Task, error) { return taskForOutput(*task), nil } -func (s *Service) CancelTask(userID string, id string) (*model.Task, error) { +func (s *Service) CancelTask(ctx context.Context, userID string, id string) (*model.Task, error) { task, err := s.repo.TaskForUser(userID, id) if err != nil { return nil, err @@ -526,6 +541,7 @@ func (s *Service) CancelTask(userID string, id string) (*model.Task, error) { return nil, errors.New("completed task cannot be cancelled") } now := time.Now() + cancelledRunningTask := false if task.Status == model.TaskStatusQueued { cancelled, err := s.repo.CancelTaskIfStatus(userID, task.ID, model.TaskStatusQueued, now) if err != nil { @@ -547,7 +563,6 @@ func (s *Service) CancelTask(userID string, id string) (*model.Task, error) { } } if task.Status == model.TaskStatusRunning { - s.cancelActiveTask(task.ID) cancelled, err := s.repo.CancelTaskIfStatus(userID, task.ID, model.TaskStatusRunning, now) if err != nil { return nil, err @@ -562,6 +577,8 @@ func (s *Service) CancelTask(userID string, id string) (*model.Task, error) { } task = latest } else { + cancelledRunningTask = true + s.cancelActiveTask(task.ID) if err := s.MarkBillingUncertain(task.BillingOrderID, "运行中的上游请求被用户取消，费用状态待核对"); err != nil { return nil, err } @@ -577,6 +594,15 @@ func (s *Service) CancelTask(userID string, id string) (*model.Task, error) { if task.SessionID != "" { _ = s.markSessionFailed(*task, "会话任务已取消。") } + if cancelledRunningTask { + if err := s.requestProviderCancellation(ctx, task); err != nil { + return nil, err + } + task, err = s.repo.TaskForUser(userID, id) + if err != nil { + return nil, err + } + } if err := s.finalizeTaskTextReplay(task.ID, model.TaskStatusCancelled); err != nil { _ = s.log(userID, task.ID, "error", "文本回放草稿归并失败", err.Error()) } @@ -630,27 +656,32 @@ func taskSummaryForOutput(task model.Task) TaskSummary { } previewURL, previewKind := taskMediaPreview(task.ResultJSON, task.Type) return TaskSummary{ - ID: task.ID, - SessionID: task.SessionID, - ProjectID: task.ProjectID, - Type: task.Type, - Status: task.Status, - Stage: task.Stage, - Progress: task.Progress, - Prompt: truncateRunes(task.Prompt, 500), - Operation: task.Operation, - Provider: task.Provider, - Model: task.Model, - ProviderRequestID: task.ProviderRequestID, - ErrorCode: errorCode, - PreviewURL: previewURL, - PreviewKind: previewKind, - Attempts: task.Attempts, - StartedAt: task.StartedAt, - CompletedAt: task.CompletedAt, - CreatedAt: task.CreatedAt, - UpdatedAt: task.UpdatedAt, - ClientContext: taskClientContext(task.InputJSON), + ID: task.ID, + SessionID: task.SessionID, + ProjectID: task.ProjectID, + Type: task.Type, + Status: task.Status, + Stage: task.Stage, + Progress: task.Progress, + Prompt: truncateRunes(task.Prompt, 500), + Operation: task.Operation, + Provider: task.Provider, + Model: task.Model, + ProviderRequestID: task.ProviderRequestID, + ProviderCancelStatus: task.ProviderCancelStatus, + ProviderCancelError: task.ProviderCancelError, + ProviderCancelAttempts: task.ProviderCancelAttempts, + ProviderCancelRequestedAt: task.ProviderCancelRequestedAt, + ProviderCancelledAt: task.ProviderCancelledAt, + ErrorCode: errorCode, + PreviewURL: previewURL, + PreviewKind: previewKind, + Attempts: task.Attempts, + StartedAt: task.StartedAt, + CompletedAt: task.CompletedAt, + CreatedAt: task.CreatedAt, + UpdatedAt: task.UpdatedAt, + ClientContext: taskClientContext(task.InputJSON), } } @@ -877,7 +908,7 @@ func (s *Service) processClaimedTask(task *model.Task) error { task.Stage = "计费准备失败" task.Error = taskFailureMessage(err) task.CompletedAt = ptr(time.Now()) - _ = s.repo.Save(task) + _, _ = s.repo.UpdateTaskTerminalState(task.ID, model.TaskStatusRunning, task.Status, task.Stage, task.Error, *task.CompletedAt) _ = s.RefundBilling(task.BillingOrderID, "计费准备失败，上游请求未发出") return err } @@ -908,7 +939,7 @@ func (s *Service) processClaimedTask(task *model.Task) error { task.Stage = "任务已取消" task.Error = "任务已取消" task.CompletedAt = ptr(time.Now()) - _ = s.repo.Save(task) + _, _ = s.repo.UpdateTaskTerminalState(task.ID, model.TaskStatusRunning, task.Status, task.Stage, task.Error, *task.CompletedAt) if channelSlotFailedBeforeRequest { _ = s.RefundBilling(task.BillingOrderID, "等待渠道槽位期间取消，上游请求未发出") } else { @@ -928,7 +959,7 @@ func (s *Service) processClaimedTask(task *model.Task) error { task.Stage = "任务失败" task.Error = taskFailureMessage(err) task.CompletedAt = ptr(time.Now()) - _ = s.repo.Save(task) + _, _ = s.repo.UpdateTaskTerminalState(task.ID, model.TaskStatusRunning, task.Status, task.Stage, task.Error, *task.CompletedAt) if compactErr := s.finalizeTaskTextReplay(task.ID, model.TaskStatusFailed); compactErr != nil { _ = s.log(task.UserID, task.ID, "error", "文本回放草稿归并失败", compactErr.Error()) } @@ -960,11 +991,20 @@ func (s *Service) processClaimedTask(task *model.Task) error { task.Progress = 90 _ = s.repo.UpdateTaskProgress(task.ID, task.Stage, task.Progress) if err := s.saveTaskCompletionWithinStorageQuota(task, resultJSON, opsJSON, len(canvasOps) > 0); err != nil { + if errors.Is(err, repository.ErrTaskStateConflict) { + latest, latestErr := s.repo.Task(task.ID) + if latestErr == nil && latest.Status == model.TaskStatusCancelled { + _ = s.MarkBillingUncertain(task.BillingOrderID, "上游已返回结果，但任务被取消") + _ = s.markSessionFailed(*latest, "会话任务已取消。") + _ = s.log(task.UserID, task.ID, "warn", "任务已取消，丢弃生成结果", "") + return nil + } + } task.Status = model.TaskStatusFailed task.Stage = "任务结果保存失败" task.Error = taskFailureMessage(err) task.CompletedAt = ptr(time.Now()) - _ = s.repo.Save(task) + _, _ = s.repo.UpdateTaskTerminalState(task.ID, model.TaskStatusRunning, task.Status, task.Stage, task.Error, *task.CompletedAt) if compactErr := s.finalizeTaskTextReplay(task.ID, model.TaskStatusFailed); compactErr != nil { _ = s.log(task.UserID, task.ID, "error", "文本回放草稿归并失败", compactErr.Error()) } diff --git a/backend/internal/service/storage_quota.go b/backend/internal/service/storage_quota.go index 19f6a33d7..b10978c23 100644 --- a/backend/internal/service/storage_quota.go +++ b/backend/internal/service/storage_quota.go @@ -141,6 +141,7 @@ func (s *Service) saveTaskCompletionWithinStorageQuota(task *model.Task, resultJ return err } + expectedStatus := task.Status completed := *task completed.Status = model.TaskStatusSucceeded completed.Stage = "任务完成" @@ -148,7 +149,7 @@ func (s *Service) saveTaskCompletionWithinStorageQuota(task *model.Task, resultJ completed.ResultJSON = string(resultJSON) completed.InputJSON = publicInputJSON completed.CompletedAt = ptr(time.Now()) - if err := s.repo.SaveTaskCompletion(&completed, session, message, results); err != nil { + if err := s.repo.SaveTaskCompletion(&completed, expectedStatus, session, message, results); err != nil { return err } *task = completed diff --git a/web/src/pages/tasks/index.tsx b/web/src/pages/tasks/index.tsx index 65e38251e..cab6f0a5c 100644 --- a/web/src/pages/tasks/index.tsx +++ b/web/src/pages/tasks/index.tsx @@ -190,7 +190,11 @@ export default function TasksPage() { setStatusFilter("active"); setPage(1); } - message.success(action === "retry" ? "任务已重新入队" : "任务已取消"); + if (action === "retry") message.success("任务已重新入队"); + else if (next.providerCancelStatus === "requested") message.info("已请求上游取消，正在确认费用状态"); + else if (next.providerCancelStatus === "confirmed") message.success("上游已确认取消，积分已退回"); + else if (next.providerCancelStatus === "uncertain") message.warning("任务已取消，上游费用待核对"); + else message.success("任务已取消，积分已退回"); } catch (error) { message.error(error instanceof Error ? error.message : "操作失败"); } finally { @@ -452,6 +456,8 @@ export default function TasksPage() { + {detailTask.providerCancelStatus ? : null} + {detailTask.providerCancelRequestedAt ? : null} {canQueryProviderTask(detailTask) ?
: null} {detailTask.error ? 
{generationErrorMessage(detailTask.error)}
: null} @@ -573,12 +579,23 @@ function getTaskCanvasContext(task: GenerationTask, canvasById: Map = { @@ -25,6 +26,11 @@ export type GenerationTask = { provider?: string; model?: string; providerRequestId?: string; + providerCancelStatus?: ProviderCancelStatus; + providerCancelError?: string; + providerCancelAttempts?: number; + providerCancelRequestedAt?: string; + providerCancelledAt?: string; errorCode?: string; previewUrl?: string; previewKind?: "image" | "video";