From 54fc0f728987c1f6a96a5d1c9450e6b4e1cec30f Mon Sep 17 00:00:00 2001 From: Levi Date: Tue, 10 Nov 2026 09:47:59 +0800 Subject: [PATCH] test(download): cover deletion during preparation and late callbacks --- pkg/download/delete_lifecycle_test.go | 122 ++++++++++++++++++++++++++ pkg/download/downloader.go | 7 +- 2 files changed, 123 insertions(+), 6 deletions(-) diff --git a/pkg/download/delete_lifecycle_test.go b/pkg/download/delete_lifecycle_test.go index b1986b22..7786844f 100644 --- a/pkg/download/delete_lifecycle_test.go +++ b/pkg/download/delete_lifecycle_test.go @@ -4,6 +4,7 @@ import ( "errors" "fmt" "sync" + "sync/atomic" "testing" "time" @@ -135,3 +136,124 @@ func TestDeleteSerializesInFlightTaskCheckpoint(t *testing.T) { 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") + } +} diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index 17434c40..e530495c 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -1550,12 +1550,7 @@ func (d *Downloader) doDelete(task *Task, force bool) (err error) { 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() + // Keep failed deletions recoverable without validating the old run. task.lifetime.deleted.Store(false) } }()