Merge remote-tracking branch 'origin/main' into codex/resolve-pr1658

pkg/download/delete_lifecycle_test.go

@@ -0,0 +1,259 @@
+package download
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/GopeedLab/gopeed/internal/fetcher"
+	"github.com/GopeedLab/gopeed/pkg/base"
+)
+
+type deleteResolveManager struct {
+	*generationTestManager
+	returned chan struct{}
+}
+
+func (m *deleteResolveManager) Build() fetcher.Fetcher {
+	return &deleteResolveFetcher{generationTestFetcher: m.generationTestManager.Build().(*generationTestFetcher), returned: m.returned}
+}
+
+type deleteResolveFetcher struct {
+	*generationTestFetcher
+	returned chan struct{}
+}
+
+func (f *deleteResolveFetcher) Resolve(req *base.Request) error {
+	defer close(f.returned)
+	return f.generationTestFetcher.Resolve(req)
+}
+func waitDeleteTest(t *testing.T, ch <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-ch:
+	case <-time.After(3 * time.Second):
+		t.Fatal("operation did not finish")
+	}
+}
+func TestDeleteDuringResolveCannotRestartOrPersist(t *testing.T) {
+	for _, force := range []bool{false, true} {
+		for _, resolveErr := range []error{nil, errors.New("late resolve error")} {
+			t.Run(fmt.Sprintf("force=%v/error=%v", force, resolveErr != nil), func(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, resolveErr: resolveErr}, returned: make(chan struct{})}
+				d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), 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://delete"}, &base.Options{Path: t.TempDir(), Name: "delete.bin"})
+				if err != nil {
+					t.Fatal(err)
+				}
+				waitDeleteTest(t, m.resolveStarted)
+				task := d.GetTask(id)
+				if err := d.Delete(&TaskFilter{IDs: []string{id}}, force); err != nil {
+					t.Fatal(err)
+				}
+				once.Do(func() { close(release) })
+				waitDeleteTest(t, m.returned)
+				task.lock.Lock()
+				task.lock.Unlock()
+				// A stale error callback can run immediately after the startup handler exits.
+				d.doOnError(task, errors.New("late callback"))
+				if got := m.starts.Load(); got != 0 {
+					t.Errorf("deleted task started %d time(s)", got)
+				}
+				if d.GetTask(id) != nil {
+					t.Error("deleted task is still listed")
+				}
+				for _, bucket := range []string{bucketTask, bucketSave, bucketTaskStats} {
+					var value any
+					present, err := d.storage.Get(bucket, id, &value)
+					if err != nil {
+						t.Fatal(err)
+					}
+					if present {
+						t.Errorf("deleted task resurrected in %s", bucket)
+					}
+				}
+			})
+		}
+	}
+}
+
+type blockedTaskStorage struct {
+	Storage
+	entered chan struct{}
+	release chan struct{}
+	blockID string
+	once    sync.Once
+}
+
+func (s *blockedTaskStorage) Put(bucket, key string, value any) error {
+	if bucket == bucketTask && key == s.blockID {
+		s.once.Do(func() { close(s.entered); <-s.release })
+	}
+	return s.Storage.Put(bucket, key, value)
+}
+func TestDeleteSerializesInFlightTaskCheckpoint(t *testing.T) {
+	s := &blockedTaskStorage{Storage: NewMemStorage(), entered: make(chan struct{}), release: make(chan struct{})}
+	d := NewDownloader(&DownloaderConfig{Storage: s, StorageDir: t.TempDir(), TempDir: t.TempDir(), FetchManagers: []fetcher.FetcherManager{&generationTestManager{}}})
+	if err := d.Setup(); err != nil {
+		t.Fatal(err)
+	}
+	defer d.Close()
+	d.cfg.MaxRunning = 0
+	id, err := d.CreateDirect(&base.Request{URL: "generation://checkpoint"}, &base.Options{Path: t.TempDir()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	task := d.GetTask(id)
+	s.blockID = id
+	saved := make(chan struct{})
+	go func() { defer close(saved); _ = d.saveTask(task) }()
+	waitDeleteTest(t, s.entered)
+	deleted := make(chan struct{})
+	go func() { defer close(deleted); _ = d.Delete(&TaskFilter{IDs: []string{id}}, false) }()
+	deadline := time.Now().Add(3 * time.Second)
+	for d.GetTask(id) != nil && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	close(s.release)
+	waitDeleteTest(t, saved)
+	waitDeleteTest(t, deleted)
+	var value any
+	present, err := s.Get(bucketTask, id, &value)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if present {
+		t.Fatal("in-flight checkpoint restored a deleted task")
+	}
+}
+
+func TestDeleteDuringPreparationCannotStartOrPersist(t *testing.T) {
+	s := &blockedTaskStorage{Storage: NewMemStorage(), entered: make(chan struct{}), release: make(chan struct{})}
+	m := &generationTestManager{}
+	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()
+	d.cfg.MaxRunning = 0
+	id, err := d.CreateDirect(&base.Request{URL: "generation://preparation"}, &base.Options{Path: t.TempDir()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	task := d.GetTask(id)
+	s.blockID = id
+	d.cfg.MaxRunning = 1
+	if err := d.Continue(&TaskFilter{IDs: []string{id}}); err != nil {
+		t.Fatal(err)
+	}
+	waitDeleteTest(t, s.entered)
+	done := make(chan struct{})
+	go func() { defer close(done); _ = d.Delete(&TaskFilter{IDs: []string{id}}, true) }()
+	deadline := time.Now().Add(2 * time.Second)
+	for d.GetTask(id) != nil && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	close(s.release)
+	waitDeleteTest(t, done)
+	task.lock.Lock()
+	task.lock.Unlock()
+	if m.starts.Load() != 0 {
+		t.Fatal("deleted preparation started downloading")
+	}
+	var saved Task
+	if exists, err := s.Get(bucketTask, id, &saved); err != nil || exists {
+		t.Fatalf("deleted preparation persisted: exists=%v err=%v", exists, err)
+	}
+}
+func TestDeletedTaskRejectsLateSnapshotsAndCompletion(t *testing.T) {
+	d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir(), FetchManagers: []fetcher.FetcherManager{&generationTestManager{}}})
+	if err := d.Setup(); err != nil {
+		t.Fatal(err)
+	}
+	defer d.Close()
+	d.cfg.MaxRunning = 0
+	id, err := d.CreateDirect(&base.Request{URL: "generation://late"}, &base.Options{Path: t.TempDir()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	task := d.GetTask(id)
+	snapshot := task.clone()
+	task.fetcher.(*generationTestFetcher).stats = &fetcher.Stats{Snapshot: map[string]any{"bytes": 1}}
+	if err := d.Delete(&TaskFilter{IDs: []string{id}}, false); err != nil {
+		t.Fatal(err)
+	}
+	if err := d.putTaskRecord(task, snapshot); err != nil {
+		t.Fatal(err)
+	}
+	if err := d.captureTaskStats(task, false, true); err != nil {
+		t.Fatal(err)
+	}
+	if err := d.saveTask(task); err != nil {
+		t.Fatal(err)
+	}
+	if d.markTaskDone(task) {
+		t.Error("deleted task accepted a late completion")
+	}
+	for _, bucket := range []string{bucketTask, bucketSave, bucketTaskStats} {
+		var value any
+		if exists, err := d.storage.Get(bucket, id, &value); err != nil || exists {
+			t.Errorf("late callback restored %s: exists=%v err=%v", bucket, exists, err)
+		}
+	}
+}
+
+type singleFailureTaskDeleteStorage struct {
+	Storage
+	fail atomic.Bool
+}
+
+func (s *singleFailureTaskDeleteStorage) Delete(bucket, key string) error {
+	if bucket == bucketTask && s.fail.Swap(false) {
+		return errors.New("task deletion failed")
+	}
+	return s.Storage.Delete(bucket, key)
+}
+
+func TestFailedDeleteCanBeRetriedWithoutRevivingOldRun(t *testing.T) {
+	s := &singleFailureTaskDeleteStorage{Storage: NewMemStorage()}
+	m := &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()
+	d.cfg.MaxRunning = 0
+	id, err := d.CreateDirect(&base.Request{URL: "generation://failed-delete"}, &base.Options{Path: t.TempDir()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	s.fail.Store(true)
+	if err := d.Delete(&TaskFilter{IDs: []string{id}}, false); err == nil {
+		t.Fatal("expected storage deletion failure")
+	}
+	task := d.GetTask(id)
+	if task == nil || d.taskStatus(task) != base.DownloadStatusWait {
+		t.Fatal("failed deletion did not restore queued task")
+	}
+	d.cfg.MaxRunning = 1
+	if err := d.Continue(&TaskFilter{IDs: []string{id}}); err != nil {
+		t.Fatal(err)
+	}
+	deadline := time.Now().Add(time.Second)
+	for m.starts.Load() < 1 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	if m.starts.Load() != 1 {
+		t.Fatal("failed deletion could not be retried")
+	}
+}

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")
+			}
+		})
+	}
+}

pkg/download/downloader.go

@@ -515,7 +515,36 @@ 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
+		}
+	}
+	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)
@@ -1227,7 +1256,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
@@ -1419,7 +1448,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...)
@@ -1525,23 +1554,55 @@ 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++
+	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 {
+			// 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)
+		}
+	}()
 	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 {
@@ -1861,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 {
@@ -1880,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.storage.Put(bucketTask, task.ID, task.clone())
+					snapshot = task.clone()
+				}
+				task.statusLock.Unlock()
+				task.lock.Unlock()
+				if current {
+					d.putTaskRecord(task, snapshot)
 				}
 			}()
 		}
