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

Fixes #142
This commit is contained in:
ddCat
2026-08-06 10:42:39 +08:00
parent 24e9c1f45a
commit 505062f574
13 changed files with 634 additions and 89 deletions

View File

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

View File

@@ -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。

View File

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

View File

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

View File

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

View File

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

View File

@@ -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,

View File

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

View File

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

View File

@@ -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())
}

View File

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

View File

@@ -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() {
<InfoItem label="模型" value={formatModelName(effectiveConfig, detailTask)} />
<InfoItem label="尝试次数" value={`第 ${detailTask.attempts || 1} 次`} />
<InfoItem label="创建时间" value={formatDate(detailTask.createdAt)} />
{detailTask.providerCancelStatus ? <InfoItem label="上游取消" value={providerCancelStatusLabel(detailTask)} /> : null}
{detailTask.providerCancelRequestedAt ? <InfoItem label="请求取消时间" value={formatDate(detailTask.providerCancelRequestedAt)} /> : null}
</div>
{canQueryProviderTask(detailTask) ? <div className="flex justify-end"><Button icon={<RefreshCw className="size-4" />} loading={actingId === detailTask.id} onClick={() => void queryProviderTask(detailTask)}>手动查询任务</Button></div> : null}
{detailTask.error ? <pre className="max-h-28 overflow-auto whitespace-pre-wrap border-l-2 border-red-500 bg-red-50 px-3 py-2 text-xs text-red-700 dark:bg-red-950/30 dark:text-red-300">{generationErrorMessage(detailTask.error)}</pre> : null}
@@ -573,12 +579,23 @@ function getTaskCanvasContext(task: GenerationTask, canvasById: Map<string, { ti
}
function taskAttentionReason(task: GenerationTask) {
if (task.status === "cancelled") return "任务已取消,可按原输入重新提交";
if (task.status === "cancelled") return providerCancelStatusLabel(task);
if (task.errorCode === CONTENT_MODERATION_ERROR_CODE || isContentModerationError(task.error)) return "内容审核未通过,请修改输入后新建任务";
if (task.error) return generationErrorMessage(task.error);
return task.stage || "生成失败,打开详情查看原因";
}
function providerCancelStatusLabel(task: GenerationTask) {
if (task.providerCancelStatus === "requested") return "已请求上游取消,正在等待确认";
if (task.providerCancelStatus === "confirmed") return "上游已确认取消,积分已退回";
if (task.providerCancelStatus === "uncertain") {
if (task.billing?.status === "settled") return "上游未能取消,费用已结算";
if (task.billing?.status === "refunded") return "上游取消结果未确认,积分已退回";
return task.providerCancelError || "上游无法确认取消,费用待核对";
}
return task.billing?.status === "refunded" ? "任务在调用上游前取消,积分已退回" : "任务已取消,可按原输入重新提交";
}
function taskEmptyState(status: TaskStatusFilter) {
if (status === "all") return { title: "还没有任务", description: "新提交的生成会在这里显示状态和实时进度。" };
if (status === "active") return { title: "没有运行中的任务", description: "新提交的生成会在这里显示排队状态和实时进度。" };

View File

@@ -4,6 +4,7 @@ import { generationErrorMessage } from "@/lib/generation-error";
export type TaskStatus = "queued" | "running" | "succeeded" | "failed" | "cancelled";
export type TaskBillingStatus = "reserved" | "running" | "settled" | "refunded" | "uncertain";
export type ProviderCancelStatus = "requested" | "confirmed" | "uncertain";
export type AgentSessionStatus = "active" | "completed" | "failed";
export type BackendEnvelope<T> = {
@@ -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";