diff --git a/pkg/download/delete_recovery_test.go b/pkg/download/delete_recovery_test.go new file mode 100644 index 00000000..2c82d9ae --- /dev/null +++ b/pkg/download/delete_recovery_test.go @@ -0,0 +1,253 @@ +package download + +import ( + "errors" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/GopeedLab/gopeed/internal/fetcher" + "github.com/GopeedLab/gopeed/pkg/base" +) + +func TestFailedDeleteDuringResolveCanContinue(t *testing.T) { + release := make(chan struct{}) + var once sync.Once + defer once.Do(func() { close(release) }) + m := &deleteResolveManager{generationTestManager: &generationTestManager{ + resolveStarted: make(chan struct{}), resolveRelease: release, holdOpen: true, + }, returned: make(chan struct{})} + s := &singleFailureTaskDeleteStorage{Storage: NewMemStorage()} + d := NewDownloader(&DownloaderConfig{Storage: s, StorageDir: t.TempDir(), TempDir: t.TempDir(), FetchManagers: []fetcher.FetcherManager{m}}) + if err := d.Setup(); err != nil { + t.Fatal(err) + } + defer d.Close() + id, err := d.CreateDirect(&base.Request{URL: "generation://failed-delete-resolve"}, &base.Options{Path: t.TempDir()}) + if err != nil { + t.Fatal(err) + } + waitDeleteTest(t, m.resolveStarted) + task := d.GetTask(id) + s.fail.Store(true) + if err := d.Delete(&TaskFilter{IDs: []string{id}}, false); err == nil { + t.Fatal("expected storage deletion failure") + } + once.Do(func() { close(release) }) + waitDeleteTest(t, m.returned) + task.lock.Lock() + task.lock.Unlock() + if m.starts.Load() != 0 { + t.Fatal("invalidated startup must not reach Start") + } + if err := d.Continue(&TaskFilter{IDs: []string{id}}); err != nil { + t.Fatalf("failed deletion left task in %s, cannot continue: %v", d.taskStatus(task), err) + } + deadline := time.Now().Add(3 * time.Second) + for m.starts.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if m.starts.Load() != 1 { + t.Fatal("continued task did not start exactly once") + } + task.lock.Lock() + task.lock.Unlock() +} + +type deleteRecoveryStorage struct { + Storage + bucket string + fail atomic.Bool +} + +func (s *deleteRecoveryStorage) Delete(bucket, key string) error { + if bucket == s.bucket && s.fail.Swap(false) { + return errors.New("injected deletion failure") + } + return s.Storage.Delete(bucket, key) +} + +// Unlike the generation fixture, this fetcher stops its watcher on Pause and +// cannot be started again after Close, as with a dropped BT torrent. +type deleteRecoveryManager struct { + generationTestManager + failClose atomic.Bool + restored atomic.Int64 +} + +func (m *deleteRecoveryManager) Build() fetcher.Fetcher { + return m.wrap(m.generationTestManager.Build().(*generationTestFetcher)) +} + +func (m *deleteRecoveryManager) wrap(f *generationTestFetcher) *deleteRecoveryFetcher { + return &deleteRecoveryFetcher{generationTestFetcher: f, manager: m, waiting: make(chan struct{})} +} + +func (m *deleteRecoveryManager) Store(fetcher.Fetcher) (any, error) { + return &recoveryTestData{SeedBytes: 42}, nil +} + +func (m *deleteRecoveryManager) Restore() (any, func(*fetcher.FetcherMeta, any) fetcher.Fetcher) { + return &recoveryTestData{}, func(meta *fetcher.FetcherMeta, value any) fetcher.Fetcher { + m.restored.Store(value.(*recoveryTestData).SeedBytes) + return m.wrap(&generationTestFetcher{manager: &m.generationTestManager, meta: meta}) + } +} + +type deleteRecoveryFetcher struct { + *generationTestFetcher + manager *deleteRecoveryManager + closed atomic.Bool + running atomic.Bool + waiting chan struct{} + once sync.Once + waitRelease <-chan struct{} + waitSuccess bool +} + +func (f *deleteRecoveryFetcher) Start() error { + if f.closed.Load() { + return errors.New("cannot restart a closed fetcher") + } + f.running.Store(true) + return f.generationTestFetcher.Start() +} + +func (f *deleteRecoveryFetcher) Pause() error { + if f.running.Swap(false) { + f.done <- errors.New("paused") + } + return nil +} + +func (f *deleteRecoveryFetcher) Close() error { + f.closed.Store(true) + _ = f.Pause() + if f.manager.failClose.Swap(false) { + return errors.New("close failed after stopping the fetcher") + } + return nil +} + +func (f *deleteRecoveryFetcher) Wait() error { + f.once.Do(func() { close(f.waiting) }) + err := f.generationTestFetcher.Wait() + if f.waitRelease != nil { + <-f.waitRelease + if f.waitSuccess { + return nil + } + } + return err +} + +func TestFailedDeleteRunningTaskCanResumeAndComplete(t *testing.T) { + for _, stage := range []string{bucketTask, bucketTaskStats, bucketSave, "close", "files", "late-close", "late-close-error"} { + t.Run(stage, func(t *testing.T) { + s := &deleteRecoveryStorage{Storage: NewMemStorage(), bucket: stage} + lateClose := stage == "late-close" || stage == "late-close-error" + if lateClose { + s.bucket = bucketSave + } + m := &deleteRecoveryManager{generationTestManager: generationTestManager{holdOpen: true}} + d := NewDownloader(&DownloaderConfig{Storage: s, StorageDir: t.TempDir(), TempDir: t.TempDir(), FetchManagers: []fetcher.FetcherManager{m}}) + if err := d.Setup(); err != nil { + t.Fatal(err) + } + defer d.Close() + id, err := d.CreateDirect(&base.Request{URL: "generation://recover-running"}, &base.Options{Path: t.TempDir()}) + if err != nil { + t.Fatal(err) + } + task := d.GetTask(id) + task.lock.Lock() + f := task.fetcher.(*deleteRecoveryFetcher) + task.lock.Unlock() + waitDeleteTest(t, f.waiting) + watcher, _ := d.watchedTasks.Load(id) + var release chan struct{} + var releaseOnce sync.Once + if lateClose { + release = make(chan struct{}) + defer releaseOnce.Do(func() { close(release) }) + f.waitRelease = release + f.waitSuccess = stage == "late-close" + } + var blockedPath string + if stage == "files" { + blockedPath = task.Meta.SingleFilepath() + if err := os.MkdirAll(blockedPath, 0700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(blockedPath, "keep"), []byte("nonempty"), 0600); err != nil { + t.Fatal(err) + } + } + s.fail.Store(true) + m.failClose.Store(stage == "close") + if err := d.Delete(&TaskFilter{IDs: []string{id}}, stage == "files"); err == nil { + t.Fatal("expected deletion failure") + } + if f.running.Load() || d.taskStatus(task) != base.DownloadStatusPause { + t.Fatal("failed deletion must restore a paused task and stop transmission") + } + if !lateClose { + waitDeleteTest(t, watcher.(*taskWatcher).done) + } + var saved Task + if exists, err := s.Get(bucketTask, id, &saved); err != nil || !exists || saved.Status != base.DownloadStatusPause { + t.Fatalf("failed deletion did not persist paused recovery: exists=%v err=%v status=%s", exists, err, saved.Status) + } + var checkpoint recoveryTestData + exists, err := s.Get(bucketSave, id, &checkpoint) + if stage == "files" { + if err != nil || exists { + t.Fatalf("file cleanup must not leave a checkpoint for removed bytes: exists=%v err=%v", exists, err) + } + } else if err != nil || !exists || checkpoint.SeedBytes != 42 { + t.Fatalf("failed deletion lost checkpoint: exists=%v err=%v checkpoint=%+v", exists, err, checkpoint) + } + if blockedPath != "" { + if err := os.RemoveAll(blockedPath); err != nil { + t.Fatal(err) + } + } + if err := d.Continue(&TaskFilter{IDs: []string{id}}); err != nil { + t.Fatal(err) + } + task.lock.Lock() + resumed := task.fetcher.(*deleteRecoveryFetcher) + task.lock.Unlock() + waitDeleteTest(t, resumed.waiting) + if stage == bucketSave || stage == "close" || stage == "files" || lateClose { + wantSeed := int64(42) + if stage == "files" { + wantSeed = 0 + } + if resumed == f || m.restored.Load() != wantSeed { + t.Fatal("closed fetcher was not rebuilt from the retained checkpoint") + } + } + if lateClose { + releaseOnce.Do(func() { close(release) }) + waitDeleteTest(t, watcher.(*taskWatcher).done) + if d.taskStatus(task) != base.DownloadStatusRunning { + t.Fatal("old completion changed the replacement run's status") + } + current, exists := d.watchedTasks.Load(id) + if !exists || current.(*taskWatcher).fetcher != resumed { + t.Fatal("old completion removed the replacement watcher") + } + } + resumed.running.Store(false) + resumed.done <- nil + waitForTaskStatus(t, d, id, base.DownloadStatusDone, 3*time.Second) + if m.starts.Load() != 2 { + t.Fatal("resuming should start one new transfer") + } + }) + } +} diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index e94950bf..0d1979a5 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -540,6 +540,11 @@ func (d *Downloader) saveTask(task *Task) error { return nil } } + return d.saveTaskLocked(task) +} + +// saveTaskLocked requires the lifetime mutex when the task has one. +func (d *Downloader) saveTaskLocked(task *Task) error { data, err := task.fetcherManager.Store(task.fetcher) if err != nil { d.Logger.Error().Stack().Err(err).Msgf("serialize fetcher failed: %s", task.ID) @@ -1552,32 +1557,52 @@ func (d *Downloader) doDelete(task *Task, force bool) (err error) { task.lifetime.deleted.Store(true) task.statusLock.Lock() task.runGeneration++ + if task.Status == base.DownloadStatusRunning { + // Close/Pause can wake the completion watcher. Make it observe a + // stopped run before touching the fetcher, including on deletion errors. + task.updateStatus(base.DownloadStatusPause) + task.Progress.Speed = 0 + } task.statusLock.Unlock() task.lifetime.mu.Lock() defer task.lifetime.mu.Unlock() + defer d.releaseBlobTask(task) defer func() { if err != nil { - // Keep failed deletions recoverable without validating the old run. + // The startup generation stays invalid. Stop any physical transfer + // before restoring a state that Continue is allowed to restart. + var pauseErr error + if task.fetcher != nil { + pauseErr = task.fetcher.Pause() + } + task.statusLock.Lock() + if pauseErr != nil { + task.updateStatus(base.DownloadStatusError) + err = errors.Join(err, pauseErr) + } + 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 { return err } - if err := d.storage.Delete(bucketSave, task.ID); err != nil { - return err - } if err := d.storage.Delete(bucketTaskStats, task.ID); err != nil { return err } if task.fetcher != nil { + task.lifetime.restoreFetcher.Store(true) if err := task.fetcher.Close(); err != nil { return err } } + // Retain recovery state through Close, which can make the fetcher + // unusable even on failure. Clear it before removing any file bytes. + if err := d.storage.Delete(bucketSave, task.ID); err != nil { + return err + } if force && task.Meta.Res != nil { if task.Meta.Res.Name != "" { if err := os.RemoveAll(task.Meta.FolderPath()); err != nil { @@ -1897,18 +1922,34 @@ func (d *Downloader) getProtocolConfig(name string, v any) bool { return true } -// wait task done -func (d *Downloader) watch(task *Task) { - if _, loaded := d.watchedTasks.LoadOrStore(task.ID, true); loaded { +type taskWatcher struct { + fetcher fetcher.Fetcher + done chan struct{} +} + +// A replacement fetcher needs its own watcher even if Close has not yet +// returned the previous fetcher's final notification. +func (d *Downloader) watch(task *Task, f fetcher.Fetcher) { + task.lock.Lock() + if task.fetcher != f { + task.lock.Unlock() return } + if existing, loaded := d.watchedTasks.Load(task.ID); loaded && existing.(*taskWatcher).fetcher == f { + task.lock.Unlock() + return + } + watcher := &taskWatcher{fetcher: f, done: make(chan struct{})} + d.watchedTasks.Store(task.ID, watcher) + task.lock.Unlock() defer func() { - d.watchedTasks.Delete(task.ID) + d.watchedTasks.CompareAndDelete(task.ID, watcher) + close(watcher.done) }() // wait task upload done if task.Uploading { - if uploader, ok := task.fetcher.(fetcher.Uploader); ok { + if uploader, ok := f.(fetcher.Uploader); ok { go func() { err := uploader.WaitUpload() if err != nil { @@ -1916,9 +1957,19 @@ func (d *Downloader) watch(task *Task) { } // Check if the task is deleted - if d.GetTask(task.ID) != nil { + listed := d.GetTask(task.ID) != nil + task.lock.Lock() + task.statusLock.Lock() + current := task.fetcher == f && listed + var snapshot *Task + if current { task.Uploading = false - d.putTaskRecord(task, task.clone()) + snapshot = task.clone() + } + task.statusLock.Unlock() + task.lock.Unlock() + if current { + d.putTaskRecord(task, snapshot) } }() } @@ -1929,7 +1980,13 @@ func (d *Downloader) watch(task *Task) { } for { - err := task.fetcher.Wait() + task.lock.Lock() + current := task.fetcher == f + task.lock.Unlock() + if !current { + return + } + err := f.Wait() if err != nil { status := d.taskStatus(task) if status == base.DownloadStatusPause || status == base.DownloadStatusWait { @@ -1940,9 +1997,9 @@ func (d *Downloader) watch(task *Task) { sourceErr = d.blob.SourceError(task.Meta.Req.URL) } if sourceErr != nil { - d.doOnBlobSourceError(task, sourceErr) + d.handleOnErrorForFetcher(task, sourceErr, true, f) } else { - d.doOnError(task, err) + d.handleOnErrorForFetcher(task, err, false, f) } if d.taskStatus(task) == base.DownloadStatusRunning { continue @@ -1951,12 +2008,12 @@ func (d *Downloader) watch(task *Task) { } // When delete a not resolved task, need check if the task resource is nil - if task.Meta.Res == nil || d.GetTask(task.ID) == nil { + if d.GetTask(task.ID) == nil { return } if d.blob != nil && task.Meta.Req != nil { if err := d.blob.SourceError(task.Meta.Req.URL); err != nil { - d.doOnBlobSourceError(task, err) + d.handleOnErrorForFetcher(task, err, true, f) if d.taskStatus(task) == base.DownloadStatusRunning { continue } @@ -1964,7 +2021,7 @@ func (d *Downloader) watch(task *Task) { } } - if !d.markTaskDone(task) { + if !d.markTaskDoneForFetcher(task, f) { return } if err := d.captureTaskStats(task, false, true); err != nil { @@ -2078,10 +2135,18 @@ func (d *Downloader) doOnBlobSourceError(task *Task, err error) { } func (d *Downloader) handleOnError(task *Task, err error, resetFetcher bool) { + d.handleOnErrorForFetcher(task, err, resetFetcher, nil) +} + +func (d *Downloader) handleOnErrorForFetcher(task *Task, err error, resetFetcher bool, expected fetcher.Fetcher) { if task == nil || task.lock == nil { return } task.lock.Lock() + if expected != nil && task.fetcher != expected { + task.lock.Unlock() + return + } if !d.markTaskError(task) { task.lock.Unlock() return @@ -2170,10 +2235,11 @@ func (d *Downloader) resetTaskFetcher(task *Task) error { } func (d *Downloader) restoreTask(task *Task) error { - if task.fetcher == nil { + if task.fetcher == nil || task.lifetime.restoreFetcher.Load() { if err := d.restoreFetcher(task); err != nil { return err } + task.lifetime.restoreFetcher.Store(false) } return nil } @@ -2181,10 +2247,16 @@ func (d *Downloader) restoreTask(task *Task) error { func (d *Downloader) restoreFetcher(task *Task) error { v, f := task.fetcherManager.Restore() if v != nil { - _, err := d.storage.Get(bucketSave, task.ID, v) + present, err := d.storage.Get(bucketSave, task.ID, v) if err != nil { return err } + if !present && task.lifetime.restoreFetcher.Load() { + // File cleanup can fail after the checkpoint was removed. Resolve + // afresh rather than starting a closed protocol with empty state. + task.Meta.Res = nil + task.Progress.Downloaded = 0 + } } task.fetcher = f(task.Meta, v) if task.fetcher == nil { @@ -2398,7 +2470,17 @@ func (d *Downloader) prepareTask(task *Task) error { task.statusLock.Unlock() preparedTask.Meta = &preparedMeta preparedTask.NeedsPreparation = false - if err := d.saveTask(&preparedTask); err != nil { + // The candidate save and publication are one operation with respect to + // deletion, which reads the live metadata to decide which files to remove. + task.lifetime.mu.Lock() + defer task.lifetime.mu.Unlock() + task.statusLock.Lock() + current := !task.lifetime.deleted.Load() && task.runGeneration == preparedTask.runGeneration + task.statusLock.Unlock() + if !current { + return nil + } + if err := d.saveTaskLocked(&preparedTask); err != nil { return err } task.statusLock.Lock() @@ -2436,6 +2518,7 @@ func (d *Downloader) doStart(task *Task) (err error) { } started := false + var startedFetcher fetcher.Fetcher handler := func() error { task.lock.Lock() defer task.lock.Unlock() @@ -2517,6 +2600,7 @@ func (d *Downloader) doStart(task *Task) (err error) { return startErr } started = true + startedFetcher = task.fetcher if err := d.captureTaskStats(task, true, true); err != nil { // Statistics are auxiliary. A storage or serialization failure must not // turn an already-started download into a failed task. @@ -2538,7 +2622,7 @@ func (d *Downloader) doStart(task *Task) (err error) { if d.taskIsRunningGeneration(task, generation) { d.emit(EventKeyStart, task) } - go d.watch(task) + go d.watch(task, startedFetcher) } }() @@ -2660,14 +2744,21 @@ func (d *Downloader) taskStatus(task *Task) base.Status { } func (d *Downloader) markTaskDone(task *Task) bool { + return d.markTaskDoneForFetcher(task, nil) +} + +func (d *Downloader) markTaskDoneForFetcher(task *Task, expected fetcher.Fetcher) bool { if task == nil || task.statusLock == nil { return false } task.lock.Lock() defer task.lock.Unlock() + if expected != nil && task.fetcher != expected { + return false + } task.statusLock.Lock() defer task.statusLock.Unlock() - if task.lifetime.deleted.Load() || task.Status != base.DownloadStatusRunning { + if task.lifetime.deleted.Load() || task.Status != base.DownloadStatusRunning || task.Meta.Res == nil { return false } task.Progress.Used = task.timer.Used() diff --git a/pkg/download/model.go b/pkg/download/model.go index 77867b30..99c635f3 100644 --- a/pkg/download/model.go +++ b/pkg/download/model.go @@ -44,17 +44,19 @@ type CreateTaskBatchResult struct { ParentID string `json:"parentId,omitempty"` } -// 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 + // Close can make a fetcher unusable even when a later deletion step fails. + restoreFetcher atomic.Bool } +// 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) type Task struct { ID string `json:"id"` Protocol string `json:"protocol"`