diff --git a/backend/internal/app/admin.go b/backend/internal/app/admin.go index b69f063f7..58690cf44 100644 --- a/backend/internal/app/admin.go +++ b/backend/internal/app/admin.go @@ -750,7 +750,7 @@ func (s *Service) LogAPICall(log model.ApiCallLog) error { stdlog.Printf("provider billing request id update failed: billing_order_id=%s provider_request_id=%s error=%v", log.BillingOrderID, log.ProviderRequestID, err) } } - if log.TaskID != "" { + if log.TaskID != "" && (log.RequestKind == "create" || log.RequestKind == "poll" || log.RequestKind == "cancel") { stage := log.RequestKind var nextPollAt *time.Time if stage == "create" && log.Status == model.ApiCallStatusSucceeded && log.ProviderRequestID != "" { @@ -798,7 +798,9 @@ func (s *Service) LogAPICall(log model.ApiCallLog) error { } func (s *Service) mergeVideoAPICallLog(log model.ApiCallLog) (bool, error) { - if log.Capability != "video" || (log.RequestKind != "poll" && log.RequestKind != "download") { + // Delivery is a separate outcome: a failed download must not rewrite a + // successful generation request as failed. + if log.Capability != "video" || log.RequestKind != "poll" { return false, nil } if log.TaskID == "" && log.ProviderRequestID == "" { diff --git a/backend/internal/app/analytics.go b/backend/internal/app/analytics.go index 68d28a3f5..1b93c091b 100644 --- a/backend/internal/app/analytics.go +++ b/backend/internal/app/analytics.go @@ -246,7 +246,7 @@ func (s *Service) decorateAPICallLogs(logs []model.ApiCallLog) error { billingOrderIDs = append(billingOrderIDs, log.BillingOrderID) } } - if (log.Capability != "image" && log.Capability != "video") || log.TaskID == "" { + if (log.Capability != "image" && log.Capability != "video" && log.Capability != "audio") || log.TaskID == "" { continue } if _, exists := seenTaskIDs[log.TaskID]; exists { @@ -303,6 +303,7 @@ func (s *Service) decorateAPICallLogs(logs []model.ApiCallLog) error { } if task, exists := taskByID[logs[index].TaskID]; exists && task.UserID == logs[index].UserID { logs[index].TaskStatus = task.Status + logs[index].MediaStage = task.MediaStage previewURL, previewKind := taskMediaPreview(task.ResultJSON, task.Type) if canvasResourceID(previewURL) != "" { logs[index].MediaPreviewURL = "/api/admin/api-logs/" + logs[index].ID + "/media" diff --git a/backend/internal/app/provider_protocol.go b/backend/internal/app/provider_protocol.go index 09ecbb952..80cdfc3e9 100644 --- a/backend/internal/app/provider_protocol.go +++ b/backend/internal/app/provider_protocol.go @@ -789,6 +789,9 @@ func finishProtocolResult(ctx context.Context, config providerConfig, mode strin if len(references) == 0 { return nil, errors.New("声明式协议已完成但没有返回媒体地址") } + if output, handled, err := recoverProtocolMedia(ctx, config, mode, references); handled { + return output, err + } items := make([]interface{}, 0, len(references)) for _, reference := range references { var data []byte diff --git a/backend/internal/app/provider_task_recovery.go b/backend/internal/app/provider_task_recovery.go index d1e13452d..455727a8b 100644 --- a/backend/internal/app/provider_task_recovery.go +++ b/backend/internal/app/provider_task_recovery.go @@ -71,6 +71,9 @@ func (s *Service) queryFailedVideoTask(ctx context.Context, task *model.Task, cl if task.Status != model.TaskStatusFailed { return nil, BadAuthRequest("只能人工查询状态为失败的任务") } + if task.MediaRecoveryJSON != "" { + return nil, BadAuthRequest("该任务已生成作品，请在任务中心重试保存，无需重新查询生成") + } if !strings.HasPrefix(task.Type, "canvas_video") && !strings.HasPrefix(task.Type, "video_") { return nil, BadAuthRequest("该任务不是视频生成任务") } diff --git a/backend/internal/app/resource.go b/backend/internal/app/resource.go index 5f5430e2c..eecf6e9e8 100644 --- a/backend/internal/app/resource.go +++ b/backend/internal/app/resource.go @@ -491,6 +491,10 @@ func (s *Service) openResourceRange(userID string, resource *model.Resource, ran } func (s *Service) storeResource(userID string, kind string, fileName string, mimeType string, size int64, width int, height int, durationMs int64, body io.Reader, uploadKey *string, forceLocal bool) (*model.Resource, bool, error) { + return s.storeResourceWithWriter(userID, kind, fileName, mimeType, size, width, height, durationMs, body, uploadKey, forceLocal, s.storeResourceObject) +} + +func (s *Service) storeResourceWithWriter(userID string, kind string, fileName string, mimeType string, size int64, width int, height int, durationMs int64, body io.Reader, uploadKey *string, forceLocal bool, writeObject func(*model.Resource, string, io.Reader) (string, error)) (*model.Resource, bool, error) { if existing, err := s.resourceForUploadKey(userID, uploadKey); err != nil { return nil, false, err } else if existing != nil { @@ -536,7 +540,7 @@ func (s *Service) storeResource(userID string, kind string, fileName string, mim return nil, false, err } var etag string - etag, err = s.storeResourceObject(&resource, fileName, body) + etag, err = writeObject(&resource, fileName, body) resource.UpdatedAt = time.Now() if err != nil { resource.Status = model.ResourceStatusFailed diff --git a/backend/internal/app/storage_quota.go b/backend/internal/app/storage_quota.go index fabcad42b..27da7a201 100644 --- a/backend/internal/app/storage_quota.go +++ b/backend/internal/app/storage_quota.go @@ -150,7 +150,18 @@ 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, expectedStatus, results); err != nil { + var register func(*repository.Repository) error + if task.MediaRecoveryJSON != "" { + completed.MediaStage = "completed" + register = func(repo *repository.Repository) error { + writer := &Service{repo: repo, dataDir: s.dataDir} + if err := writer.RegisterTaskOutputFromTask(completed); err != nil { + return err + } + return writer.registerRecoveredMediaAssets(completed) + } + } + if err := s.repo.SaveTaskCompletionWithRegistration(&completed, expectedStatus, results, register); err != nil { return err } *task = completed diff --git a/backend/internal/app/task_execution.go b/backend/internal/app/task_execution.go index a4cdfa3b0..5f99108a2 100644 --- a/backend/internal/app/task_execution.go +++ b/backend/internal/app/task_execution.go @@ -22,6 +22,7 @@ func (s *Service) processTask(ctx context.Context, task model.Task) (map[string] task.InputJSON = decryptedInput ctx = withTaskExecutionID(ctx, task.ID) ctx = withProviderAnalytics(ctx, s, task) + ctx = context.WithValue(ctx, mediaExecutionTaskKey{}, task) if task.Type == "canvas_text" || task.Type == "canvas_image" || task.Type == "canvas_video" || task.Type == "canvas_audio" { result, err := s.processCanvasGenerationTask(ctx, task.UserID, task.ProjectID, task.Type, task.Prompt, task.InputJSON) diff --git a/backend/internal/app/task_lifecycle.go b/backend/internal/app/task_lifecycle.go index 0c4deacef..eaea04bb2 100644 --- a/backend/internal/app/task_lifecycle.go +++ b/backend/internal/app/task_lifecycle.go @@ -39,6 +39,9 @@ func (w *taskLifecycleCoordinator) retryTask(userID string, id string) (*model.T if err != nil { return nil, err } + if task.MediaRecoveryJSON != "" { + return nil, BadAuthRequest("作品已生成，请使用重试保存，不要重复生成") + } if task.CreationSubmissionID != nil { return nil, creationConflict("智能创作重做需要新的报价批准，请回到创作会话继续") } diff --git a/backend/internal/app/task_media_assets.go b/backend/internal/app/task_media_assets.go new file mode 100644 index 000000000..59462e4ff --- /dev/null +++ b/backend/internal/app/task_media_assets.go @@ -0,0 +1,129 @@ +package app + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "time" + + "gorm.io/gorm" + "infinite-canvas/backend/internal/model" +) + +// A deleted canvas/project must not be resurrected by an old download worker. +func (s *Service) mediaTaskProject(task *model.Task) (string, error) { + id := task.ProjectID + if id == "" { + return "", nil + } + canvas, err := s.repo.CanvasProjectForUser(task.UserID, id) + if err == nil { + id = canvas.ProjectID + if id == "" { + return "", nil + } + } else if !errors.Is(err, gorm.ErrRecordNotFound) { + return "", err + } + project, err := s.repo.ProjectForUser(task.UserID, id) + if errors.Is(err, gorm.ErrRecordNotFound) { + return "", NotFound("任务所属画布或项目已不存在") + } + if err != nil { + return "", err + } + if project.Status == model.ProjectStatusArchived { + return "", BadAuthRequest("项目已归档，无法恢复作品") + } + return project.ID, nil +} + +// Uses the same effect key and ID as browser materialization, so reconnecting +// the canvas cannot create a second asset for a server-delivered output. +func (s *Service) registerRecoveredMediaAssets(task model.Task) error { + projectID, err := s.mediaTaskProject(&task) + if err != nil { + return err + } + checkpoint, err := s.decodeMediaCheckpoint(&task) + if err != nil { + return err + } + representations, err := s.repo.AssetRepresentationsForTask(task.ID) + if err != nil { + return err + } + policy, err := s.RuntimePolicy() + if err != nil { + return err + } + for index, item := range checkpoint.Items { + registered := false + for _, representation := range representations { + if representation.ResourceID == item.ResourceID && representation.AssetVersionID != "" { + registered = true + break + } + } + if registered { + continue + } + resource, err := s.repo.ResourceForUser(task.UserID, item.ResourceID) + if err != nil { + return err + } + if resource.Status != model.ResourceStatusReady { + return errors.New("作品文件尚未保存完成") + } + effectKey := fmt.Sprintf("materialize:%s:%d", task.ID, index) + sum := sha256.Sum256([]byte(effectKey)) + assetID := "generation_" + hex.EncodeToString(sum[:]) + // Preserve existing user edits on replay; the deterministic key is unique. + if _, err := s.repo.AssetForUser(task.UserID, assetID); err == nil { + continue + } else if !errors.Is(err, gorm.ErrRecordNotFound) { + return err + } + now := time.Now() + url := resourceFileURL(resource.ID) + data := map[string]any{"storageKey": "resource:" + resource.ID, "mimeType": resource.MimeType, "bytes": resource.Size, "width": resource.Width, "height": resource.Height, "durationMs": resource.DurationMs} + if checkpoint.Mode == "image" { + data["dataUrl"] = url + } else { + data["url"] = url + } + title := "生成作品" + metadata := map[string]any{"source": "generation-task", "generationEffectKey": effectKey, "taskId": task.ID, "outputIndex": index} + if projectID != "" { + metadata["projectIds"] = []string{projectID} + } + payload, err := json.Marshal(map[string]any{"id": assetID, "kind": checkpoint.Mode, "category": model.AssetCategoryMaterial, "status": model.AssetVersionStatusConfirmed, "title": title, "coverUrl": url, "tags": []string{"生成"}, "createdAt": now.UTC().Format(time.RFC3339Nano), "updatedAt": now.UTC().Format(time.RFC3339Nano), "data": data, "metadata": metadata}) + if err != nil { + return err + } + usage, err := s.repo.UserStorageUsage(task.UserID) + if err != nil { + return err + } + if err := validateStructuredStorageQuotaWithPolicy(usage, "asset", true, int64(len(payload)), policy.Resource); err != nil { + return err + } + asset := &model.Asset{ID: assetID, UserID: task.UserID, Kind: checkpoint.Mode, Category: model.AssetCategoryMaterial, Status: model.AssetVersionStatusConfirmed, Title: title, PayloadJSON: string(payload), CreatedAt: now, UpdatedAt: now} + if err := s.repo.UpsertAsset(asset); err != nil { + return err + } + if projectID != "" { + position, err := s.repo.NextProjectAssetPosition(projectID, "") + if err != nil { + return err + } + _, err = s.repo.LinkProjectAsset(asset, nil, &model.ProjectAssetLink{ID: newID(), ProjectID: projectID, AssetID: assetID, Position: position, CreatedAt: now}) + if err != nil { + return err + } + } + } + return nil +} diff --git a/backend/internal/app/task_media_recovery.go b/backend/internal/app/task_media_recovery.go new file mode 100644 index 000000000..b81540e0b --- /dev/null +++ b/backend/internal/app/task_media_recovery.go @@ -0,0 +1,428 @@ +package app + +import ( + "context" + "encoding/json" + "errors" + "math/rand/v2" + "os" + "path/filepath" + "strings" + "time" + + "gorm.io/gorm" + + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/protocol" + "infinite-canvas/backend/internal/repository" +) + +// Recovery is keyed by the task, not a second generation or billing order. +type mediaCheckpoint struct { + Mode string `json:"mode"` + Items []mediaCheckpointItem `json:"items"` + StartedAt time.Time `json:"startedAt"` + Attempts int `json:"attempts"` + ManualAttempts int `json:"manualAttempts"` + LastManualAt time.Time `json:"lastManualAt"` +} + +type mediaCheckpointItem struct { + Reference protocol.MediaReference `json:"reference"` + TempName string `json:"tempName,omitempty"` + MIMEType string `json:"mimeType,omitempty"` + ResourceID string `json:"resourceId,omitempty"` +} + +type mediaRecoveryError struct { + stage string + retryable bool + cause error +} +type mediaExecutionTaskKey struct{} + +func (e *mediaRecoveryError) Error() string { + return "作品已生成，但保存未完成：" + mediaStageLabel(e.stage) +} +func (e *mediaRecoveryError) Unwrap() error { return e.cause } + +var mediaRecoverySlots = make(chan struct{}, 2) +var mediaRecoveryDelays = []time.Duration{15 * time.Second, time.Minute, 3 * time.Minute, 10 * time.Minute} + +func mediaStageLabel(stage string) string { + switch stage { + case "download": + return "下载结果失败" + case "upload": + return "上传 OSS 失败" + case "local_save": + return "保存本地文件失败" + case "register": + return "登记作品失败" + default: + return "记录恢复信息失败" + } +} + +func (s *Service) decodeMediaCheckpoint(task *model.Task) (*mediaCheckpoint, error) { + plain, err := s.decryptSettingSecret(task.MediaRecoveryJSON) + if err != nil { + return nil, err + } + var checkpoint mediaCheckpoint + if err := json.Unmarshal([]byte(plain), &checkpoint); err != nil { + return nil, err + } + if len(checkpoint.Items) == 0 || len(checkpoint.Items) > 32 || (checkpoint.Mode != "image" && checkpoint.Mode != "video" && checkpoint.Mode != "audio") { + return nil, errors.New("作品恢复信息无效") + } + return &checkpoint, nil +} + +func (s *Service) saveMediaCheckpoint(task *model.Task, checkpoint *mediaCheckpoint, stage string) error { + data, err := json.Marshal(checkpoint) + if err != nil { + return err + } + encoded, err := s.encryptSettingSecret(string(data)) + if err != nil { + return err + } + return s.repo.SaveTaskMediaCheckpoint(task, encoded, stage) +} + +func recoverProtocolMedia(ctx context.Context, config providerConfig, mode string, references []protocol.MediaReference) (map[string]interface{}, bool, error) { + metadata, ok := ctx.Value(providerAnalyticsKey{}).(providerAnalyticsContext) + if !ok || metadata.Service == nil || metadata.TaskID == "" { + return nil, false, nil + } + s := metadata.Service + executing, ok := ctx.Value(mediaExecutionTaskKey{}).(model.Task) + if !ok { + return nil, false, nil + } + task := &executing + var err error + // Manual video queries retain their existing recovery lease and completion path. + if task.Status != model.TaskStatusRunning { + return nil, false, nil + } + if len(references) == 0 || len(references) > 32 { + return nil, true, &mediaRecoveryError{stage: "checkpoint", cause: errors.New("生成结果数量超过恢复限制")} + } + if mode != "image" { + references = references[:1] + } + if task.MediaRecoveryJSON != "" { + result, err := s.resumeTaskMedia(ctx, task) + return result, true, err + } + checkpoint := &mediaCheckpoint{Mode: mode, StartedAt: time.Now()} + for _, reference := range references { + if len(reference.URL) > 16_384 { + return nil, true, &mediaRecoveryError{stage: "checkpoint", cause: errors.New("上游媒体地址过长")} + } + item := mediaCheckpointItem{Reference: protocol.MediaReference{URL: reference.URL, MIMEType: reference.MIMEType}} + // Inline bytes are staged, never stored as a large base64 DB checkpoint. + if reference.DataURL != "" { + mimeType, data, decodeErr := decodeProviderDataURL(reference.DataURL) + if decodeErr != nil { + return nil, true, &mediaRecoveryError{stage: "checkpoint", cause: decodeErr} + } + policy, policyErr := s.RuntimePolicy() + if policyErr != nil { + return nil, true, &mediaRecoveryError{stage: "checkpoint", cause: policyErr} + } + if int64(len(data)) > min(megabytes(policy.Resource.GeneratedFileMB), mediaStagingBudget) || !strings.HasPrefix(mimeType, mode+"/") { + return nil, true, &mediaRecoveryError{stage: "checkpoint", cause: errors.New("生成文件类型或大小不符合限制")} + } + item.TempName, err = s.stageInlineMedia(data) + if err != nil { + return nil, true, &mediaRecoveryError{stage: "checkpoint", cause: err} + } + item.MIMEType = mimeType + item.Reference.DataURL = "" + } + checkpoint.Items = append(checkpoint.Items, item) + } + if err := s.saveMediaCheckpoint(task, checkpoint, "download"); err != nil { + return nil, true, &mediaRecoveryError{stage: "checkpoint", cause: err} + } + result, err := s.materializeTaskMedia(ctx, task, config, checkpoint) + return result, true, err +} + +func (s *Service) resumeTaskMedia(ctx context.Context, task *model.Task) (map[string]interface{}, error) { + checkpoint, err := s.decodeMediaCheckpoint(task) + if err != nil { + return nil, &mediaRecoveryError{stage: "checkpoint", cause: err} + } + raw, err := s.decryptTaskInputJSON(task.InputJSON) + if err != nil { + return nil, &mediaRecoveryError{stage: "checkpoint", cause: err} + } + var input canvasGenerationInput + if err := json.Unmarshal([]byte(raw), &input); err != nil { + return nil, &mediaRecoveryError{stage: "checkpoint", cause: err} + } + return s.materializeTaskMedia(withProviderAnalytics(ctx, s, *task), task, input.Config, checkpoint) +} + +func (s *Service) materializeTaskMedia(ctx context.Context, task *model.Task, config providerConfig, checkpoint *mediaCheckpoint) (result map[string]interface{}, err error) { + defer func() { + var failure *mediaRecoveryError + if err != nil && !errors.As(err, &failure) { + err = &mediaRecoveryError{stage: task.MediaStage, cause: err} + } + }() + // Do not occupy a worker waiting for a download slot. + select { + case mediaRecoverySlots <- struct{}{}: + defer func() { <-mediaRecoverySlots }() + default: + return nil, &mediaRecoveryError{stage: task.MediaStage, retryable: true, cause: errMediaRecoveryBusy} + } + if _, err := s.mediaTaskProject(task); err != nil { + return nil, &mediaRecoveryError{stage: "register", cause: err} + } + items := make([]interface{}, 0, len(checkpoint.Items)) + for index := range checkpoint.Items { + item := &checkpoint.Items[index] + if err := ctx.Err(); err != nil { + return nil, &mediaRecoveryError{stage: task.MediaStage, retryable: true, cause: err} + } + resource, err := s.resourceForUploadKey(task.UserID, mediaUploadKey(task.ID, index)) + if err != nil { + return nil, &mediaRecoveryError{stage: "register", retryable: true, cause: err} + } + if resource == nil || resource.Status != model.ResourceStatusReady { + path := s.mediaTempPath(item.TempName) + info, statErr := os.Lstat(path) + if path == "" || statErr != nil || !info.Mode().IsRegular() { + if item.Reference.URL == "" { + return nil, &mediaRecoveryError{stage: "download", cause: errors.New("暂存结果已过期且上游没有可重新下载的地址")} + } + if err := s.saveMediaCheckpoint(task, checkpoint, "download"); err != nil { + return nil, err + } + config, err = s.resolveProviderConfig(config) + if err != nil { + return nil, &mediaRecoveryError{stage: "download", cause: err} + } + item.TempName, item.MIMEType, err = s.downloadTaskMedia(ctx, config, item.Reference.URL, checkpoint.Mode) + if err != nil { + return nil, &mediaRecoveryError{stage: "download", retryable: retryableMediaRecovery(err), cause: err} + } + path = s.mediaTempPath(item.TempName) + if err := s.saveMediaCheckpoint(task, checkpoint, "download"); err != nil { + return nil, err + } + } + stage := "local_save" + _, _, oss, settingErr := s.activeResourceOSSSetting(task.UserID) + if settingErr != nil { + return nil, &mediaRecoveryError{stage: "upload", cause: settingErr} + } + if oss || (resource != nil && resource.Provider != "local") { + stage = "upload" + } + if err := s.saveMediaCheckpoint(task, checkpoint, stage); err != nil { + return nil, err + } + started := time.Now() + resource, err = s.storeTaskMediaFile(task, index, path, item.MIMEType, checkpoint.Mode, resource) + s.logMediaStage(*task, stage, started, err) + if err != nil { + return nil, &mediaRecoveryError{stage: stage, retryable: retryableMediaRecovery(err), cause: err} + } + } + item.ResourceID = resource.ID + if err := s.saveMediaCheckpoint(task, checkpoint, "register"); err != nil { + return nil, err + } + url := resourceFileURL(resource.ID) + items = append(items, map[string]interface{}{"dataUrl": url, "url": url, "resourceId": resource.ID, "storageKey": "resource:" + resource.ID, "mimeType": resource.MimeType, "bytes": resource.Size, "width": resource.Width, "height": resource.Height, "durationMs": resource.DurationMs}) + } + result = map[string]interface{}{"mode": checkpoint.Mode} + if checkpoint.Mode == "image" { + result["images"] = items + } else { + result[checkpoint.Mode] = items[0] + } + return result, nil +} + +var errMediaRecoveryBusy = errors.New("作品保存并发已满") + +func retryableMediaRecovery(err error) bool { + if retryableProtocolMediaDownload(err) || errors.Is(err, context.DeadlineExceeded) { + return true + } + var upstream providerHTTPError + return errors.As(err, &upstream) && (upstream.StatusCode == 429 || upstream.StatusCode == 408 || upstream.StatusCode >= 500) +} + +// Called before the generic execution failure path, so result delivery cannot be +// mistaken for model failure or trigger a second generation. +func (s *Service) handleMediaRecoveryFailure(task *model.Task, cause error) error { + latest, err := s.repo.Task(task.ID) + if err != nil { + return err + } + if latest.Status != model.TaskStatusRunning || latest.LeaseOwner != task.LeaseOwner { + return repository.ErrTaskStateConflict + } + task.MediaRecoveryJSON, task.MediaStage = latest.MediaRecoveryJSON, latest.MediaStage + var failure *mediaRecoveryError + if !errors.As(cause, &failure) { + failure = &mediaRecoveryError{stage: task.MediaStage, retryable: true, cause: cause} + } + if task.MediaRecoveryJSON != "" { + checkpoint, decodeErr := s.decodeMediaCheckpoint(task) + if decodeErr != nil { + return decodeErr + } + if errors.Is(cause, errMediaRecoveryBusy) && time.Since(checkpoint.StartedAt) < 20*time.Minute { + return s.repo.DeferRunningTaskForProviderPoll(task.ID, task.LeaseOwner, "作品已生成，等待保存", 5*time.Second) + } + if failure.retryable && checkpoint.Attempts < len(mediaRecoveryDelays) && time.Since(checkpoint.StartedAt) < 20*time.Minute { + delay := mediaRecoveryDelays[checkpoint.Attempts] + time.Duration(rand.IntN(5_000))*time.Millisecond + var upstream providerHTTPError + if errors.As(cause, &upstream) { + delay = max(delay, min(upstream.RetryAfter, 10*time.Minute)) + } + checkpoint.Attempts++ + if err := s.saveMediaCheckpoint(task, checkpoint, failure.stage); err != nil { + return err + } + return s.repo.DeferRunningTaskForProviderPoll(task.ID, task.LeaseOwner, "作品已生成，保存遇到问题，正在自动恢复", delay) + } + if err := s.saveMediaCheckpoint(task, checkpoint, failure.stage); err != nil { + return err + } + } + // Billing belongs to the original generation. A delivery failure cannot refund + // or submit it again implicitly; the existing reconciliation path is retained. + return s.terminalCoordinator().handleResultPersistenceFailureError(task, failure) +} + +func (s *Service) RecoverTaskMedia(userID, id string) (*model.Task, error) { + if s.IsDraining() { + return nil, BadAuthRequest("服务正在维护，请稍后重试保存") + } + task, err := s.repo.TaskForUser(userID, strings.TrimSpace(id)) + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, NotFound("任务不存在") + } + if err != nil { + return nil, err + } + if task.MediaRecoveryJSON == "" { + return nil, BadAuthRequest("该任务没有可恢复的作品，请先核对上游结果") + } + if task.Status == model.TaskStatusRunning || task.Status == model.TaskStatusQueued || task.Status == model.TaskStatusSucceeded { + return taskForOutput(*task), nil + } + if task.Status != model.TaskStatusFailed { + return nil, BadAuthRequest("仅保存失败的作品可以恢复") + } + if _, err := s.mediaTaskProject(task); err != nil { + return nil, err + } + if err := s.taskBilling().CheckRetryEligibility(task.BillingOrderID); err != nil && !errors.Is(err, errTaskBillingReview) { + return nil, err + } + // Never silently restore a refunded order (the old video recovery does that). + if task.BillingOrderID != "" { + order, err := s.repo.BillingOrder(task.BillingOrderID) + if err != nil { + return nil, err + } + if order.UserID != task.UserID || order.TaskID != task.ID { + return nil, BadAuthRequest("任务与计费订单归属不一致") + } + if order.Status == model.BillingStatusRefunded { + return nil, BadAuthRequest("原任务已退款，请联系管理员处理作品，不会自动重新扣费") + } + } + checkpoint, err := s.decodeMediaCheckpoint(task) + if err != nil { + return nil, err + } + if checkpoint.ManualAttempts >= 3 { + return nil, BadAuthRequest("已达到重试保存上限，请联系管理员核对渠道结果") + } + if time.Since(checkpoint.LastManualAt) < time.Minute { + return nil, BadAuthRequest("请稍后再试，重试保存间隔至少一分钟") + } + checkpoint.ManualAttempts++ + checkpoint.LastManualAt = time.Now() + // Manual recovery is one attempt, not another full automatic retry budget. + checkpoint.Attempts = len(mediaRecoveryDelays) + data, err := json.Marshal(checkpoint) + if err != nil { + return nil, err + } + encrypted, err := s.encryptSettingSecret(string(data)) + if err != nil { + return nil, err + } + if err := s.repo.RequeueTaskMediaRecovery(task, encrypted); err != nil { + if errors.Is(err, repository.ErrTaskStateConflict) { + if latest, readErr := s.repo.TaskForUser(userID, task.ID); readErr == nil && (latest.Status == model.TaskStatusQueued || latest.Status == model.TaskStatusRunning || latest.Status == model.TaskStatusSucceeded) { + return taskForOutput(*latest), nil + } + } + return nil, err + } + next, err := s.repo.TaskForUser(userID, task.ID) + if err != nil { + return nil, err + } + return taskForOutput(*next), nil +} + +func (s *Service) logMediaStage(task model.Task, stage string, started time.Time, err error) { + status := model.ApiCallStatusSucceeded + code, message := "", "" + if err != nil { + status = model.ApiCallStatusFailed + code = "media_" + stage + "_failed" + message = mediaStageLabel(stage) + } + if logErr := s.LogAPICall(model.ApiCallLog{UserID: task.UserID, TaskID: task.ID, BillingOrderID: task.BillingOrderID, ProviderRequestID: task.ProviderRequestID, Source: "backend-task", Capability: capabilityFromTaskType(task.Type), Model: task.Model, RequestKind: stage, Method: "INTERNAL", Path: "task/media/" + stage, Status: status, ErrorCode: code, Error: message, DurationMs: time.Since(started).Milliseconds()}); logErr != nil { + _ = s.log(task.UserID, task.ID, "error", "记录作品保存阶段失败", logErr.Error()) + } +} + +func (s *Service) mediaTempPath(name string) string { + if !strings.HasPrefix(name, "result-") || filepath.Base(name) != name || strings.ContainsAny(name, `/\\`) { + return "" + } + return filepath.Join(s.dataDir, "media-staging", name) +} + +func (c *taskTerminalCoordinator) handleResultPersistenceFailureError(task *model.Task, err error) error { + _, failure := c.handleResultPersistenceFailure(task, err) + return failure +} + +func (s *Service) finishTaskMediaRecovery(task *model.Task, result map[string]interface{}, err error) error { + if err != nil { + return s.handleMediaRecoveryFailure(task, err) + } + started := time.Now() + if _, err = s.finalizeCharacterTurnaroundTask(*task, result); err == nil { + var encoded []byte + encoded, err = json.Marshal(result) + if err == nil { + err = s.saveTaskCompletionWithinStorageQuota(task, encoded, nil, false) + } + } + s.logMediaStage(*task, "register", started, err) + if err != nil { + return s.handleMediaRecoveryFailure(task, &mediaRecoveryError{stage: "register", cause: err}) + } + s.cleanupMediaCheckpointFiles(task) + return s.terminalCoordinator().handleSuccess(task) +} diff --git a/backend/internal/app/task_media_recovery_test.go b/backend/internal/app/task_media_recovery_test.go new file mode 100644 index 000000000..d3ba3e95d --- /dev/null +++ b/backend/internal/app/task_media_recovery_test.go @@ -0,0 +1,411 @@ +package app + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "image" + "image/png" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "gorm.io/gorm" + "infinite-canvas/backend/internal/database" + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/protocol" + "infinite-canvas/backend/internal/repository" +) + +func newMediaRecoveryTestService(t *testing.T) (*Service, *gorm.DB) { + t.Helper() + db := newSQLiteTestDB(t) + if err := database.MigrateSchema(db); err != nil { + t.Fatal(err) + } + return &Service{repo: repository.New(db), dataDir: t.TempDir(), activeCancels: make(map[string]context.CancelFunc)}, db +} + +func mediaTestPNG(t *testing.T) []byte { + t.Helper() + var buffer bytes.Buffer + if err := png.Encode(&buffer, image.NewRGBA(image.Rect(0, 0, 2, 3))); err != nil { + t.Fatal(err) + } + return buffer.Bytes() +} + +func seedMediaTask(t *testing.T, db *gorm.DB, config providerConfig) *model.Task { + t.Helper() + input, err := json.Marshal(canvasGenerationInput{Mode: "image", Config: config, Prompt: "test"}) + if err != nil { + t.Fatal(err) + } + expires := time.Now().Add(time.Hour) + task := &model.Task{ID: newID(), UserID: "media-user", Type: "canvas_image", Status: model.TaskStatusRunning, InputJSON: string(input), LeaseOwner: "worker-a", LeaseExpiresAt: &expires, RouteRun: 1, CreatedAt: time.Now()} + if err := db.Create(task).Error; err != nil { + t.Fatal(err) + } + return task +} + +func TestMediaRecoveryDownloadFailureResumesSameTaskAfterRestart(t *testing.T) { + t.Setenv("CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS", "127.0.0.1") + s, db := newMediaRecoveryTestService(t) + picture := mediaTestPNG(t) + var downloads atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + t.Errorf("recovery submitted generation: %s", r.Method) + } + if downloads.Add(1) == 1 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "image/png") + _, _ = w.Write(picture) + })) + defer server.Close() + config := providerConfig{BaseURL: server.URL} + task := seedMediaTask(t, db, config) + ctx := context.WithValue(withProviderAnalytics(context.Background(), s, *task), mediaExecutionTaskKey{}, *task) + _, handled, failure := recoverProtocolMedia(ctx, config, "image", []protocol.MediaReference{{URL: server.URL + "/result?token=private-test-token"}}) + if !handled || failure == nil { + t.Fatalf("handled=%v failure=%v", handled, failure) + } + if err := s.handleMediaRecoveryFailure(task, failure); err != nil { + t.Fatal(err) + } + saved, err := s.repo.Task(task.ID) + if err != nil { + t.Fatal(err) + } + if saved.Status != model.TaskStatusRunning || saved.NextPollAt == nil || saved.LeaseOwner != "" { + t.Fatalf("not deferred: %+v", saved) + } + if saved.MediaRecoveryJSON == "" || strings.Contains(saved.MediaRecoveryJSON, "private-test-token") { + t.Fatal("missing/encrypted checkpoint") + } + public, _ := json.Marshal(taskForOutput(*saved)) + if bytes.Contains(public, []byte("mediaRecovery")) || bytes.Contains(public, []byte("private-test-token")) { + t.Fatal("checkpoint leaked") + } + if err := db.Model(&model.Task{}).Where("id = ?", task.ID).Update("next_poll_at", time.Now().Add(-time.Second)).Error; err != nil { + t.Fatal(err) + } + // A fresh service has no in-memory state or downloaded bytes from generation. + restarted := &Service{repo: s.repo, dataDir: s.dataDir, activeCancels: make(map[string]context.CancelFunc)} + claimed, err := restarted.repo.ClaimNextTask("worker-b", time.Minute) + if err != nil || claimed == nil { + t.Fatalf("claim: %v", err) + } + if err := restarted.taskWorker().processClaimedTask(claimed, nil); err != nil { + t.Fatal(err) + } + completed, _ := s.repo.Task(task.ID) + if completed.Status != model.TaskStatusSucceeded || completed.MediaStage != "completed" { + t.Fatalf("not delivered: %s %s", completed.Status, completed.MediaStage) + } + if downloads.Load() != 2 { + t.Fatalf("downloads=%d", downloads.Load()) + } + var assets []model.Asset + if err := db.Find(&assets, "user_id = ?", task.UserID).Error; err != nil { + t.Fatal(err) + } + if len(assets) != 1 || !strings.HasPrefix(assets[0].ID, "generation_") { + t.Fatalf("assets: %+v", assets) + } + if err := restarted.registerRecoveredMediaAssets(*completed); err != nil { + t.Fatal(err) + } + var count int64 + db.Model(&model.Asset{}).Where("user_id = ?", task.UserID).Count(&count) + if count != 1 { + t.Fatalf("duplicate assets: %d", count) + } + db.Model(&model.BillingOrder{}).Count(&count) + if count != 0 { + t.Fatal("recovery created a billing order") + } + files, err := os.ReadDir(filepath.Join(s.dataDir, "media-staging")) + if err != nil || len(files) != 0 { + t.Fatalf("staging not cleaned: %v %v", files, err) + } +} + +func TestMediaRecoveryUploadFailureReusesStagedFileAndResource(t *testing.T) { + t.Setenv("CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS", "127.0.0.1") + s, db := newMediaRecoveryTestService(t) + seedOSSEnabled(t, db, "media-user", "http://127.0.0.1:1") + task := seedMediaTask(t, db, providerConfig{}) + temp, err := s.stageInlineMedia(mediaTestPNG(t)) + if err != nil { + t.Fatal(err) + } + checkpoint := &mediaCheckpoint{Mode: "image", StartedAt: time.Now(), Items: []mediaCheckpointItem{{TempName: temp, MIMEType: "image/png"}}} + if err := s.saveMediaCheckpoint(task, checkpoint, "upload"); err != nil { + t.Fatal(err) + } + _, failure := s.materializeTaskMedia(context.Background(), task, providerConfig{}, checkpoint) + var delivery *mediaRecoveryError + if !errors.As(failure, &delivery) || delivery.stage != "upload" { + t.Fatalf("expected OSS failure, got %v", failure) + } + resource, err := s.resourceForUploadKey(task.UserID, mediaUploadKey(task.ID, 0)) + if err != nil || resource == nil || resource.Status == model.ResourceStatusReady || resource.Provider == "local" { + t.Fatalf("silently degraded upload: %+v %v", resource, err) + } + if _, err := os.Stat(s.mediaTempPath(temp)); err != nil { + t.Fatal("upload failure discarded download", err) + } + // Simulate fixing storage configuration on the same existing object identity. + resource.Provider, resource.ObjectKey = "local", "recovered/picture.png" + if err := s.repo.SaveResource(resource); err != nil { + t.Fatal(err) + } + output, err := s.resumeTaskMedia(context.Background(), task) + if err != nil { + t.Fatal(err) + } + if err := s.finishTaskMediaRecovery(task, output, nil); err != nil { + t.Fatal(err) + } + ready, _ := s.resourceForUploadKey(task.UserID, mediaUploadKey(task.ID, 0)) + if ready.ID != resource.ID || ready.Status != model.ResourceStatusReady { + t.Fatal("resource identity changed") + } + var count int64 + db.Model(&model.Resource{}).Where("user_id = ?", task.UserID).Count(&count) + if count != 1 { + t.Fatalf("duplicate resource count=%d", count) + } +} + +func TestMediaRecoveryRegistrationRollbackAndRetryWithoutDownload(t *testing.T) { + s, db := newMediaRecoveryTestService(t) + task := seedMediaTask(t, db, providerConfig{}) + temp, err := s.stageInlineMedia(mediaTestPNG(t)) + if err != nil { + t.Fatal(err) + } + checkpoint := &mediaCheckpoint{Mode: "image", StartedAt: time.Now(), Items: []mediaCheckpointItem{{TempName: temp, MIMEType: "image/png"}}} + if err := s.saveMediaCheckpoint(task, checkpoint, "local_save"); err != nil { + t.Fatal(err) + } + output, err := s.materializeTaskMedia(context.Background(), task, providerConfig{}, checkpoint) + if err != nil { + t.Fatal(err) + } + encoded, _ := json.Marshal(output) + if err := db.Exec(`CREATE TRIGGER reject_media_asset BEFORE INSERT ON assets BEGIN SELECT RAISE(ABORT, 'injected asset failure'); END`).Error; err != nil { + t.Fatal(err) + } + if err := s.saveTaskCompletionWithinStorageQuota(task, encoded, nil, false); err == nil { + t.Fatal("registration failure was ignored") + } + saved, _ := s.repo.Task(task.ID) + if saved.Status != model.TaskStatusRunning || saved.ResultJSON != "" { + t.Fatal("false succeeded task escaped transaction") + } + if err := db.Exec(`DROP TRIGGER reject_media_asset`).Error; err != nil { + t.Fatal(err) + } + if err := os.Remove(s.mediaTempPath(temp)); err != nil { + t.Fatal(err) + } + output, err = s.resumeTaskMedia(context.Background(), saved) + if err != nil { + t.Fatal("ready resource must not need staged file or URL", err) + } + if err := s.finishTaskMediaRecovery(saved, output, nil); err != nil { + t.Fatal(err) + } +} + +func TestMediaRecoveryFencesLeasesCancellationAndManualRetries(t *testing.T) { + s, db := newMediaRecoveryTestService(t) + task := seedMediaTask(t, db, providerConfig{}) + checkpoint := &mediaCheckpoint{Mode: "image", StartedAt: time.Now(), Items: []mediaCheckpointItem{{Reference: protocol.MediaReference{URL: "https://example.test/result"}}}} + if err := s.saveMediaCheckpoint(task, checkpoint, "download"); err != nil { + t.Fatal(err) + } + stale := *task + stale.LeaseOwner = "obsolete-worker" + if err := s.saveMediaCheckpoint(&stale, checkpoint, "upload"); !errors.Is(err, repository.ErrTaskStateConflict) { + t.Fatalf("stale write: %v", err) + } + if err := db.Model(task).Updates(map[string]any{"status": model.TaskStatusFailed, "lease_owner": "", "lease_expires_at": nil}).Error; err != nil { + t.Fatal(err) + } + if _, err := s.RecoverTaskMedia("other-user", task.ID); err == nil { + t.Fatal("cross-user recovery") + } + queued, err := s.RecoverTaskMedia(task.UserID, task.ID) + if err != nil || queued.ID != task.ID { + t.Fatalf("manual recovery: %v", err) + } + duplicate, err := s.RecoverTaskMedia(task.UserID, task.ID) + if err != nil || duplicate.ID != task.ID { + t.Fatalf("duplicate click: %v", err) + } + if err := db.Model(task).Update("status", model.TaskStatusCancelled).Error; err != nil { + t.Fatal(err) + } + if _, err := s.RecoverTaskMedia(task.UserID, task.ID); err == nil { + t.Fatal("cancelled task revived") + } + if err := s.saveMediaCheckpoint(task, checkpoint, "register"); !errors.Is(err, repository.ErrTaskStateConflict) { + t.Fatalf("cancelled write: %v", err) + } +} + +func TestMediaRecoveryDoesNotFailOverGenerationRoute(t *testing.T) { + p := &taskRouteExecutionPortStub{nextAttempts: map[string]*model.RouteAttempt{"first": {RouteID: "second"}}, processResults: map[string]taskRouteExecutionStubResult{"first": {err: &mediaRecoveryError{stage: "download", retryable: true, cause: errors.New("connection reset")}}}} + result, err := (&taskRouteExecutor{port: p}).execute(context.Background(), &model.Task{RouteID: "first"}, &model.RouteAttempt{RouteID: "first"}) + if err != nil || !result.providerSucceeded || len(p.processCalls) != 1 || p.finished[0] != "first:succeeded" { + t.Fatalf("delivery entered model failover: %+v %v", result, err) + } +} + +func TestMediaRecoveryTerminalStagesLimitsAndPublicList(t *testing.T) { + s, db := newMediaRecoveryTestService(t) + for _, tc := range []struct { + name string + attempts int + age time.Duration + failure *mediaRecoveryError + }{ + {"permanent", 0, 0, &mediaRecoveryError{stage: "upload", cause: providerHTTPError{StatusCode: 403}}}, + {"budget", 4, 0, &mediaRecoveryError{stage: "download", retryable: true, cause: errors.New("reset")}}, + {"expired", 0, time.Hour, &mediaRecoveryError{stage: "register", retryable: true, cause: errors.New("database unavailable")}}, + } { + t.Run(tc.name, func(t *testing.T) { + task := seedMediaTask(t, db, providerConfig{}) + checkpoint := &mediaCheckpoint{Mode: "image", StartedAt: time.Now().Add(-tc.age), Attempts: tc.attempts, Items: []mediaCheckpointItem{{Reference: protocol.MediaReference{URL: "https://example.test/private"}}}} + if err := s.saveMediaCheckpoint(task, checkpoint, "download"); err != nil { + t.Fatal(err) + } + if err := s.handleMediaRecoveryFailure(task, tc.failure); err == nil { + t.Fatal("terminal failure swallowed") + } + saved, err := s.repo.Task(task.ID) + if err != nil || saved.Status != model.TaskStatusFailed || saved.MediaStage != tc.failure.stage { + t.Fatalf("terminal stage: %+v %v", saved, err) + } + if !strings.Contains(saved.Error, "作品已生成") { + t.Fatalf("wrong user message: %s", saved.Error) + } + listed, err := s.repo.Tasks(task.UserID, 100, "", false) + if err != nil { + t.Fatal(err) + } + found := false + for _, item := range taskSummariesForOutput(listed) { + if item.ID == task.ID { + found = item.CanRecoverMedia && item.MediaStage == tc.failure.stage + } + } + if !found { + t.Fatal("list lost recovery capability") + } + }) + } +} + +func TestMediaRecoveryManualGuardsDoNotChangeBilling(t *testing.T) { + for _, name := range []string{"refunded", "wrong-owner", "limit", "cooldown", "deleted-project"} { + t.Run(name, func(t *testing.T) { + s, db := newMediaRecoveryTestService(t) + task := seedMediaTask(t, db, providerConfig{}) + checkpoint := &mediaCheckpoint{Mode: "image", StartedAt: time.Now(), Items: []mediaCheckpointItem{{Reference: protocol.MediaReference{URL: "https://example.test/result"}}}} + order := model.BillingOrder{ID: newID(), TaskID: task.ID, UserID: task.UserID, IdempotencyKey: newID(), Status: model.BillingStatusUncertain, ReservedAmountMicrocredits: 100} + switch name { + case "refunded": + order.Status = model.BillingStatusRefunded + case "wrong-owner": + order.UserID = "another-user" + case "limit": + checkpoint.ManualAttempts = 3 + case "cooldown": + checkpoint.LastManualAt = time.Now() + case "deleted-project": + task.ProjectID = "deleted-project" + } + if err := db.Create(&order).Error; err != nil { + t.Fatal(err) + } + if err := s.saveMediaCheckpoint(task, checkpoint, "upload"); err != nil { + t.Fatal(err) + } + if err := db.Model(task).Updates(map[string]any{"status": model.TaskStatusFailed, "lease_owner": "", "lease_expires_at": nil, "billing_order_id": order.ID, "project_id": task.ProjectID}).Error; err != nil { + t.Fatal(err) + } + if _, err := s.RecoverTaskMedia(task.UserID, task.ID); err == nil { + t.Fatal("guard bypassed") + } + saved, _ := s.repo.Task(task.ID) + if saved.Status != model.TaskStatusFailed { + t.Fatal("guard changed task") + } + unchanged, err := s.repo.BillingOrder(order.ID) + if err != nil || unchanged.Status != order.Status || unchanged.ReservedAmountMicrocredits != 100 { + t.Fatalf("billing changed: %+v %v", unchanged, err) + } + }) + } +} + +func TestMediaRecoveryVideoDownloadDoesNotRewriteGenerationLog(t *testing.T) { + s, db := newMediaRecoveryTestService(t) + task := seedMediaTask(t, db, providerConfig{}) + for _, item := range []model.ApiCallLog{ + {ID: "media-create", UserID: task.UserID, TaskID: task.ID, Capability: "video", RequestKind: "create", Method: "POST", Billable: true, Status: model.ApiCallStatusSucceeded, StatusCode: 200}, + {ID: "media-download", UserID: task.UserID, TaskID: task.ID, Capability: "video", RequestKind: "download", Method: "GET", Status: model.ApiCallStatusFailed, StatusCode: 503}, + } { + if err := s.LogAPICall(item); err != nil { + t.Fatal(err) + } + } + var logs []model.ApiCallLog + if err := db.Where("task_id = ?", task.ID).Order("id").Find(&logs).Error; err != nil { + t.Fatal(err) + } + if len(logs) != 2 || logs[0].Status != model.ApiCallStatusSucceeded || logs[1].Status != model.ApiCallStatusFailed || logs[1].Billable { + t.Fatalf("merged delivery into generation: %+v", logs) + } +} + +func TestMediaRecoveryExpiredLeaseAndStagingLimits(t *testing.T) { + s, db := newMediaRecoveryTestService(t) + task := seedMediaTask(t, db, providerConfig{}) + checkpoint := &mediaCheckpoint{Mode: "image", StartedAt: time.Now(), Items: []mediaCheckpointItem{{}}} + if err := db.Model(task).Update("lease_expires_at", time.Now().Add(-time.Second)).Error; err != nil { + t.Fatal(err) + } + if err := s.saveMediaCheckpoint(task, checkpoint, "download"); !errors.Is(err, repository.ErrTaskStateConflict) { + t.Fatalf("expired lease wrote: %v", err) + } + for _, name := range []string{"../result-secret", "other", "", "result-dir/file"} { + if s.mediaTempPath(name) != "" { + t.Fatalf("unsafe staging name %s", name) + } + } + if _, err := s.newMediaTemp(mediaStagingBudget + 1); err == nil { + t.Fatal("oversized reservation accepted") + } + file, err := s.newMediaTemp(mediaStagingBudget) + if err != nil { + t.Fatal(err) + } + defer file.Close() + if _, err := s.newMediaTemp(1); err == nil { + t.Fatal("staging budget exceeded") + } +} diff --git a/backend/internal/app/task_media_storage.go b/backend/internal/app/task_media_storage.go new file mode 100644 index 000000000..5dc108513 --- /dev/null +++ b/backend/internal/app/task_media_storage.go @@ -0,0 +1,301 @@ +package app + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "image" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "infinite-canvas/backend/internal/model" +) + +const mediaStagingBudget int64 = 512 << 20 +const mediaStagingTTL = 24 * time.Hour + +var mediaStagingMu sync.Mutex + +// Reserve the maximum response size before opening a file. Temporary files are +// sparse reservations, so concurrent downloads cannot oversubscribe this budget. +func (s *Service) newMediaTemp(limit int64) (*os.File, error) { + mediaStagingMu.Lock() + defer mediaStagingMu.Unlock() + dir := filepath.Join(s.dataDir, "media-staging") + if err := os.MkdirAll(dir, 0700); err != nil { + return nil, err + } + entries, err := os.ReadDir(dir) + if err != nil { + return nil, err + } + var used int64 + for _, entry := range entries { + if !strings.HasPrefix(entry.Name(), "result-") || !entry.Type().IsRegular() { + continue + } + info, err := entry.Info() + if err != nil { + return nil, err + } + if time.Since(info.ModTime()) > mediaStagingTTL { + if err := os.Remove(filepath.Join(dir, entry.Name())); err != nil { + return nil, err + } + continue + } + used += info.Size() + } + if limit <= 0 || limit > mediaStagingBudget || used > mediaStagingBudget-limit { + return nil, errors.New("作品暂存空间不足，请稍后重试保存") + } + file, err := os.CreateTemp(dir, "result-*") + if err != nil { + return nil, err + } + if err := file.Truncate(limit); err != nil { + file.Close() + os.Remove(file.Name()) + return nil, err + } + return file, nil +} + +func (s *Service) stageInlineMedia(data []byte) (string, error) { + if len(data) == 0 { + return "", errors.New("上游返回空作品") + } + file, err := s.newMediaTemp(int64(len(data))) + if err != nil { + return "", err + } + _, err = file.Write(data) + if err == nil { + err = file.Sync() + } + closeErr := file.Close() + if err == nil { + err = closeErr + } + if err != nil { + os.Remove(file.Name()) + return "", err + } + return filepath.Base(file.Name()), nil +} + +func (s *Service) downloadTaskMedia(ctx context.Context, config providerConfig, url, mode string) (name, mimeType string, err error) { + policy, err := s.RuntimePolicy() + if err != nil { + return "", "", err + } + limit := min(megabytes(policy.Resource.GeneratedFileMB), mediaStagingBudget) + downloadCtx, cancel := context.WithTimeout(withProviderRequestKind(ctx, "download"), 3*time.Minute) + defer cancel() + req, err := http.NewRequestWithContext(downloadCtx, http.MethodGet, url, nil) + if err != nil { + return "", "", err + } + started := time.Now() + statusCode := 0 + defer func() { + // Signed paths and net/url errors can contain bearer credentials. Keep + // the real error for retry classification, never in request logs. + logged := req.Clone(req.Context()) + redactedURL := *req.URL + redactedURL.Path, redactedURL.RawPath, redactedURL.RawQuery, redactedURL.Fragment = "/task-media", "", "", "" + redactedURL.User = nil + logged.URL = &redactedURL + var safeErr error + if err != nil { + safeErr = errors.New("结果下载失败") + } + recordProviderRequest(logged, started, statusCode, nil, safeErr) + }() + if _, err = ValidateOutboundURL(url); err != nil { + return "", "", err + } + if sameProviderOrigin(config.BaseURL, url) { + applyProviderAuth(req, config) + ApplyOutboundHeaders(req, config.Headers) + } + ApplyDefaultOutboundHeaders(req) + response, err := OutboundHTTPClient(3 * time.Minute).Do(req) + if err != nil { + return "", "", err + } + defer response.Body.Close() + statusCode = response.StatusCode + if statusCode < 200 || statusCode >= 300 { + return "", "", providerHTTPError{StatusCode: statusCode, Status: response.Status, RetryAfter: parseRetryAfter(response.Header.Get("Retry-After"), time.Now())} + } + if response.ContentLength > limit { + return "", "", errors.New("生成文件超过大小限制") + } + if response.ContentLength > 0 { + limit = response.ContentLength + } + file, err := s.newMediaTemp(limit) + if err != nil { + return "", "", err + } + defer func() { + _ = file.Close() + if err != nil { + _ = os.Remove(file.Name()) + } + }() + size, err := io.Copy(file, io.LimitReader(response.Body, limit+1)) + if err != nil { + return "", "", err + } + if size == 0 || size > limit { + return "", "", errors.New("生成文件为空或超过大小限制") + } + if response.ContentLength >= 0 && size != response.ContentLength { + return "", "", io.ErrUnexpectedEOF + } + if err = file.Truncate(size); err != nil { + return "", "", err + } + if _, err = file.Seek(0, io.SeekStart); err != nil { + return "", "", err + } + head := make([]byte, min(size, 512)) + if _, err = io.ReadFull(file, head); err != nil { + return "", "", err + } + mimeType = normalizedMediaMimeType(response.Header.Get("Content-Type"), head) + if mode == "audio" && mimeType == "application/ogg" { + mimeType = "audio/ogg" + } + if !strings.HasPrefix(mimeType, mode+"/") { + return "", "", errors.New("上游返回的内容不是有效的" + mode + "媒体") + } + if err = file.Sync(); err != nil { + return "", "", err + } + if err = file.Close(); err != nil { + return "", "", err + } + return filepath.Base(file.Name()), mimeType, nil +} + +func mediaUploadKey(taskID string, index int) *string { + sum := sha256.Sum256([]byte(fmt.Sprintf("task-media:%s:%d", taskID, index))) + key := hex.EncodeToString(sum[:]) + return &key +} + +func (s *Service) storeTaskMediaFile(task *model.Task, index int, path, mimeType, kind string, existing *model.Resource) (*model.Resource, error) { + info, err := os.Lstat(path) + if err != nil { + return nil, err + } + if !info.Mode().IsRegular() { + return nil, errors.New("暂存作品不是普通文件") + } + file, err := os.Open(path) + if err != nil { + return nil, err + } + defer file.Close() + info, err = file.Stat() + if err != nil { + return nil, err + } + if !info.Mode().IsRegular() { + return nil, errors.New("暂存作品不是普通文件") + } + width, height := 0, 0 + if kind == "image" { + config, _, err := image.DecodeConfig(file) + if err != nil { + // The standard library does not decode WebP/AVIF. Verify their + // container signatures instead of trusting a response MIME header. + head := make([]byte, 64) + n, _ := file.ReadAt(head, 0) + head = head[:n] + webp := mimeType == "image/webp" && n >= 12 && string(head[:4]) == "RIFF" && string(head[8:12]) == "WEBP" + avif := mimeType == "image/avif" && n >= 16 && string(head[4:8]) == "ftyp" && (strings.Contains(string(head[8:]), "avif") || strings.Contains(string(head[8:]), "avis")) + if !webp && !avif { + return nil, fmt.Errorf("图片内容校验失败：%w", err) + } + } + width, height = config.Width, config.Height + if _, err := file.Seek(0, io.SeekStart); err != nil { + return nil, err + } + } + day, err := s.reserveGeneratedResourceQuota(task.UserID, info.Size()) + if err != nil { + return nil, err + } + committed := false + defer func() { + if !committed { + s.releaseUserUploadQuota(task.UserID, day, info.Size()) + } + }() + var resource *model.Resource + if existing == nil { + resource, _, err = s.storeResourceWithWriter(task.UserID, kind, "generated."+extensionFromMimeType(mimeType), mimeType, info.Size(), width, height, 0, file, mediaUploadKey(task.ID, index), false, s.storeTaskMediaObject) + } else { + // Reuse the same object key after a failed upload; no orphan per retry. + resource = existing + if resource.UserID != task.UserID || resource.Size != info.Size() || resource.MimeType != mimeType { + return nil, errors.New("恢复文件与原始资源不一致") + } + resource.ETag, err = s.storeTaskMediaObject(resource, "generated."+extensionFromMimeType(mimeType), file) + if err == nil { + resource.Status, resource.Error, resource.UpdatedAt = model.ResourceStatusReady, "", time.Now() + err = s.repo.SaveResource(resource) + } + } + if err != nil { + return nil, err + } + s.commitUserUploadQuota(task.UserID, info.Size()) + committed = true + if existing != nil { + s.recordActivity(task.UserID, "resource", 1) + s.maybeStartPlaybackTranscode(resource) + } + return resource, nil +} + +// Unlike ordinary uploads, generated output preserves an OSS failure for +// explicit recovery instead of silently changing the storage destination. +func (s *Service) storeTaskMediaObject(resource *model.Resource, _ string, body io.Reader) (string, error) { + if resource.Provider == "local" { + return "", writeLocalResourceObject(filepath.Join(s.dataDir, "resources", filepath.FromSlash(resource.ObjectKey)), body) + } + setting, err := s.ossSettingForResource(resource.UserID, resource) + if err != nil { + return "", err + } + return putOSSObject(setting, resource.ObjectKey, resource.MimeType, resource.Size, body) +} + +// Cleanup is best effort only after durable completion. Expired leftovers are +// also collected on staging allocation, never by deleting resource objects. +func (s *Service) cleanupMediaCheckpointFiles(task *model.Task) { + checkpoint, err := s.decodeMediaCheckpoint(task) + if err != nil { + return + } + for _, item := range checkpoint.Items { + if path := s.mediaTempPath(item.TempName); path != "" { + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + _ = s.log(task.UserID, task.ID, "warn", "清理作品暂存文件失败", "") + } + } + } +} diff --git a/backend/internal/app/task_output.go b/backend/internal/app/task_output.go index f36f69bc8..89387800e 100644 --- a/backend/internal/app/task_output.go +++ b/backend/internal/app/task_output.go @@ -17,6 +17,8 @@ type TaskSummary struct { Type string `json:"type"` Status model.TaskStatus `json:"status"` Stage string `json:"stage"` + MediaStage string `json:"mediaStage,omitempty"` + CanRecoverMedia bool `json:"canRecoverMedia,omitempty"` Progress int `json:"progress"` Prompt string `json:"prompt"` Operation string `json:"operation,omitempty"` @@ -107,6 +109,8 @@ func taskSummaryForOutput(task model.Task) TaskSummary { Type: task.Type, Status: task.Status, Stage: task.Stage, + MediaStage: task.MediaStage, + CanRecoverMedia: task.MediaRecoveryJSON != "" && task.Status == model.TaskStatusFailed, Progress: task.Progress, Prompt: truncateRunes(task.Prompt, 500), Operation: task.Operation, @@ -285,6 +289,7 @@ func truncateRunes(value string, limit int) string { } func taskForOutput(task model.Task) *model.Task { + task.CanRecoverMedia = task.MediaRecoveryJSON != "" && task.Status == model.TaskStatusFailed task.Diagnostic = taskExecutionDiagnostic(&task) task.InputJSON = publicTaskInputJSON(task.InputJSON) // 普通任务接口只暴露前台模型身份；渠道模型和供应线路属于管理员内部信息。 diff --git a/backend/internal/app/task_route_executor.go b/backend/internal/app/task_route_executor.go index 31222c8b2..8020dd507 100644 --- a/backend/internal/app/task_route_executor.go +++ b/backend/internal/app/task_route_executor.go @@ -2,6 +2,7 @@ package app import ( "context" + "errors" "infinite-canvas/backend/internal/model" ) @@ -51,6 +52,13 @@ func (e *taskRouteExecutor) execute(ctx context.Context, task *model.Task, attem if stateErr := e.port.refreshTaskProviderState(task); stateErr != nil { return taskRouteExecutionResult{}, stateErr } + var deliveryFailure *mediaRecoveryError + if errors.As(execution.err, &deliveryFailure) { + // The provider succeeded. Delivery must never enter route failover. + e.port.finishTaskRouteAttempt(attempt, task, nil) + execution.providerSucceeded = true + return execution, nil + } e.port.finishTaskRouteAttempt(attempt, task, execution.err) if execution.err == nil { break diff --git a/backend/internal/app/task_terminal.go b/backend/internal/app/task_terminal.go index 93cd8d54a..c335cb4c9 100644 --- a/backend/internal/app/task_terminal.go +++ b/backend/internal/app/task_terminal.go @@ -218,7 +218,7 @@ func (c *taskTerminalCoordinator) handleSuccess(task *model.Task) error { if fetchErr != nil { completionErr = fmt.Errorf("任务成功后读取任务产物失败：%w", fetchErr) _ = c.logger.log(task.UserID, task.ID, "error", "任务成功但读取任务产物失败", fetchErr.Error()) - } else { + } else if completedTask.MediaRecoveryJSON == "" { if registerErr := c.outputs.RegisterTaskOutputFromTask(*completedTask); registerErr != nil { // 任务成功与产物登记分开记账；登记失败保持步骤异常，允许项目页幂等补登记。 _ = c.logger.log(task.UserID, task.ID, "error", "任务成功但项目产物登记失败", registerErr.Error()) diff --git a/backend/internal/app/task_worker.go b/backend/internal/app/task_worker.go index 285af21aa..a3c8d1450 100644 --- a/backend/internal/app/task_worker.go +++ b/backend/internal/app/task_worker.go @@ -185,6 +185,10 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot if task.Type == model.TaskTypeTimelineRender { return w.processTimelineRender(task, ctx) } + if task.MediaRecoveryJSON != "" { + result, recoveryErr := s.resumeTaskMedia(ctx, task) + return s.finishTaskMediaRecovery(task, result, recoveryErr) + } s.markAgentMemoryCompactRunning(*task) task.Stage = "调用生成模型" @@ -215,6 +219,15 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot default: } result, canvasOps, err := routeResult.result, routeResult.canvasOps, routeResult.err + latestMedia, readErr := s.repo.Task(task.ID) + if readErr != nil { + return readErr + } + task.MediaRecoveryJSON, task.MediaStage = latestMedia.MediaRecoveryJSON, latestMedia.MediaStage + var deliveryFailure *mediaRecoveryError + if task.MediaRecoveryJSON != "" || errors.As(err, &deliveryFailure) { + return s.finishTaskMediaRecovery(task, result, err) + } providerSucceeded := routeResult.providerSucceeded if err == nil { result, err = s.persistGeneratedMediaResult(task.UserID, result) diff --git a/backend/internal/database/migrations.go b/backend/internal/database/migrations.go index 30133162f..f2bc85c51 100644 --- a/backend/internal/database/migrations.go +++ b/backend/internal/database/migrations.go @@ -11,7 +11,7 @@ import ( "gorm.io/gorm" ) -const CurrentSchemaVersion int64 = 33 +const CurrentSchemaVersion int64 = 34 const baselineSchemaChecksum = "sha256:open-ai-canvas-schema-v1-20260830" const schemaMigrationAppliedAtIndexChecksum = "sha256:schema-migrations-applied-at-index-v2-20260830" @@ -107,6 +107,16 @@ var schemaMigrations = []migration{ }}, {version: 32, name: "channel_model_tags", checksum: "sha256:channel-model-tags-v32", apply: migrateChannelModelTags}, {version: 33, name: "oauth_state_accepted_terms", checksum: "sha256:oauth-state-accepted-terms-v33", apply: migrateOAuthStateAcceptedTerms}, + {version: 34, name: "task_media_recovery", checksum: "sha256:task-media-recovery-v34", apply: func(tx *gorm.DB) error { + for _, field := range []string{"MediaRecoveryJSON", "MediaStage"} { + if !tx.Migrator().HasColumn(&model.Task{}, field) { + if err := tx.Migrator().AddColumn(&model.Task{}, field); err != nil { + return err + } + } + } + return nil + }}, } func migrateChannelModelTags(tx *gorm.DB) error { diff --git a/backend/internal/database/task_media_recovery_test.go b/backend/internal/database/task_media_recovery_test.go new file mode 100644 index 000000000..edd75e35f --- /dev/null +++ b/backend/internal/database/task_media_recovery_test.go @@ -0,0 +1,38 @@ +package database + +import ( + "infinite-canvas/backend/internal/model" + "testing" +) + +func TestTaskMediaRecoveryMigrationPreservesHistoricalTasks(t *testing.T) { + db, err := Open(Config{Driver: "sqlite", DSN: ":memory:"}) + if err != nil { + t.Fatal(err) + } + sqlDB, _ := db.DB() + t.Cleanup(func() { _ = sqlDB.Close() }) + if err := db.Exec(`CREATE TABLE tasks (id TEXT PRIMARY KEY, status TEXT, error TEXT)`).Error; err != nil { + t.Fatal(err) + } + if err := db.Exec(`INSERT INTO tasks VALUES ('old-task', 'failed', 'download failed')`).Error; err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + if err := schemaMigrations[33].apply(db); err != nil { + t.Fatal(err) + } + } + var task model.Task + if err := db.First(&task, "id = ?", "old-task").Error; err != nil { + t.Fatal(err) + } + if task.Status != model.TaskStatusFailed || task.Error != "download failed" || task.MediaRecoveryJSON != "" || task.MediaStage != "" { + t.Fatalf("migration rewrote historical task: %+v", task) + } + for _, field := range []string{"media_stage", "media_recovery_json"} { + if !db.Migrator().HasColumn(&model.Task{}, field) { + t.Fatalf("missing %s", field) + } + } +} diff --git a/backend/internal/handler/api_test.go b/backend/internal/handler/api_test.go index 919b6092d..b0fc516a9 100644 --- a/backend/internal/handler/api_test.go +++ b/backend/internal/handler/api_test.go @@ -17,10 +17,11 @@ func TestRegisterCanvasAPIExposesOpenAPIAndProjects(t *testing.T) { RegisterCanvasAPI(router.Group("/api"), &service.Service{}) wanted := map[string]bool{ - "GET /api/openapi.yaml": false, - "GET /api/projects": false, - "POST /api/tasks": false, - "GET /api/resources": false, + "GET /api/openapi.yaml": false, + "GET /api/projects": false, + "POST /api/tasks": false, + "POST /api/tasks/:id/recover-media": false, + "GET /api/resources": false, } for _, route := range router.Routes() { key := route.Method + " " + route.Path diff --git a/backend/internal/handler/openapi.yaml b/backend/internal/handler/openapi.yaml index 2d6f194a6..1e644de14 100644 --- a/backend/internal/handler/openapi.yaml +++ b/backend/internal/handler/openapi.yaml @@ -1067,6 +1067,25 @@ paths: $ref: "#/components/responses/Envelope" "400": $ref: "#/components/responses/Error" + /tasks/{id}/recover-media: + post: + tags: [tasks] + summary: 按原任务 ID 重试作品保存，不重新生成或创建计费订单 + description: 仅当前用户有恢复记录的保存失败任务可重入队；已在队列或运行中时幂等返回。每任务最多手动恢复三次，间隔至少一分钟。已取消、已退款或归属项目不可用时拒绝。 + parameters: + - in: path + name: id + required: true + schema: { type: string } + responses: + "200": + $ref: "#/components/responses/Envelope" + "400": + $ref: "#/components/responses/Error" + "401": + $ref: "#/components/responses/Error" + "404": + $ref: "#/components/responses/Error" /tasks/{id}/text-events: get: tags: [tasks] diff --git a/backend/internal/handler/routes.go b/backend/internal/handler/routes.go index d3c1ed901..7055530a5 100644 --- a/backend/internal/handler/routes.go +++ b/backend/internal/handler/routes.go @@ -202,6 +202,23 @@ func RegisterTaskRoutes(r *gin.RouterGroup, svc *service.Service) { } ok(c, task) }) + r.POST("/tasks/:id/recover-media", func(c *gin.Context) { + user, err := currentUser(c, svc) + if err != nil { + failService(c, err) + return + } + policy, available := loadRuntimePolicy(c, svc) + if !available || !enforceRateLimit(c, "task-media-recovery:"+user.ID, policy.Request.TaskCreatePerMinute, time.Minute) { + return + } + task, err := svc.RecoverTaskMedia(user.ID, c.Param("id")) + if err != nil { + failService(c, err) + return + } + ok(c, task) + }) r.POST("/tasks/:id/query-provider", func(c *gin.Context) { user, err := currentUser(c, svc) if err != nil { diff --git a/backend/internal/handler/task_media_recovery_test.go b/backend/internal/handler/task_media_recovery_test.go new file mode 100644 index 000000000..42e7a42ad --- /dev/null +++ b/backend/internal/handler/task_media_recovery_test.go @@ -0,0 +1,64 @@ +package handler + +import ( + "github.com/gin-gonic/gin" + "infinite-canvas/backend/internal/auth" + "infinite-canvas/backend/internal/database" + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/repository" + "infinite-canvas/backend/internal/service" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestTaskMediaRecoveryHTTPAuthenticationAndOwnership(t *testing.T) { + gin.SetMode(gin.TestMode) + db, err := database.Open(database.Config{Driver: "sqlite", DSN: ":memory:"}) + if err != nil { + t.Fatal(err) + } + sqlDB, _ := db.DB() + sqlDB.SetMaxOpenConns(1) + t.Cleanup(func() { _ = sqlDB.Close() }) + if err := database.MigrateSchema(db); err != nil { + t.Fatal(err) + } + for _, item := range []any{ + &model.User{ID: "media-user", Username: "media-user", Email: "media@example.invalid", Role: model.UserRoleUser, Status: model.UserStatusActive}, + &model.AuthSession{ID: "media-session", UserID: "media-user", TokenHash: auth.HashToken("test-token"), ExpiresAt: time.Now().Add(time.Hour)}, + &model.Task{ID: "legacy-task", UserID: "media-user", Status: model.TaskStatusFailed}, + &model.Task{ID: "foreign-task", UserID: "other-user", Status: model.TaskStatusFailed}, + } { + if err := db.Create(item).Error; err != nil { + t.Fatal(err) + } + } + router := gin.New() + svc := service.New(repository.New(db), t.TempDir()) + previous := runtimeService + ConfigureRuntime(svc) + t.Cleanup(func() { ConfigureRuntime(previous) }) + RegisterTaskRoutes(router.Group("/api"), svc) + for _, tc := range []struct { + id string + auth bool + want int + }{ + {"legacy-task", false, http.StatusUnauthorized}, + {"legacy-task", true, http.StatusBadRequest}, + {"foreign-task", true, http.StatusNotFound}, + {"missing-task", true, http.StatusNotFound}, + } { + r := httptest.NewRequest(http.MethodPost, "/api/tasks/"+tc.id+"/recover-media", nil) + if tc.auth { + r.AddCookie(&http.Cookie{Name: service.SessionCookieName, Value: "media-session.test-token"}) + } + w := httptest.NewRecorder() + router.ServeHTTP(w, r) + if w.Code != tc.want { + t.Fatalf("%s auth=%v: %d %s", tc.id, tc.auth, w.Code, w.Body.String()) + } + } +} diff --git a/backend/internal/model/models_channel.go b/backend/internal/model/models_channel.go index f877b5d9c..fec005805 100644 --- a/backend/internal/model/models_channel.go +++ b/backend/internal/model/models_channel.go @@ -105,6 +105,7 @@ type ApiCallLog struct { ChannelName string `json:"channelName" gorm:"-"` TaskID string `json:"taskId,omitempty" gorm:"index;size:36"` TaskStatus TaskStatus `json:"taskStatus,omitempty" gorm:"-"` + MediaStage string `json:"mediaStage,omitempty" gorm:"-"` BillingOrderID string `json:"billingOrderId,omitempty" gorm:"index;size:36"` BillingStatus BillingStatus `json:"billingStatus,omitempty" gorm:"-"` BillingAmount int64 `json:"billingAmountMicrocredits" gorm:"-"` diff --git a/backend/internal/model/models_task.go b/backend/internal/model/models_task.go index d743f3f97..43b86537c 100644 --- a/backend/internal/model/models_task.go +++ b/backend/internal/model/models_task.go @@ -46,13 +46,17 @@ type Task struct { 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;index:idx_tasks_user_project_created,priority:3"` - UpdatedAt time.Time `json:"updatedAt"` + // MediaRecoveryJSON is an encrypted server-only checkpoint, never a public result. + MediaRecoveryJSON string `json:"-" gorm:"type:text"` + MediaStage string `json:"mediaStage,omitempty" gorm:"size:24"` + CanRecoverMedia bool `json:"canRecoverMedia,omitempty" gorm:"-"` + 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;index:idx_tasks_user_project_created,priority:3"` + UpdatedAt time.Time `json:"updatedAt"` } // TaskTextDelta 只保存可回放窗口内的文本增量；最终正文和失败草稿分别归并到 Task.ResultJSON 与 Task.TextDraft。 diff --git a/backend/internal/repository/analytics.go b/backend/internal/repository/analytics.go index 89ebc5db6..316799483 100644 --- a/backend/internal/repository/analytics.go +++ b/backend/internal/repository/analytics.go @@ -144,7 +144,7 @@ func (r *Repository) filteredAPICallLogQuery(filter APICallLogFilter) *gorm.DB { query = query.Where("api_call_logs.request_kind = ?", "download") case "all": default: - query = visibleAPICallLogQuery(query).Where("COALESCE(api_call_logs.request_kind, '') <> ?", "download") + query = visibleAPICallLogQuery(query).Where("COALESCE(api_call_logs.request_kind, '') NOT IN ?", []string{"download", "upload", "local_save", "register"}) } if value := strings.TrimSpace(filter.Keyword); value != "" { pattern := "%" + strings.ToLower(value) + "%" @@ -167,7 +167,7 @@ func (r *Repository) APICallLogTasks(ids []string) ([]model.Task, error) { return []model.Task{}, nil } var tasks []model.Task - err := r.db.Select("id", "user_id", "type", "status", "result_json").Where("id IN ?", ids).Find(&tasks).Error + err := r.db.Select("id", "user_id", "type", "status", "result_json", "media_stage").Where("id IN ?", ids).Find(&tasks).Error return tasks, err } diff --git a/backend/internal/repository/analytics_test.go b/backend/internal/repository/analytics_test.go index 56c35e462..21a801cbc 100644 --- a/backend/internal/repository/analytics_test.go +++ b/backend/internal/repository/analytics_test.go @@ -32,6 +32,9 @@ func TestAPICallLogRecordTypeFiltersListAndExport(t *testing.T) { {ID: "poll", Capability: "video", RequestKind: "poll", CreatedAt: now}, {ID: "video-download", Capability: "video", RequestKind: "download", CreatedAt: now}, {ID: "image-download", Capability: "image", RequestKind: "download", CreatedAt: now}, + {ID: "upload", Capability: "image", RequestKind: "upload", CreatedAt: now}, + {ID: "local-save", Capability: "image", RequestKind: "local_save", CreatedAt: now}, + {ID: "register", Capability: "image", RequestKind: "register", CreatedAt: now}, } { if err := db.Create(&item).Error; err != nil { t.Fatal(err) @@ -41,7 +44,7 @@ func TestAPICallLogRecordTypeFiltersListAndExport(t *testing.T) { for _, tc := range []struct { kind string count int - }{{"", 1}, {"request", 1}, {"download", 2}, {"all", 4}} { + }{{"", 1}, {"request", 1}, {"download", 2}, {"all", 7}} { filter := APICallLogFilter{AnalyticsFilter: AnalyticsFilter{From: now.Add(-time.Hour), To: now.Add(time.Hour)}, RecordType: tc.kind} logs, total, err := repo.QueryAPICallLogs(filter) if err != nil { diff --git a/backend/internal/repository/repository.go b/backend/internal/repository/repository.go index 6903a733d..7d7d952ff 100644 --- a/backend/internal/repository/repository.go +++ b/backend/internal/repository/repository.go @@ -110,7 +110,7 @@ func (r *Repository) UserStorageUsage(userID string) (UserStorageUsage, error) { (SELECT COUNT(*) FROM canvas_projects WHERE user_id = ?) AS canvas_count, (SELECT COALESCE(SUM(length(CAST(COALESCE(payload_json, '') AS BLOB))), 0) FROM canvas_projects WHERE user_id = ?) AS canvas_bytes, (SELECT COUNT(*) FROM tasks WHERE user_id = ?) AS task_count, - (SELECT COALESCE(SUM(length(CAST(COALESCE(prompt, '') AS BLOB)) + length(CAST(COALESCE(input_json, '') AS BLOB)) + length(CAST(COALESCE(result_json, '') AS BLOB)) + length(CAST(COALESCE(text_draft, '') AS BLOB)) + length(CAST(COALESCE(error, '') AS BLOB))), 0) FROM tasks WHERE user_id = ?) + (SELECT COALESCE(SUM(length(CAST(COALESCE(prompt, '') AS BLOB)) + length(CAST(COALESCE(input_json, '') AS BLOB)) + length(CAST(COALESCE(result_json, '') AS BLOB)) + length(CAST(COALESCE(media_recovery_json, '') AS BLOB)) + length(CAST(COALESCE(text_draft, '') AS BLOB)) + length(CAST(COALESCE(error, '') AS BLOB))), 0) FROM tasks WHERE user_id = ?) + (SELECT COALESCE(SUM(length(CAST(COALESCE(message, '') AS BLOB)) + length(CAST(COALESCE(payload, '') AS BLOB))), 0) FROM task_logs WHERE user_id = ?) + (SELECT COALESCE(SUM(length(CAST(COALESCE(url, '') AS BLOB)) + length(CAST(COALESCE(payload, '') AS BLOB))), 0) FROM results WHERE user_id = ?) + (SELECT COALESCE(SUM(byte_count), 0) FROM task_text_delta WHERE user_id = ?) @@ -536,6 +536,12 @@ func (r *Repository) UpdateTaskProviderProgress(id string, progress int) error { } func (r *Repository) SaveTaskCompletion(task *model.Task, expected model.TaskStatus, results []model.Result) error { + return r.SaveTaskCompletionWithRegistration(task, expected, results, nil) +} + +// Registration and terminal success commit together; a failed asset write leaves +// the task recoverable and never exposes a false success to the canvas. +func (r *Repository) SaveTaskCompletionWithRegistration(task *model.Task, expected model.TaskStatus, results []model.Result, register func(*Repository) error) error { return r.db.Transaction(func(tx *gorm.DB) error { updated := taskLeaseWriter(tx.Model(&model.Task{}), task.LeaseOwner). Where("id = ? AND status = ?", task.ID, expected). @@ -551,6 +557,9 @@ func (r *Repository) SaveTaskCompletion(task *model.Task, expected model.TaskSta return err } } + if register != nil { + return register(New(tx)) + } return nil }) } @@ -696,7 +705,7 @@ func (r *Repository) Tasks(userID string, limit int, projectID string, activeOnl if limit <= 0 || limit > 100 { limit = 50 } - query := r.db.Select("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", "agent_run_id", "generation_id", "approval_id", "authorized_charge_microcredits", "execution_diagnostic_json", "cancellation_source", "cancellation_actor_id", "cancellation_requested_at"). + query := r.db.Select("id", "project_id", "type", "status", "stage", "media_stage", "media_recovery_json", "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", "agent_run_id", "generation_id", "approval_id", "authorized_charge_microcredits", "execution_diagnostic_json", "cancellation_source", "cancellation_actor_id", "cancellation_requested_at"). Where("user_id = ?", userID) if strings.TrimSpace(projectID) != "" { query = query.Where("project_id = ?", strings.TrimSpace(projectID)) diff --git a/backend/internal/repository/task_media_recovery.go b/backend/internal/repository/task_media_recovery.go new file mode 100644 index 000000000..55f3e803b --- /dev/null +++ b/backend/internal/repository/task_media_recovery.go @@ -0,0 +1,37 @@ +package repository + +import ( + "time" + + "infinite-canvas/backend/internal/model" +) + +// The existing task lease fences every checkpoint write, including after cancellation. +func (r *Repository) SaveTaskMediaCheckpoint(task *model.Task, encrypted, stage string) error { + result := taskLeaseWriter(r.db.Model(&model.Task{}), task.LeaseOwner). + Where("id = ? AND user_id = ? AND status = ? AND route_run = ?", task.ID, task.UserID, model.TaskStatusRunning, task.RouteRun). + Updates(map[string]any{"media_recovery_json": encrypted, "media_stage": stage, "stage": "作品已生成，正在保存", "updated_at": time.Now()}) + if result.Error != nil { + return result.Error + } + if result.RowsAffected != 1 { + return ErrTaskStateConflict + } + task.MediaRecoveryJSON, task.MediaStage = encrypted, stage + return nil +} + +// Manual recovery requeues the same task and order, never a new generation. +func (r *Repository) RequeueTaskMediaRecovery(task *model.Task, encrypted string) error { + now := time.Now() + result := r.db.Model(&model.Task{}). + Where("id = ? AND user_id = ? AND status = ? AND media_recovery_json = ? AND (lease_expires_at IS NULL OR lease_expires_at <= ?)", task.ID, task.UserID, model.TaskStatusFailed, task.MediaRecoveryJSON, now). + Updates(map[string]any{"status": model.TaskStatusQueued, "stage": "等待恢复作品保存", "error": "", "completed_at": nil, "next_poll_at": nil, "lease_owner": "", "lease_expires_at": nil, "media_recovery_json": encrypted, "updated_at": now}) + if result.Error != nil { + return result.Error + } + if result.RowsAffected != 1 { + return ErrTaskStateConflict + } + return nil +} diff --git a/docs/content/docs/backend/backend-database.mdx b/docs/content/docs/backend/backend-database.mdx index 97d21b4d8..81f404a8b 100644 --- a/docs/content/docs/backend/backend-database.mdx +++ b/docs/content/docs/backend/backend-database.mdx @@ -15,7 +15,14 @@ description: 影策后端核心持久化表与状态边界 ## 结构迁移与版本 -当前结构版本为 33。版本 33 `oauth_state_accepted_terms` 为 `o_auth_states` 增加非空布尔列 `accepted_terms`，默认 `false`；保留旧 state 的身份、PKCE、跳转目标及有效期，不替历史流程补造同意记录。升级前备份，按既有流程执行 `migrate-schema up` 和 `verify` 后再启动新后端。旧版本二进制不支持版本 33，回退需恢复匹配的数据库备份。 +当前结构版本为 34。版本 34 `task_media_recovery` 为 `tasks` 增加 `media_recovery_json`（text）和 `media_stage`（最长 24 字）。版本 33 `oauth_state_accepted_terms` 为 `o_auth_states` 增加非空布尔列 `accepted_terms`，默认 `false`，不替历史流程补造同意记录。升级前备份，按既有流程执行 `migrate-schema up` 和 `verify` 后再启动新后端。旧版本二进制不支持版本 34，回退需恢复匹配的数据库备份。 + +### 生成成功后的作品保存恢复 + +- 使用原任务主键、计费订单及现有 worker 租约，不新增执行批次或第二套队列。`media_stage` 为 `download | upload | local_save | register | checkpoint | completed`，表示当前结果的保存阶段，不改变上游生成日志的成功状态。 +- `media_recovery_json` 使用已有 settings-key 加密，记录媒体地址、暂存文件名、资源 ID 和有界重试计数；不保存媒体 base64，不向任务 JSON/摘要返回该字段。数据库与 `.settings-key` 必须一起备份；该字段计入任务存储用量。 +- 下载前写入恢复记录，下载后记录暂存文件，上传后记录资源。通过任务 ID 和输出索引生成稳定的资源上传键、素材 ID；任务成功、结果记录及素材登记在同一数据库事务提交，登记失败则不发布任务成功。 +- 记录保留在终态任务中，成功后清理对应暂存文件；任务的既有保留/删除策略不变。结构迁移只增加列，不扫描日志，不修改历史任务、结果或账单。没有恢复记录的历史失败任务不会自动变为可恢复。 ### 云端 Agent 运行时 diff --git a/docs/content/docs/backend/http-api.mdx b/docs/content/docs/backend/http-api.mdx index 5c1e42321..3fef7f8e9 100644 --- a/docs/content/docs/backend/http-api.mdx +++ b/docs/content/docs/backend/http-api.mdx @@ -13,6 +13,18 @@ description: 信封、错误码与 OpenAPI 模型 HTTP 失败及经过 Provider 错误转换的业务失败，在通用提示后追加经过过滤的 `；上游：具体原因`，供任务错误和画布卡片展示内容安全、提示词、尺寸等可操作信息。只读取上游 `error` / `message` / `msg` / `detail` 消息字段，不回传整个 JSON；消息中的 HTTP(S) URL 隐藏，包含余额、账务、凭据、请求诊断或 HTML 的消息不追加原文，保留通用分类提示。后台响应拦截规则仍对最终用户文案生效；不会修改历史任务错误或改变重试、退款策略。 +## 生成成功后的重试保存 + +`POST /tasks/:id/recover-media` 无需请求体。当前用户只能恢复自己有保存恢复记录的失败任务；重入队仍返回原任务 ID，不请求重新生成、不新建计费订单。已排队、运行或完成时幂等返回当前任务。取消任务、已退款订单、任务/账单归属不一致、项目已删除或归档时拒绝恢复。 + +任务详情和摘要新增 `mediaStage`、`canRecoverMedia`。后者表示失败任务存在保存恢复记录，实际操作仍校验权限、账单、项目及重试额度。每任务最多手动恢复三次，两次之间至少一分钟，每次手动操作只执行一次保存尝试；接口另受用户维度任务请求限流约束。普通 `/tasks/:id/retry` 不接受带保存恢复记录的任务，避免误走重新生成。 + +保存网络故障在原 worker 中最多自动重试四次（15 秒、1 分钟、3 分钟、10 分钟并加抖动；超过 20 分钟不再安排自动重试），使用 `next_poll_at` 释放 worker，而不是在请求内睡眠。不可重试错误直接保留为保存失败。画布和任务中心显示“作品已生成，正在保存”或“作品保存未完成”，失败按钮为“重试保存”。完成后客户端消费原任务的结果。 + +请求明细保留上游生成与结果下载的独立状态，新增 `upload`、`local_save`、`register` 内部非计费明细；列表和详情通过 `mediaStage` 展示交付摘要。`recordType=request` 排除这些保存明细和下载，`all` 查看全部阶段，`download` 仍仅查看下载。视频下载失败不会再覆盖生成请求的成功记录。 + +此恢复能力接入声明式媒体协议返回 URL/data URL 的任务路径；旧协议自行下载、协议二进制下载端点以及无恢复记录的历史失败任务不自动补录。历史任务需先核实上游结果可用性、资源归属及原账单，再单独制定修复动作。 + ## 素材彻底删除 `DELETE /assets/:id` 表示用户确认后的彻底删除，不依赖服务端当前记录是否已经同步为 `archived`。接口仍校验登录用户和素材归属，但不会因画布、任务、项目或历史快照仍有引用而拒绝；事务会删除素材及对应历史资源索引，并通过删除 Outbox 清理未被其他素材复用的物理对象。其他素材共享的同一物理对象继续保留。移入回收站仍通过素材写入接口把状态更新为 `archived`，不能用 DELETE 代替可恢复归档。 diff --git a/docs/content/docs/backend/task-runtime-stability.mdx b/docs/content/docs/backend/task-runtime-stability.mdx index 96cafc9b9..087dfd697 100644 --- a/docs/content/docs/backend/task-runtime-stability.mdx +++ b/docs/content/docs/backend/task-runtime-stability.mdx @@ -9,7 +9,18 @@ description: 热点回源保护、并发租约、故障边界和上线验收 任务中心打开详情只读取任务和日志，不在该路径调用画布结果同步。后台同步遇到短剧领域项目任务（`domainProjectId` 等于任务 `projectId`）或没有生成节点关联时跳过画布写回，避免把短剧项目 ID 当作画布 ID 查询。 -管理员请求明细列表及 CSV 导出支持 `recordType`：省略或 `request` 默认仅请求（继续隐藏视频轮询并排除下载）；`download` 只包含 `request_kind=download`；`all` 包含请求、轮询与下载。非法值返回参数错误，管理员权限校验保持不变；该筛选不改变统计聚合接口。 +管理员请求明细列表及 CSV 导出支持 `recordType`：省略或 `request` 默认仅请求（隐藏视频轮询，并排除下载、上传、本地保存、素材登记明细）；`download` 只包含 `request_kind=download`；`all` 包含全部阶段。非法值返回参数错误，管理员权限校验保持不变；该筛选不改变统计聚合接口。 + +## 媒体交付恢复的资源边界 + +声明式协议已经返回作品地址后，先加密保存任务恢复记录，再执行下载、存储、素材登记。交付错误不进入模型路由切换，不创建新生成任务。已下载文件优先复用；资源已就绪时跳过下载和上传，只补登记。配置为 OSS 的生成结果上传失败会明确保留错误，不像普通文件上传那样自动降级到本地存储；普通上传行为不变。 + +- 单进程同时最多两个媒体交付操作，满载时通过原任务 `next_poll_at` 延迟领取，不启动无界 goroutine 或额外轮询器。此限制不是跨实例全局限制，集群仍依赖既有任务租约与调度限制。 +- 下载流式写入数据目录的 `media-staging`，单次下载最多三分钟，文件上限取运行时生成文件限制与 512 MiB 中较小者；暂存目录逻辑占用上限为 512 MiB，先预留空间再写入。 +- 成功后立即清理对应暂存文件；超过 24 小时的暂存文件在下次分配空间时清理，不为闲置服务常驻一个清理循环。暂存文件不可用时只尝试原地址，不重新生成；只有内联结果且文件丢失时明确失败。 +- 多实例切换时，暂存文件只有位于共享数据卷才可直接复用；没有共享暂存文件但有有效源地址时可以重下载。数据目录和 settings-key 需要稳定保留，不能以临时容器目录代替持久化配置。 +- 出站下载沿用 SSRF 防护；同源才附带渠道鉴权。恢复地址仅加密留存，下载明细隐藏路径和查询参数，错误日志不写入原始带签名的 URL。 +- 自动重试有次数和时间窗口上限，手动重试也有限额。源地址过期、权限问题、配额不足等不会无限循环；不承诺已经过期且无暂存文件的作品一定可恢复。 章节提取采用独立提示词操作 `chapter_assets_extract`，受保护输出契约为 `chapter-assets/v1`，要求 `characters`、`scenes`、`props` 三个数组。场景和道具要求非空的 `name`、`description`、`prompt`；角色继续执行原角色契约校验。写入候选资产时场景映射为 `environment`，继续使用现有服务的项目归属、去重与写入校验，不新增数据库字段。 diff --git a/docs/content/docs/progress/pending-test.mdx b/docs/content/docs/progress/pending-test.mdx index f633e84f8..af15b5e76 100644 --- a/docs/content/docs/progress/pending-test.mdx +++ b/docs/content/docs/progress/pending-test.mdx @@ -5,6 +5,17 @@ description: 当前项目已实现功能的手工验证清单与未确认边界 # 待测试 +## 原任务 ID 媒体保存恢复 + +- [x] 故障注入：下载 503 后服务实例重建继续原任务、OSS 上传失败保留暂存文件及资源 ID、素材登记异常使任务成功事务回滚，登记恢复无需再次下载。 +- [x] 防重复与权限：不触发模型路由切换、不新增账单、重复点击幂等、跨用户/已取消/已退款/项目删除拒绝恢复、过期租约写入拒绝、自动/手动重试预算与暂存容量校验。 +- [x] 后端 `app`、`repository`、`database`、`handler` 四包回归通过；最后补验媒体恢复、路由与请求日志专项测试，以及 schema 34 幂等迁移和恢复接口鉴权/归属冒烟。测试使用隔离数据库。 +- [x] 前端类型检查与 31 项生成显示、画布结果同步、生成提交及重新生成相关测试通过。 +- [ ] 在真实画布和任务中心触发下载失败、OSS 上传失败、登记失败，确认各阶段提示、刷新后“重试保存”、结果回写及资产库仅一条；真实浏览器验收尚未执行。 +- [ ] 使用真实 OSS 配置进行网络中断/凭证修复后的恢复验收；当前上传故障测试使用不可达端点，不代表真实 OSS 成功链路已验证。 +- [ ] PostgreSQL 部署执行 schema 34 升级及多实例租约/共享暂存卷验收。未修改当前本地业务数据库、历史失败任务或账单。 +- [ ] 无恢复记录的历史失败任务单独核实源地址有效期与计费状态；本功能不从旧日志自动回填，不承诺恢复已过期的源文件。 + ## Live2D 运行库随部署交付 - Cubism Core 已纳入版本控制，保留原始版权声明并附第三方许可说明；前端构建和静态镜像直接包含运行文件，不再要求部署者手工安装。 diff --git a/web/src/components/canvas/canvas-node-content.tsx b/web/src/components/canvas/canvas-node-content.tsx index 55bd5190b..454071a74 100644 --- a/web/src/components/canvas/canvas-node-content.tsx +++ b/web/src/components/canvas/canvas-node-content.tsx @@ -184,6 +184,7 @@ function LoadingContent({ node, theme, onOpenTaskDetails }: Pick event.stopPropagation()} > - {node.metadata?.isBatchRoot ? "重新生成失败项" : "重新生成"} + {node.metadata?.taskCanRecoverMedia ? "重试保存（不重新生成）" : node.metadata?.isBatchRoot ? "重新生成失败项" : "重新生成"} )} diff --git a/web/src/components/canvas/canvas-node.tsx b/web/src/components/canvas/canvas-node.tsx index 8c00340fe..48b9bdd18 100644 --- a/web/src/components/canvas/canvas-node.tsx +++ b/web/src/components/canvas/canvas-node.tsx @@ -453,7 +453,7 @@ export const CanvasNode = React.memo(function CanvasNode({ } onClick={() => onSetBatchPrimary?.(data)} /> ) : null} {data.metadata?.status === "error" && data.metadata.resourceReloadAvailable ? } onClick={() => onReloadResource?.(data)} /> : null} - {data.metadata?.status === "error" ? } onClick={() => onRetry?.(data)} /> : null} + {data.metadata?.status === "error" ? } onClick={() => onRetry?.(data)} /> : null} {data.metadata?.status === "error" ? } danger onClick={() => deleteNode?.(data)} /> : null} diff --git a/web/src/lib/canvas/canvas-generation-task-sync.ts b/web/src/lib/canvas/canvas-generation-task-sync.ts index da09980fa..9fd929057 100644 --- a/web/src/lib/canvas/canvas-generation-task-sync.ts +++ b/web/src/lib/canvas/canvas-generation-task-sync.ts @@ -321,6 +321,8 @@ function completedTaskMetadata(task: GenerationTask): CanvasNodeMetadata { taskStatus: task.status, taskProgress: typeof task.progress === "number" && Number.isFinite(task.progress) ? Math.max(0, Math.min(100, Math.round(task.progress))) : 100, taskStage: task.stage, + taskMediaStage: task.mediaStage, + taskCanRecoverMedia: task.canRecoverMedia, taskStartedAt: task.startedAt, taskCompletedAt: task.completedAt, taskDurationMs: task.startedAt && task.completedAt ? Math.max(0, Date.parse(task.completedAt) - Date.parse(task.startedAt)) : undefined, diff --git a/web/src/lib/canvas/canvas-project-generation.ts b/web/src/lib/canvas/canvas-project-generation.ts index 352740053..095c52fb0 100644 --- a/web/src/lib/canvas/canvas-project-generation.ts +++ b/web/src/lib/canvas/canvas-project-generation.ts @@ -158,6 +158,8 @@ export function generationTaskMetadata(task: GenerationTask): CanvasNodeMetadata taskStatus: task.status, taskProgress: progress, taskStage: task.stage, + taskMediaStage: task.mediaStage, + taskCanRecoverMedia: task.canRecoverMedia, taskProvider: task.provider, taskStartedAt: task.startedAt, taskCompletedAt: task.completedAt, diff --git a/web/src/lib/canvas/canvas-task-state.ts b/web/src/lib/canvas/canvas-task-state.ts index 35be9e0d2..9e4614305 100644 --- a/web/src/lib/canvas/canvas-task-state.ts +++ b/web/src/lib/canvas/canvas-task-state.ts @@ -23,6 +23,8 @@ export function resetGenerationTaskMetadata(metadata: CanvasNodeMetadata | undef delete next.taskStatus; delete next.taskProgress; delete next.taskStage; + delete next.taskMediaStage; + delete next.taskCanRecoverMedia; delete next.taskProvider; delete next.taskStartedAt; delete next.taskCompletedAt; diff --git a/web/src/lib/generation-task-display.ts b/web/src/lib/generation-task-display.ts index 9eec8c80d..27014739a 100644 --- a/web/src/lib/generation-task-display.ts +++ b/web/src/lib/generation-task-display.ts @@ -8,13 +8,23 @@ export const statusLabel: Record = { cancelled: "已取消", }; -type GenerationTaskDisplayTarget = Pick; +type GenerationTaskDisplayTarget = Pick; + +export function mediaDeliverySummary(status: TaskStatus | undefined, stage: GenerationTask["mediaStage"]) { + if (!stage) return ""; + if (stage === "completed") return "生成成功 · 文件保存成功 · 素材登记成功"; + const prefix = stage === "download" || stage === "checkpoint" ? "生成成功" : stage === "register" ? "生成成功 · 文件保存成功" : "生成成功 · 下载成功"; + const action = { download: "下载", upload: "上传 OSS", local_save: "保存文件", register: "登记素材", checkpoint: "记录恢复信息" }[stage]; + return `${prefix} · ${action}${status === "failed" ? "失败" : status === "cancelled" ? "已停止" : "待完成"}`; +} export function isGenerationTaskSubmissionUncertain(task: GenerationTaskDisplayTarget) { return task.stage === "submission_unknown"; } export function generationTaskStatusLabel(task: GenerationTaskDisplayTarget) { + if (task.mediaStage && task.status === "failed") return "作品保存未完成"; + if (task.mediaStage && (task.status === "queued" || task.status === "running")) return "作品已生成，正在保存"; if (isGenerationTaskSubmissionUncertain(task)) return "提交结果待确认"; return statusLabel[task.status]; } @@ -27,6 +37,7 @@ export function generationTaskStageLabel(task: GenerationTaskDisplayTarget) { } export function generationTaskShowsProgress(task: GenerationTaskDisplayTarget) { + if (task.mediaStage && task.mediaStage !== "completed") return false; if (isGenerationTaskSubmissionUncertain(task)) return false; // 排队、后端接管和连接供应商都没有真实百分比。只有上游状态响应 // 已经写回任务后才显示进度，避免所有图片/视频长期停在同一个假数值。 diff --git a/web/src/pages/admin/components/api-log-detail-drawer.tsx b/web/src/pages/admin/components/api-log-detail-drawer.tsx index 0eba6baee..0febbe8d0 100644 --- a/web/src/pages/admin/components/api-log-detail-drawer.tsx +++ b/web/src/pages/admin/components/api-log-detail-drawer.tsx @@ -4,6 +4,7 @@ import { AdminEmpty } from "@/pages/admin/components/admin-ui"; import { RefreshCw } from "lucide-react"; import { formatCredits } from "@/constant/credits"; +import { mediaDeliverySummary } from "@/lib/generation-task-display"; import { getAdminApiLog, queryAdminApiLogTask, type ApiCallLog } from "@/services/api/auth"; import { AdminStatusBadge } from "./admin-ui"; @@ -80,6 +81,7 @@ function LogDetail({ log, querying, onQueryProviderTask }: { log: ApiCallLog; qu ], ["能力", capabilityText(log.capability)], ["请求阶段", requestKindText(log.requestKind)], + ["作品交付", mediaDeliverySummary(log.taskStatus, log.mediaStage) || "未记录保存阶段"], ["计费属性", log.billable ? "计费调用" : "不计费"], ["总耗时", {formatDuration(log.durationMs)}], ["视频轮询", log.capability === "video" ? {log.pollCount || 0} 次 : --], @@ -125,7 +127,7 @@ function LogDetail({ log, querying, onQueryProviderTask }: { log: ApiCallLog; qu ["上游地址", log.upstreamUrl ? {log.upstreamUrl} : --], ].map(([label, children], index) => ({ key: String(index), label, children })); - const canQueryProviderTask = log.capability === "video" && log.taskStatus === "failed" && Boolean(log.taskId && log.providerRequestId); + const canQueryProviderTask = !log.mediaStage && log.capability === "video" && log.taskStatus === "failed" && Boolean(log.taskId && log.providerRequestId); return (
@@ -159,7 +161,7 @@ function billingText(log: ApiCallLog) { } function requestKindText(value: ApiCallLog["requestKind"]) { - const labels: Partial> = { create: "模型生成", poll: "状态查询", download: "结果下载", repair: "结果修复" }; + const labels: Partial> = { create: "模型生成", poll: "状态查询", download: "结果下载", upload: "上传 OSS", local_save: "保存文件", register: "登记素材", repair: "结果修复" }; return labels[value] || "上游请求"; } diff --git a/web/src/pages/admin/logs/logs-page.tsx b/web/src/pages/admin/logs/logs-page.tsx index bc734d4c5..d79975716 100644 --- a/web/src/pages/admin/logs/logs-page.tsx +++ b/web/src/pages/admin/logs/logs-page.tsx @@ -10,6 +10,7 @@ import { PaginationBar } from "@/pages/admin/components/admin-ui"; import { MediaPreview } from "@/components/media-preview"; import { formatCredits } from "@/constant/credits"; import { useDebouncedValue } from "@/hooks/use-debounced-value"; +import { mediaDeliverySummary } from "@/lib/generation-task-display"; import { exportAdminApiLogs, listAdminApiLogs, type ApiCallLog } from "@/services/api/auth"; import { ApiLogDetailDrawer } from "../components/api-log-detail-drawer"; import { AdminPageFrame } from "../components/admin-shell"; @@ -361,12 +362,13 @@ function CallStatus({ log }: { log: ApiCallLog }) {
{requestKindText(log.requestKind)}
+ {log.mediaStage ?
{mediaDeliverySummary(log.taskStatus, log.mediaStage)}
: null} {log.capability === "video" ?
已轮询 {log.pollCount || 0} 次
: null}
); } function requestKindText(value: ApiCallLog["requestKind"]) { - const labels: Partial> = { create: "模型生成", poll: "状态查询", download: "结果下载", repair: "结果修复" }; + const labels: Partial> = { create: "模型生成", poll: "状态查询", download: "结果下载", upload: "上传 OSS", local_save: "保存文件", register: "登记素材", repair: "结果修复" }; return labels[value] || "上游请求"; } diff --git a/web/src/pages/canvas/use-canvas-generation-retry.ts b/web/src/pages/canvas/use-canvas-generation-retry.ts index 842ef797c..07b695ec4 100644 --- a/web/src/pages/canvas/use-canvas-generation-retry.ts +++ b/web/src/pages/canvas/use-canvas-generation-retry.ts @@ -30,7 +30,7 @@ import { modelCapabilityConfigFor } from "@/lib/model-capabilities"; import { navigateToSettings } from "@/lib/settings-navigation"; import type { Skill } from "@/services/api/skills"; import { skillRuntime, type SkillRuntimeMetadata } from "@/services/skill-runtime"; -import type { GenerationTask } from "@/services/api/task-center"; +import { queryGenerationTask, recoverGenerationTaskMedia, waitForGenerationTask, type GenerationTask } from "@/services/api/task-center"; import { resolveImageUrl } from "@/services/image-storage"; import { resolveModelRequestConfig, useConfigStore, useEffectiveConfig } from "@/stores/use-config-store"; import type { Asset } from "@/stores/use-asset-store"; @@ -75,6 +75,35 @@ export function useCanvasGenerationRetry({ return useCallback( async (node: CanvasNodeData) => { + // Canvas caches may predate the failure. Consult the original task + // before deciding to submit another paid generation. + let sourceTask: GenerationTask | undefined; + if (node.metadata?.taskId) { + try { + sourceTask = await queryGenerationTask(node.metadata.taskId); + } catch (error) { + message.error(error instanceof Error ? error.message : "无法核对原任务，请稍后重试"); + return; + } + } + if (sourceTask?.mediaStage && sourceTask.status !== "cancelled") { + const controller = startGenerationRequest(node.id, node.id, node.id); + try { + const initialTask = sourceTask.status === "failed" ? await recoverGenerationTaskMedia(sourceTask.id) : sourceTask; + if (controller.signal.aborted) return; + bindGenerationTask(node.id, initialTask); + const completed = await waitForGenerationTask(initialTask.id, { + initialTask, signal: controller.signal, timeoutMs: 25 * 60_000, + onTaskUpdate: (task) => bindGenerationTask(node.id, task), + }); + await applyGenerationTaskResult(node.id, completed); + } catch (error) { + if (!controller.signal.aborted) message.error(error instanceof Error ? error.message : "恢复作品保存失败"); + } finally { + finishGenerationRequest(node.id, controller); + } + return; + } const retryMode = retryModeForNode(node.type); if (!retryMode) { message.warning("当前节点不能使用通用生成重试"); diff --git a/web/src/pages/tasks/index.tsx b/web/src/pages/tasks/index.tsx index 25ad4a703..cc68d7b09 100644 --- a/web/src/pages/tasks/index.tsx +++ b/web/src/pages/tasks/index.tsx @@ -10,11 +10,11 @@ import { MediaPreview } from "@/components/media-preview"; import { PageHeader, PaginationBar, WorkspacePage } from "@/components/layout/workspace-page"; import { WorkspaceState } from "@/components/layout/workspace-state"; import { CONTENT_MODERATION_ERROR_CODE, generationErrorMessage, isContentModerationError } from "@/lib/generation-error"; -import { formatTaskKind, operationOptions, statusLabel } from "@/lib/generation-task-display"; +import { formatTaskKind, generationTaskStatusLabel, mediaDeliverySummary, operationOptions, statusLabel } from "@/lib/generation-task-display"; import { buildVideoOperationPrompt } from "@/lib/prompts"; import { backendProviderConfig, logicalModelIDForConfig } from "@/services/api/generation-task"; -import { createGenerationTask, formatTaskLog, listGenerationTasks, listTaskLogs, queryFailedVideoProviderTask, queryGenerationTask, retryGenerationTask, type CreateTaskInput, type GenerationTask, type TaskLog } from "@/services/api/task-center"; +import { createGenerationTask, formatTaskLog, listGenerationTasks, listTaskLogs, queryFailedVideoProviderTask, queryGenerationTask, retryGenerationTask, recoverGenerationTaskMedia, type CreateTaskInput, type GenerationTask, type TaskLog } from "@/services/api/task-center"; import { syncGenerationTaskToCanvasStore } from "@/lib/canvas/canvas-generation-task-sync"; import { useCanvasStore } from "@/stores/canvas/use-canvas-store"; import { resolveModelRequestConfig, useConfigStore, useEffectiveConfig } from "@/stores/use-config-store"; @@ -299,15 +299,18 @@ export default function TasksPage() { }, [loadTasks]); const runAction = async (id: string) => { - const currentTask = tasksRef.current.find((task) => task.id === id); setActingId(id); try { - const next = await retryGenerationTask(id); + const currentTask = await queryGenerationTask(id); + const savingMedia = Boolean(currentTask.mediaStage); + const next = currentTask.status === "queued" || currentTask.status === "running" || currentTask.status === "succeeded" + ? currentTask + : savingMedia ? await recoverGenerationTaskMedia(id) : await retryGenerationTask(id); setTasks((items) => items.map((item) => (item.id === id ? next : item))); setDetailTask((current) => (current?.id === id ? { ...current, ...next } : current)); setStatusFilter("active"); setPage(1); - message.success("任务已重新入队"); + message.success(next.status === "succeeded" ? "任务已完成" : savingMedia ? "正在恢复作品保存，不会重新生成" : "任务已在队列中"); } catch (error) { message.error(error instanceof Error ? error.message : "操作失败"); } finally { @@ -482,7 +485,8 @@ export default function TasksPage() { {detailTask ? (
- + + {detailTask.mediaStage ? : null} @@ -495,6 +499,7 @@ export default function TasksPage() { {detailTask.providerCancelRequestedAt ? : null}
+ {detailTask.canRecoverMedia ? : null} {canQueryProviderTask(detailTask) ? : null} {isTaskFailed(detailTask) ? : null}
@@ -538,7 +543,7 @@ export default function TasksPage() { } function canQueryProviderTask(task: GenerationTask) { - return task.status === "failed" && (task.type.startsWith("canvas_video") || task.type.startsWith("video_")) && Boolean(task.providerRequestId); + return task.status === "failed" && !task.mediaStage && (task.type.startsWith("canvas_video") || task.type.startsWith("video_")) && Boolean(task.providerRequestId); } function reconcileTaskSummaries(current: GenerationTask[], next: GenerationTask[]) { diff --git a/web/src/pages/tasks/task-grid-card.tsx b/web/src/pages/tasks/task-grid-card.tsx index c4a8684a0..75d54dd26 100644 --- a/web/src/pages/tasks/task-grid-card.tsx +++ b/web/src/pages/tasks/task-grid-card.tsx @@ -5,7 +5,7 @@ import { Eye, FileText, Image as ImageIcon, RotateCcw, Video } from "lucide-reac import { MediaPreview } from "@/components/media-preview"; import { CONTENT_MODERATION_ERROR_CODE, isContentModerationError } from "@/lib/generation-error"; -import { statusLabel } from "@/lib/generation-task-display"; +import { generationTaskStatusLabel } from "@/lib/generation-task-display"; import type { GenerationTask } from "@/services/api/task-center"; import { isTaskFailed, statusDotClassName, taskAttentionReason, TaskDate } from "./task-shared"; import { TaskVideoThumbnail } from "./task-video-thumbnail"; @@ -33,12 +33,12 @@ export function TaskGridCard({ task, actingId, onOpen, onRetry }: { task: Genera {isFailed ? ( - +
- {isActive ? ( + {isActive && task.mediaStage ?

{task.stage || "作品已生成，正在保存"}

: isActive ? (
{task.stage || "正在生成"} {task.progress || 0}% @@ -84,12 +84,12 @@ export function TaskListRow({ {isFailed ? ( - +