@@ -1893,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 {
@@ -1904,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
@@ -1915,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
 				}
@@ -1928,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 {
@@ -1937,7 +2030,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()
@@ -2042,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
@@ -2080,7 +2181,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)
 		}
 	}
@@ -2134,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
 }
@@ -2145,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 {
@@ -2208,7 +2316,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
@@ -2362,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()
@@ -2375,7 +2493,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
 		}
@@ -2400,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()
@@ -2432,6 +2551,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
 		}
 
@@ -2456,8 +2578,14 @@ 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
 		}
@@ -2466,10 +2594,13 @@ func (d *Downloader) doStart(task *Task) (err error) {
 		task.downloadSampledAt = time.Time{}
 		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
+		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.
@@ -2491,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)
 		}
 	}()
 
@@ -2578,7 +2709,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
 		}
 	}
@@ -2591,7 +2722,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 {
@@ -2600,7 +2731,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 {
@@ -2613,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.Status != base.DownloadStatusRunning {
+	if task.lifetime.deleted.Load() || task.Status != base.DownloadStatusRunning || task.Meta.Res == nil {
 		return false
 	}
 	task.Progress.Used = task.timer.Used()
@@ -2648,7 +2786,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)
@@ -2710,7 +2848,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
@@ -2739,7 +2877,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
 	}
@@ -2755,14 +2893,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)
@@ -2851,7 +2989,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) {
@@ -2871,7 +3009,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)
 
@@ -3015,13 +3153,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 {
@@ -3061,12 +3199,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{}

pkg/download/model.go

@@ -3,6 +3,7 @@ package download
 import (
 	"encoding/json"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/GopeedLab/gopeed/internal/controller"
@@ -43,6 +44,16 @@ type CreateTaskBatchResult struct {
 	ParentID string   `json:"parentId,omitempty"`
 }
 
+// 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)
@@ -87,6 +98,7 @@ type Task struct {
 	blobRefLock       *sync.Mutex
 	blobURL           string
 	runGeneration     uint64
+	lifetime          *taskLifetime
 	speedArr          []speedSample
 	uploadSpeedArr    []speedSample
 	downloadSampledAt time.Time
@@ -126,6 +138,7 @@ func NewTask() *Task {
 	}
 	return &Task{
 		ID:        id,
+		lifetime:  &taskLifetime{},
 		Status:    base.DownloadStatusReady,
 		CreatedAt: time.Now(),
 		UpdatedAt: time.Now(),