fix(canvas): 任务取消 - 向异步 Provider 传播取消并完成费用对账 (#142)

Fixes #142

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

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

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