fix(engine): allow JS resource cleanup after interrupting execution

pkg/download/delete_lifecycle_test.go

@@ -1,259 +0,0 @@
-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/downloader.go

@@ -508,31 +508,7 @@ 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)
@@ -1244,7 +1220,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.putTaskRecord(task, task.clone()); saveErr != nil {
+			if saveErr := d.storage.Put(bucketTask, task.ID, task.clone()); saveErr != nil {
 				d.Logger.Error().Err(saveErr).Msg("restore failed task deletion failed")
 				if firstErr == nil {
 					firstErr = saveErr
@@ -1436,7 +1412,7 @@ func (d *Downloader) captureTaskStats(task *Task, live, force bool) error {
 		return nil
 	}
 
-	if err := d.putTaskValue(task, bucketTaskStats, &storedTaskStats{Snapshot: snapshot}); err != nil {
+	if err := d.storage.Put(bucketTaskStats, task.ID, &storedTaskStats{Snapshot: snapshot}); err != nil {
 		return err
 	}
 	task.statsSnapshot = append(task.statsSnapshot[:0], snapshot...)
@@ -1542,18 +1518,6 @@ 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 without validating the old run.
-			task.lifetime.deleted.Store(false)
-		}
-	}()
 	defer d.releaseBlobTask(task)
 	err = func() error {
 		if err := d.storage.Delete(bucketTask, task.ID); err != nil {
@@ -1911,7 +1875,7 @@ func (d *Downloader) watch(task *Task) {
 				// Check if the task is deleted
 				if d.GetTask(task.ID) != nil {
 					task.Uploading = false
-					d.putTaskRecord(task, task.clone())
+					d.storage.Put(bucketTask, task.ID, task.clone())
 				}
 			}()
 		}
@@ -1966,7 +1930,7 @@ func (d *Downloader) watch(task *Task) {
 		task.statusLock.Lock()
 		snapshot := task.clone()
 		task.statusLock.Unlock()
-		d.putTaskRecord(task, snapshot)
+		d.storage.Put(bucketTask, task.ID, snapshot)
 		d.emit(EventKeyDone, task)
 		d.emit(EventKeyFinally, task, err)
 		d.notifyRunning()
@@ -2109,7 +2073,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.putTaskRecord(task, task.clone()); persistErr != nil {
+		if persistErr := d.storage.Put(bucketTask, task.ID, task.clone()); persistErr != nil {
 			d.Logger.Error().Stack().Err(persistErr).Msgf("persist failed task failed, task id: %s", task.ID)
 		}
 	}
@@ -2237,7 +2201,7 @@ func (d *Downloader) doCreate(f fetcher.Fetcher, opts *base.Options, groupID str
 				return
 			}
 		}
-		if err = d.putTaskRecord(task, task.clone()); err != nil {
+		if err = d.storage.Put(bucketTask, task.ID, task.clone()); err != nil {
 			return
 		}
 		taskId = task.ID
@@ -2404,7 +2368,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.lifetime.deleted.Load() || task.Status == base.DownloadStatusRunning || task.Status == base.DownloadStatusDone {
+		if task.Status == base.DownloadStatusRunning || task.Status == base.DownloadStatusDone {
 			isReturn = true
 			return
 		}
@@ -2441,7 +2405,9 @@ func (d *Downloader) doStart(task *Task) (err error) {
 				return err
 			}
 		}
-		d.triggerOnStart(task)
+		if err := d.triggerOnStart(task); err != nil {
+			return err
+		}
 		if !d.taskIsRunningGeneration(task, generation) {
 			return nil
 		}
@@ -2459,9 +2425,6 @@ 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
 		}
 
@@ -2486,24 +2449,16 @@ 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()
-		startErr := task.fetcher.Start()
-		task.lifetime.mu.Unlock()
-		if startErr != nil {
-			return startErr
+		if err := task.fetcher.Start(); err != nil {
+			return err
 		}
 		started = true
 		if err := d.captureTaskStats(task, true, true); err != nil {
@@ -2614,7 +2569,7 @@ func (d *Downloader) runPauseHandler(task *Task, generation uint64, pauseFetcher
 			return false, err
 		}
 	} else {
-		if err := d.putTaskRecord(task, task.clone()); err != nil {
+		if err := d.storage.Put(bucketTask, task.ID, task.clone()); err != nil {
 			return false, err
 		}
 	}
@@ -2627,7 +2582,7 @@ func (d *Downloader) taskIsRunningGeneration(task *Task, generation uint64) bool
 	}
 	task.statusLock.Lock()
 	defer task.statusLock.Unlock()
-	return !task.lifetime.deleted.Load() && task.Status == base.DownloadStatusRunning && task.runGeneration == generation
+	return task.Status == base.DownloadStatusRunning && task.runGeneration == generation
 }
 
 func (d *Downloader) taskIsPausedGeneration(task *Task, generation uint64) bool {
@@ -2636,7 +2591,7 @@ func (d *Downloader) taskIsPausedGeneration(task *Task, generation uint64) bool
 	}
 	task.statusLock.Lock()
 	defer task.statusLock.Unlock()
-	return !task.lifetime.deleted.Load() && (task.Status == base.DownloadStatusPause || task.Status == base.DownloadStatusWait) && task.runGeneration == generation
+	return (task.Status == base.DownloadStatusPause || task.Status == base.DownloadStatusWait) && task.runGeneration == generation
 }
 
 func (d *Downloader) taskStatus(task *Task) base.Status {
@@ -2656,7 +2611,7 @@ func (d *Downloader) markTaskDone(task *Task) bool {
 	defer task.lock.Unlock()
 	task.statusLock.Lock()
 	defer task.statusLock.Unlock()
-	if task.lifetime.deleted.Load() || task.Status != base.DownloadStatusRunning {
+	if task.Status != base.DownloadStatusRunning {
 		return false
 	}
 	task.Progress.Used = task.timer.Used()
@@ -2684,7 +2639,7 @@ func (d *Downloader) markTaskError(task *Task) bool {
 	}
 	task.statusLock.Lock()
 	defer task.statusLock.Unlock()
-	if task.lifetime.deleted.Load() || task.Status != base.DownloadStatusRunning {
+	if task.Status != base.DownloadStatusRunning {
 		return false
 	}
 	task.updateStatus(base.DownloadStatusError)
@@ -2745,7 +2700,7 @@ func (d *Downloader) enqueueSingleExtraction(task *Task, downloadFilePath string
 	// Set extraction status to queued
 	task.Progress.ExtractStatus = ExtractStatusQueued
 	d.emit(EventKeyProgress, task)
-	d.putTaskRecord(task, task.clone())
+	d.storage.Put(bucketTask, task.ID, task.clone())
 	d.Logger.Info().Msgf("extraction queued, task id: %s, job id: %s", task.ID, jobID)
 
 	// Create and enqueue the extraction job
@@ -2774,7 +2729,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.putTaskRecord(task, task.clone())
+		d.storage.Put(bucketTask, task.ID, task.clone())
 		d.Logger.Info().Msgf("multi-part archive waiting for other parts, task id: %s, missing: %v", task.ID, missingParts)
 		return
 	}
@@ -2790,14 +2745,14 @@ func (d *Downloader) enqueueMultiPartExtraction(task *Task, downloadFilePath str
 		task.Progress.ExtractStatus = ExtractStatusDone
 		task.Progress.ExtractProgress = 100
 		d.emit(EventKeyProgress, task)
-		d.putTaskRecord(task, task.clone())
+		d.storage.Put(bucketTask, task.ID, 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.putTaskRecord(task, task.clone())
+	d.storage.Put(bucketTask, task.ID, task.clone())
 
 	jobID := "multipart:" + fullBaseName
 	d.Logger.Info().Msgf("multi-part extraction queued, task id: %s, job id: %s", task.ID, jobID)
@@ -2886,7 +2841,7 @@ func (d *Downloader) performExtraction(task *Task, archivePath string, destDir s
 	task.Progress.ExtractStatus = ExtractStatusExtracting
 	task.Progress.ExtractProgress = 0
 	d.emit(EventKeyProgress, task)
-	d.putTaskRecord(task, task.clone())
+	d.storage.Put(bucketTask, task.ID, task.clone())
 
 	// Extract the archive
 	extractErr := extractArchive(archivePath, destDir, opts.ArchivePassword, func(extractedFiles int, totalFiles int, progress int) {
@@ -2906,7 +2861,7 @@ func (d *Downloader) performMultiPartExtraction(task *Task, firstPartPath string
 	task.Progress.ExtractStatus = ExtractStatusExtracting
 	task.Progress.ExtractProgress = 0
 	d.emit(EventKeyProgress, task)
-	d.putTaskRecord(task, task.clone())
+	d.storage.Put(bucketTask, task.ID, task.clone())
 
 	d.Logger.Info().Msgf("starting multi-part archive extraction, first part: %s, task id: %s", firstPartPath, task.ID)
 
@@ -3050,13 +3005,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.putTaskRecord(task, task.clone())
+		d.storage.Put(bucketTask, task.ID, 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.putTaskRecord(task, task.clone())
+		d.storage.Put(bucketTask, task.ID, task.clone())
 
 		// Delete archive files after successful extraction if enabled
 		if deleteAfterExtract {
@@ -3096,15 +3051,12 @@ func (d *Downloader) updateMultiPartTasksStatus(sourceTask *Task, extractErr err
 			task.Progress.ExtractStatus = status
 			task.Progress.ExtractProgress = progress
 			d.emit(EventKeyProgress, task)
-			d.putTaskRecord(task, task.clone())
+			d.storage.Put(bucketTask, task.ID, 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/engine/cancellation_test.go

@@ -0,0 +1,47 @@
+package engine
+
+import (
+	"context"
+	"errors"
+	"github.com/dop251/goja"
+	"testing"
+	"time"
+)
+
+func TestCloseUnblocksPendingPromise(t *testing.T) {
+	e := NewEngine(nil)
+	entered := make(chan struct{})
+	_, err := e.runOnLoop(func(r *goja.Runtime) (goja.Value, error) {
+		return goja.Undefined(), r.Set("entered", func() { close(entered) })
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	done := make(chan error, 1)
+	go func() { _, err := e.RunString(`entered(); new Promise(()=>{});`); done <- err }()
+	select {
+	case <-entered:
+	case <-time.After(time.Second):
+		t.Fatal("script did not enter")
+	}
+	e.Close()
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatal("closed execution returned success")
+		}
+	case <-time.After(300 * time.Millisecond):
+		t.Fatal("Engine.Close left the Promise waiter blocked")
+	}
+}
+
+func TestExecutionDeadlineInterruptsSynchronousScript(t *testing.T) {
+	e := NewEngine(nil)
+	defer e.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
+	defer cancel()
+	_, err := e.RunStringContext(ctx, `while (true) {}`)
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("got %v, want deadline exceeded", err)
+	}
+}

pkg/download/engine/engine.go

@@ -1,6 +1,7 @@
 package engine
 
 import (
+	"context"
 	_ "embed"
 	"errors"
 	"fmt"
@@ -26,6 +27,7 @@ var polyfillScript string
 
 type Engine struct {
 	loop *eventloop.EventLoop
+	done chan struct{}
 
 	closeOnce sync.Once
 	cleanupMu sync.Mutex
@@ -39,7 +41,13 @@ type JSFunction func(goja.FunctionCall) goja.Value
 // RunString executes the script and returns the go type value
 // if script result is promise, it will be resolved
 func (e *Engine) RunString(script string) (value any, err error) {
-	return e.runOnLoop(func(runtime *goja.Runtime) (goja.Value, error) {
+	return e.RunStringContext(context.Background(), script)
+}
+
+// RunStringContext bounds an execution, including awaited promises. Cancelling
+// an execution closes its engine so outstanding operations cannot outlive it.
+func (e *Engine) RunStringContext(ctx context.Context, script string) (any, error) {
+	return e.runOnLoopContext(ctx, func(runtime *goja.Runtime) (goja.Value, error) {
 		return runtime.RunString(script)
 	})
 }
@@ -47,7 +55,11 @@ func (e *Engine) RunString(script string) (value any, err error) {
 // CallFunction calls the function and returns the go type value
 // if function result is promise, it will be resolved
 func (e *Engine) CallFunction(fn any, args ...any) (value any, err error) {
-	return e.runOnLoop(func(runtime *goja.Runtime) (goja.Value, error) {
+	return e.CallFunctionContext(context.Background(), fn, args...)
+}
+
+func (e *Engine) CallFunctionContext(ctx context.Context, fn any, args ...any) (value any, err error) {
+	return e.runOnLoopContext(ctx, func(runtime *goja.Runtime) (goja.Value, error) {
 		var jsArgs []goja.Value
 		for _, arg := range args {
 			jsArgs = append(jsArgs, runtime.ToValue(arg))
@@ -69,6 +81,20 @@ func (e *Engine) CallFunction(fn any, args ...any) (value any, err error) {
 }
 
 func (e *Engine) runOnLoop(fn func(runtime *goja.Runtime) (goja.Value, error)) (any, error) {
+	return e.runOnLoopContext(context.Background(), fn)
+}
+
+var ErrClosed = errors.New("engine closed")
+
+func (e *Engine) runOnLoopContext(ctx context.Context, fn func(runtime *goja.Runtime) (goja.Value, error)) (any, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	select {
+	case <-e.done:
+		return nil, ErrClosed
+	default:
+	}
 	type result struct {
 		value any
 		err   error
@@ -132,15 +158,31 @@ func (e *Engine) runOnLoop(fn func(runtime *goja.Runtime) (goja.Value, error)) (
 	if !ok {
 		return nil, errors.New("engine loop terminated")
 	}
-	res := <-ch
-	if res.err != nil {
-		return nil, res.err
+	select {
+	case res := <-ch:
+		return res.value, res.err
+	case <-ctx.Done():
+		// Interrupt synchronous JS too; a select alone only stops Promise waiting.
+		e.Close()
+		return nil, ctx.Err()
+	case <-e.done:
+		return nil, ErrClosed
 	}
-	return res.value, nil
 }
 
 func (e *Engine) Close() {
 	e.closeOnce.Do(func() {
+		close(e.done)
+		e.Runtime.Interrupt(ErrClosed)
+		// Wait for interrupted JS to leave the loop, then allow cleanup hooks
+		// to execute their JS cancellation callbacks before terminating it.
+		cleared := make(chan struct{})
+		if e.loop.RunOnLoop(func(runtime *goja.Runtime) {
+			runtime.ClearInterrupt()
+			close(cleared)
+		}) {
+			<-cleared
+		}
 		e.cleanupMu.Lock()
 		cleanups := e.cleanups
 		e.cleanups = nil
@@ -177,6 +219,7 @@ func NewEngine(cfg *Config) *Engine {
 	loop := eventloop.NewEventLoop()
 	engine := &Engine{
 		loop: loop,
+		done: make(chan struct{}),
 	}
 	loop.Start()
 	done := make(chan struct{})

pkg/download/extension.go

@@ -3,6 +3,7 @@ package download
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/url"
@@ -327,8 +328,8 @@ func (d *Downloader) triggerOnResolve(req *base.Request) (res *base.Resource, er
 	return
 }
 
-func (d *Downloader) triggerOnStart(task *Task) {
-	doTrigger(d,
+func (d *Downloader) triggerOnStart(task *Task) error {
+	return doTrigger(d,
 		EventOnStart,
 		task.Meta.Req,
 		&OnStartContext{
@@ -413,6 +414,13 @@ func doTrigger[T any](d *Downloader, event ActivationEvent, req *base.Request, c
 					}
 					engine, session := d.newExtensionEngine()
 					defer session.CloseIfIdle()
+					eventCtx, cancelEvent := context.WithTimeout(d.backgroundCtx, d.cfg.ExtensionEventTimeout)
+					defer cancelEvent()
+					defer func() {
+						if eventCtx.Err() != nil {
+							session.Close()
+						}
+					}()
 					gopeed.Runtime = &InstanceRuntime{
 						WebView: d.newExtensionWebViewRuntime(session, ext.buildIdentity()),
 					}
@@ -421,13 +429,13 @@ func doTrigger[T any](d *Downloader, event ActivationEvent, req *base.Request, c
 						gopeed.Logger.logger.Error().Err(err).Msgf("[%s] engine inject failed", ext.buildIdentity())
 						return
 					}
-					_, err = engine.RunString(string(scriptBuf))
+					_, err = engine.RunStringContext(eventCtx, string(scriptBuf))
 					if err != nil {
 						gopeed.Logger.logger.Error().Err(err).Msgf("[%s] run script failed", ext.buildIdentity())
 						return
 					}
 					if fn, ok := gopeed.Events[event]; ok {
-						_, err = engine.CallFunction(fn, ctx)
+						_, err = engine.CallFunctionContext(eventCtx, fn, ctx)
 						if err != nil {
 							gopeed.Logger.logger.Error().Err(err).Msgf("[%s] call function failed: %s", ext.buildIdentity(), event)
 							return
@@ -437,6 +445,9 @@ func doTrigger[T any](d *Downloader, event ActivationEvent, req *base.Request, c
 						}
 					}
 				}()
+				if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+					return err
+				}
 			}
 		}
 	}

pkg/download/extension_cancellation_test.go

@@ -0,0 +1,149 @@
+package download
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/GopeedLab/gopeed/internal/fetcher"
+	"github.com/GopeedLab/gopeed/pkg/base"
+)
+
+func installCancellationExtension(t *testing.T, d *Downloader, event, script string) {
+	t.Helper()
+	dir := t.TempDir()
+	manifest := fmt.Sprintf(`{"name":"cancel-%s","author":"test","title":"Cancel","version":"1.0.0","scripts":[{"event":%q,"match":{"urls":["*://example.com/*"]},"entry":"index.js"}]}`, event, event)
+	for name, body := range map[string]string{"manifest.json": manifest, "index.js": script} {
+		if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0600); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := d.InstallExtensionByFolder(dir, true); err != nil {
+		t.Fatal(err)
+	}
+}
+func newCancellationDownloader(t *testing.T) *Downloader {
+	t.Helper()
+	d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir(), FetchManagers: []fetcher.FetcherManager{&generationTestManager{holdOpen: true}}})
+	if err := d.Setup(); err != nil {
+		t.Fatal(err)
+	}
+	return d
+}
+func TestResolveExtensionStopsOnShutdown(t *testing.T) {
+	entered, release := make(chan struct{}), make(chan struct{})
+	var once sync.Once
+	defer once.Do(func() { close(release) })
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		close(entered)
+		select {
+		case <-release:
+		case <-r.Context().Done():
+		}
+		w.Write([]byte("ok"))
+	}))
+	defer server.Close()
+	d := newCancellationDownloader(t)
+	defer d.Close()
+	installCancellationExtension(t, d, "onResolve", fmt.Sprintf(`gopeed.events.onResolve(async ctx=>{await fetch(%q);});`, server.URL))
+	done := make(chan error, 1)
+	go func() { _, err := d.Resolve(&base.Request{URL: "https://example.com/cancel"}); done <- err }()
+	select {
+	case <-entered:
+	case <-time.After(2 * time.Second):
+		t.Fatal("handler did not enter")
+	}
+	if err := d.Close(); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case err := <-done:
+		if !errors.Is(err, context.Canceled) {
+			t.Errorf("got %v, want cancellation", err)
+		}
+	case <-time.After(300 * time.Millisecond):
+		once.Do(func() { close(release) })
+		t.Fatal("shutdown did not cancel extension Resolve")
+	}
+}
+func TestOnErrorCancellationReleasesTaskAPIs(t *testing.T) {
+	entered, release := make(chan struct{}), make(chan struct{})
+	var once sync.Once
+	defer once.Do(func() { close(release) })
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		close(entered)
+		select {
+		case <-release:
+		case <-r.Context().Done():
+		}
+		w.Write([]byte("ok"))
+	}))
+	defer server.Close()
+	d := newCancellationDownloader(t)
+	defer d.Close()
+	parent, cancel := context.WithCancel(d.backgroundCtx)
+	d.backgroundCtx = parent
+	defer cancel()
+	installCancellationExtension(t, d, "onError", fmt.Sprintf(`gopeed.events.onError(async ctx=>{await fetch(%q);});`, server.URL))
+	d.cfg.MaxRunning = 0
+	id, err := d.CreateDirect(&base.Request{URL: "generation://example.com/task"}, &base.Options{Path: t.TempDir()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	task := d.GetTask(id)
+	task.statusLock.Lock()
+	task.Status = base.DownloadStatusRunning
+	task.statusLock.Unlock()
+	handled := make(chan struct{})
+	go func() { d.doOnError(task, errors.New("source failed")); close(handled) }()
+	select {
+	case <-entered:
+	case <-time.After(2 * time.Second):
+		t.Fatal("onError did not enter")
+	}
+	status, stats := make(chan struct{}), make(chan struct{})
+	go func() { d.RuntimeStatus(id); close(status) }()
+	go func() { d.Stats(id); close(stats) }()
+	cancel()
+	for name, ch := range map[string]<-chan struct{}{"onError": handled, "RuntimeStatus": status, "Stats": stats} {
+		select {
+		case <-ch:
+		case <-time.After(300 * time.Millisecond):
+			once.Do(func() { close(release) })
+			<-handled
+			t.Fatalf("%s stayed blocked after cancellation", name)
+		}
+	}
+}
+
+func TestExtensionEventDeadlineRejectsUnsettledResolve(t *testing.T) {
+	d := newCancellationDownloader(t)
+	defer d.Close()
+	d.cfg.ExtensionEventTimeout = 30 * time.Millisecond
+	installCancellationExtension(t, d, "onResolve", `gopeed.events.onResolve(async ctx=>{await new Promise(()=>{});});`)
+	_, err := d.Resolve(&base.Request{URL: "https://example.com/timeout"})
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("got %v, want event deadline exceeded", err)
+	}
+}
+func TestExtensionStartDeadlinePreventsFetcherStart(t *testing.T) {
+	d := newCancellationDownloader(t)
+	defer d.Close()
+	d.cfg.ExtensionEventTimeout = 30 * time.Millisecond
+	installCancellationExtension(t, d, "onStart", `gopeed.events.onStart(async ctx=>{await new Promise(()=>{});});`)
+	id, err := d.CreateDirect(&base.Request{URL: "generation://example.com/timeout"}, &base.Options{Path: t.TempDir()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitForTaskStatus(t, d, id, base.DownloadStatusError, 2*time.Second)
+	if starts := d.cfg.FetchManagers[0].(*generationTestManager).starts.Load(); starts != 0 {
+		t.Fatalf("timed out onStart still started fetcher %d times", starts)
+	}
+}

pkg/download/extension_engine_session.go

@@ -59,6 +59,21 @@ func (s *engineSession) Release() {
 	}
 }
 
+// Close releases retained resources when the owning event is cancelled.
+func (s *engineSession) Close() {
+	s.mu.Lock()
+	if s.closed {
+		s.mu.Unlock()
+		return
+	}
+	s.closed = true
+	s.mu.Unlock()
+	s.runClosers()
+	if s.engine != nil {
+		s.engine.Close()
+	}
+}
+
 func (s *engineSession) CloseIfIdle() {
 	s.mu.Lock()
 	shouldClose := s.refs == 0 && !s.closed

pkg/download/model.go

@@ -3,7 +3,6 @@ package download
 import (
 	"encoding/json"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/GopeedLab/gopeed/internal/controller"
@@ -47,14 +46,6 @@ 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"`
@@ -96,7 +87,6 @@ type Task struct {
 	blobRefLock    *sync.Mutex
 	blobURL        string
 	runGeneration  uint64
-	lifetime       *taskLifetime
 	speedArr       []int64
 	uploadSpeedArr []int64
 }
@@ -133,7 +123,6 @@ func NewTask() *Task {
 	}
 	return &Task{
 		ID:        id,
-		lifetime:  &taskLifetime{},
 		Status:    base.DownloadStatusReady,
 		CreatedAt: time.Now(),
 		UpdatedAt: time.Now(),
@@ -243,12 +232,15 @@ type DownloaderConfig struct {
 	Controller    *controller.Controller
 	FetchManagers []fetcher.FetcherManager
 
-	RefreshInterval   int // RefreshInterval time duration to refresh task progress(ms)
-	Storage           Storage
-	StorageDir        string
-	TempDir           string
-	WhiteDownloadDirs []string
-	WebViewProvider   enginewebview.Provider
+	// ExtensionEventTimeout bounds one script load and its awaited event handler.
+	// Zero uses one minute. Embedders may increase it for long-running handlers.
+	ExtensionEventTimeout time.Duration
+	RefreshInterval       int // RefreshInterval time duration to refresh task progress(ms)
+	Storage               Storage
+	StorageDir            string
+	TempDir               string
+	WhiteDownloadDirs     []string
+	WebViewProvider       enginewebview.Provider
 
 	ProductionMode bool
 
@@ -271,6 +263,9 @@ func (cfg *DownloaderConfig) Init() *DownloaderConfig {
 			new(ed2k.FetcherManager),
 		}
 	}
+	if cfg.ExtensionEventTimeout <= 0 {
+		cfg.ExtensionEventTimeout = time.Minute
+	}
 	if cfg.RefreshInterval == 0 {
 		cfg.RefreshInterval = 350
 	}