mirror of
https://github.com/GopeedLab/gopeed.git
synced 2026-11-10 18:40:27 +00:00
fix(download): invalidate deleted runs and serialize task persistence
This commit is contained in:
Levi committed
1 parent
4ffb172259
commit
917fc64046
2 files changed
+89
-23
No files matched your search
+78
-23
@@ -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{}
|
||||
|
||||
@@ -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(),
|
||||
|
||||
Reference in new issue
Block a user