diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index e7da2caf..17434c40 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -508,7 +508,31 @@ func (d *Downloader) setupFetcher(fm fetcher.FetcherManager, fetcher fetcher.Fet fetcher.Setup(ctl) } +// putTaskValue serializes late progress/completion callbacks with deletion. +// A snapshot may have been captured before Delete removed the task. +func (d *Downloader) putTaskValue(task *Task, bucket string, value any) error { + if task.lifetime != nil { + task.lifetime.mu.Lock() + defer task.lifetime.mu.Unlock() + if task.lifetime.deleted.Load() { + return nil + } + } + return d.storage.Put(bucket, task.ID, value) +} + +func (d *Downloader) putTaskRecord(task *Task, snapshot *Task) error { + return d.putTaskValue(task, bucketTask, snapshot) +} + func (d *Downloader) saveTask(task *Task) error { + if task.lifetime != nil { + task.lifetime.mu.Lock() + defer task.lifetime.mu.Unlock() + if task.lifetime.deleted.Load() { + return nil + } + } data, err := task.fetcherManager.Store(task.fetcher) if err != nil { d.Logger.Error().Stack().Err(err).Msgf("serialize fetcher failed: %s", task.ID) @@ -1220,7 +1244,7 @@ func (d *Downloader) Delete(filter *TaskFilter, force bool) error { for _, task := range tasks { if err := d.doDelete(task, force); err != nil { // Deletion can fail after removing bucketTask. Keep recovery durable. - if saveErr := d.storage.Put(bucketTask, task.ID, task.clone()); saveErr != nil { + if saveErr := d.putTaskRecord(task, task.clone()); saveErr != nil { d.Logger.Error().Err(saveErr).Msg("restore failed task deletion failed") if firstErr == nil { firstErr = saveErr @@ -1412,7 +1436,7 @@ func (d *Downloader) captureTaskStats(task *Task, live, force bool) error { return nil } - if err := d.storage.Put(bucketTaskStats, task.ID, &storedTaskStats{Snapshot: snapshot}); err != nil { + if err := d.putTaskValue(task, bucketTaskStats, &storedTaskStats{Snapshot: snapshot}); err != nil { return err } task.statsSnapshot = append(task.statsSnapshot[:0], snapshot...) @@ -1518,6 +1542,23 @@ func (d *Downloader) RuntimeStatus(id string) (*TaskRuntimeStatus, error) { } func (d *Downloader) doDelete(task *Task, force bool) (err error) { + task.lifetime.deleted.Store(true) + task.statusLock.Lock() + task.runGeneration++ + task.statusLock.Unlock() + task.lifetime.mu.Lock() + defer task.lifetime.mu.Unlock() + defer func() { + if err != nil { + // Keep failed deletions recoverable, but never revive the invalidated run. + task.statusLock.Lock() + if task.Status == base.DownloadStatusRunning { + task.updateStatus(base.DownloadStatusPause) + } + task.statusLock.Unlock() + task.lifetime.deleted.Store(false) + } + }() defer d.releaseBlobTask(task) err = func() error { if err := d.storage.Delete(bucketTask, task.ID); err != nil { @@ -1875,7 +1916,7 @@ func (d *Downloader) watch(task *Task) { // Check if the task is deleted if d.GetTask(task.ID) != nil { task.Uploading = false - d.storage.Put(bucketTask, task.ID, task.clone()) + d.putTaskRecord(task, task.clone()) } }() } @@ -1930,7 +1971,7 @@ func (d *Downloader) watch(task *Task) { task.statusLock.Lock() snapshot := task.clone() task.statusLock.Unlock() - d.storage.Put(bucketTask, task.ID, snapshot) + d.putTaskRecord(task, snapshot) d.emit(EventKeyDone, task) d.emit(EventKeyFinally, task, err) d.notifyRunning() @@ -2073,7 +2114,7 @@ func (d *Downloader) handleOnError(task *Task, err error, resetFetcher bool) { d.Logger.Warn().Err(statsErr).Msgf("persist failed task stats failed, task id: %s", task.ID) } if d.taskStatus(task) == base.DownloadStatusError { - if persistErr := d.storage.Put(bucketTask, task.ID, task.clone()); persistErr != nil { + if persistErr := d.putTaskRecord(task, task.clone()); persistErr != nil { d.Logger.Error().Stack().Err(persistErr).Msgf("persist failed task failed, task id: %s", task.ID) } } @@ -2201,7 +2242,7 @@ func (d *Downloader) doCreate(f fetcher.Fetcher, opts *base.Options, groupID str return } } - if err = d.storage.Put(bucketTask, task.ID, task.clone()); err != nil { + if err = d.putTaskRecord(task, task.clone()); err != nil { return } taskId = task.ID @@ -2368,7 +2409,7 @@ func (d *Downloader) prepareTask(task *Task) error { func (d *Downloader) doStart(task *Task) (err error) { var generation uint64 isReturn, err := d.statusMut(task, func() (isReturn bool, err error) { - if task.Status == base.DownloadStatusRunning || task.Status == base.DownloadStatusDone { + if task.lifetime.deleted.Load() || task.Status == base.DownloadStatusRunning || task.Status == base.DownloadStatusDone { isReturn = true return } @@ -2423,6 +2464,9 @@ func (d *Downloader) doStart(task *Task) (err error) { if err != nil { return err } + if !d.taskIsRunningGeneration(task, generation) { + return nil + } task.Meta.Res = task.fetcher.Meta().Res } @@ -2447,16 +2491,24 @@ func (d *Downloader) doStart(task *Task) (err error) { return err } + task.lifetime.mu.Lock() + if task.lifetime.deleted.Load() { + task.lifetime.mu.Unlock() + return nil + } task.statusLock.Lock() if task.Status != base.DownloadStatusRunning || task.runGeneration != generation { + task.lifetime.mu.Unlock() task.statusLock.Unlock() return nil } task.Progress.Speed = 0 task.timer.Start() task.statusLock.Unlock() - if err := task.fetcher.Start(); err != nil { - return err + startErr := task.fetcher.Start() + task.lifetime.mu.Unlock() + if startErr != nil { + return startErr } started = true if err := d.captureTaskStats(task, true, true); err != nil { @@ -2567,7 +2619,7 @@ func (d *Downloader) runPauseHandler(task *Task, generation uint64, pauseFetcher return false, err } } else { - if err := d.storage.Put(bucketTask, task.ID, task.clone()); err != nil { + if err := d.putTaskRecord(task, task.clone()); err != nil { return false, err } } @@ -2580,7 +2632,7 @@ func (d *Downloader) taskIsRunningGeneration(task *Task, generation uint64) bool } task.statusLock.Lock() defer task.statusLock.Unlock() - return task.Status == base.DownloadStatusRunning && task.runGeneration == generation + return !task.lifetime.deleted.Load() && task.Status == base.DownloadStatusRunning && task.runGeneration == generation } func (d *Downloader) taskIsPausedGeneration(task *Task, generation uint64) bool { @@ -2589,7 +2641,7 @@ func (d *Downloader) taskIsPausedGeneration(task *Task, generation uint64) bool } task.statusLock.Lock() defer task.statusLock.Unlock() - return (task.Status == base.DownloadStatusPause || task.Status == base.DownloadStatusWait) && task.runGeneration == generation + return !task.lifetime.deleted.Load() && (task.Status == base.DownloadStatusPause || task.Status == base.DownloadStatusWait) && task.runGeneration == generation } func (d *Downloader) taskStatus(task *Task) base.Status { @@ -2609,7 +2661,7 @@ func (d *Downloader) markTaskDone(task *Task) bool { defer task.lock.Unlock() task.statusLock.Lock() defer task.statusLock.Unlock() - if task.Status != base.DownloadStatusRunning { + if task.lifetime.deleted.Load() || task.Status != base.DownloadStatusRunning { return false } task.Progress.Used = task.timer.Used() @@ -2637,7 +2689,7 @@ func (d *Downloader) markTaskError(task *Task) bool { } task.statusLock.Lock() defer task.statusLock.Unlock() - if task.Status != base.DownloadStatusRunning { + if task.lifetime.deleted.Load() || task.Status != base.DownloadStatusRunning { return false } task.updateStatus(base.DownloadStatusError) @@ -2698,7 +2750,7 @@ func (d *Downloader) enqueueSingleExtraction(task *Task, downloadFilePath string // Set extraction status to queued task.Progress.ExtractStatus = ExtractStatusQueued d.emit(EventKeyProgress, task) - d.storage.Put(bucketTask, task.ID, task.clone()) + d.putTaskRecord(task, task.clone()) d.Logger.Info().Msgf("extraction queued, task id: %s, job id: %s", task.ID, jobID) // Create and enqueue the extraction job @@ -2727,7 +2779,7 @@ func (d *Downloader) enqueueMultiPartExtraction(task *Task, downloadFilePath str // Not all parts are ready yet - just set status to waiting, don't queue anything task.Progress.ExtractStatus = ExtractStatusWaitingParts d.emit(EventKeyProgress, task) - d.storage.Put(bucketTask, task.ID, task.clone()) + d.putTaskRecord(task, task.clone()) d.Logger.Info().Msgf("multi-part archive waiting for other parts, task id: %s, missing: %v", task.ID, missingParts) return } @@ -2743,14 +2795,14 @@ func (d *Downloader) enqueueMultiPartExtraction(task *Task, downloadFilePath str task.Progress.ExtractStatus = ExtractStatusDone task.Progress.ExtractProgress = 100 d.emit(EventKeyProgress, task) - d.storage.Put(bucketTask, task.ID, task.clone()) + d.putTaskRecord(task, task.clone()) d.Logger.Info().Msgf("multi-part archive extraction already handled by another part, task id: %s", task.ID) return } // This task claimed the extraction - status already set to queued in tryClaimMultiPartExtraction d.emit(EventKeyProgress, task) - d.storage.Put(bucketTask, task.ID, task.clone()) + d.putTaskRecord(task, task.clone()) jobID := "multipart:" + fullBaseName d.Logger.Info().Msgf("multi-part extraction queued, task id: %s, job id: %s", task.ID, jobID) @@ -2839,7 +2891,7 @@ func (d *Downloader) performExtraction(task *Task, archivePath string, destDir s task.Progress.ExtractStatus = ExtractStatusExtracting task.Progress.ExtractProgress = 0 d.emit(EventKeyProgress, task) - d.storage.Put(bucketTask, task.ID, task.clone()) + d.putTaskRecord(task, task.clone()) // Extract the archive extractErr := extractArchive(archivePath, destDir, opts.ArchivePassword, func(extractedFiles int, totalFiles int, progress int) { @@ -2859,7 +2911,7 @@ func (d *Downloader) performMultiPartExtraction(task *Task, firstPartPath string task.Progress.ExtractStatus = ExtractStatusExtracting task.Progress.ExtractProgress = 0 d.emit(EventKeyProgress, task) - d.storage.Put(bucketTask, task.ID, task.clone()) + d.putTaskRecord(task, task.clone()) d.Logger.Info().Msgf("starting multi-part archive extraction, first part: %s, task id: %s", firstPartPath, task.ID) @@ -3003,13 +3055,13 @@ func (d *Downloader) handleExtractionResult(task *Task, extractErr error, archiv d.Logger.Error().Err(extractErr).Msgf("auto extract archive failed, task id: %s", task.ID) task.Progress.ExtractStatus = ExtractStatusError d.emit(EventKeyProgress, task) - d.storage.Put(bucketTask, task.ID, task.clone()) + d.putTaskRecord(task, task.clone()) } else { d.Logger.Info().Msgf("auto extract archive completed, task id: %s", task.ID) task.Progress.ExtractStatus = ExtractStatusDone task.Progress.ExtractProgress = 100 d.emit(EventKeyProgress, task) - d.storage.Put(bucketTask, task.ID, task.clone()) + d.putTaskRecord(task, task.clone()) // Delete archive files after successful extraction if enabled if deleteAfterExtract { @@ -3049,12 +3101,15 @@ func (d *Downloader) updateMultiPartTasksStatus(sourceTask *Task, extractErr err task.Progress.ExtractStatus = status task.Progress.ExtractProgress = progress d.emit(EventKeyProgress, task) - d.storage.Put(bucketTask, task.ID, task.clone()) + d.putTaskRecord(task, task.clone()) } } } func initTask(task *Task) { + if task.lifetime == nil { + task.lifetime = &taskLifetime{} + } task.timer = util.NewTimer(task.Progress.Used) task.statusLock = &sync.Mutex{} diff --git a/pkg/download/model.go b/pkg/download/model.go index 766a7a69..d7bc655b 100644 --- a/pkg/download/model.go +++ b/pkg/download/model.go @@ -3,6 +3,7 @@ package download import ( "encoding/json" "sync" + "sync/atomic" "time" "github.com/GopeedLab/gopeed/internal/controller" @@ -46,6 +47,14 @@ type CreateTaskBatchResult struct { // Task represents a download unit. // Options flow path: // REST Request (model.CreateTask/ResolveTask) -> pkg/download Downloader -> Task.Meta.Opts (base.Options) -> Fetcher.Resolve/Start (e.g. internal/protocol/http/fetcher.go) +// taskLifetime is shared with preparation candidates. Its mutex serializes +// physical starts and persistence with deletion; the tombstone also invalidates +// work that is currently outside that mutex (for example Resolve). +type taskLifetime struct { + mu sync.Mutex + deleted atomic.Bool +} + type Task struct { ID string `json:"id"` Protocol string `json:"protocol"` @@ -87,6 +96,7 @@ type Task struct { blobRefLock *sync.Mutex blobURL string runGeneration uint64 + lifetime *taskLifetime speedArr []int64 uploadSpeedArr []int64 } @@ -123,6 +133,7 @@ func NewTask() *Task { } return &Task{ ID: id, + lifetime: &taskLifetime{}, Status: base.DownloadStatusReady, CreatedAt: time.Now(), UpdatedAt: time.Now(),