Merge main into fix/deleted-task-lifecycle-1610

pkg/download/crash_log_test.go

@@ -0,0 +1,26 @@
+package download
+
+import (
+	"os"
+	"path/filepath"
+	"runtime/debug"
+	"testing"
+)
+
+func TestLogPanicPreservesPreviousCrash(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "crash.log")
+	const previous = "previous crash traceback\n"
+	if err := os.WriteFile(path, []byte(previous), 0600); err != nil {
+		t.Fatal(err)
+	}
+	logPanic(dir)
+	t.Cleanup(func() { _ = debug.SetCrashOutput(nil, debug.CrashOptions{}) })
+	data, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(data) != previous {
+		t.Fatalf("previous crash was overwritten: %q", data)
+	}
+}

pkg/download/downloader.go

@@ -345,7 +345,12 @@ func (d *Downloader) Setup() error {
 						}
 
 						current := task.fetcher.Progress().TotalDownloaded()
-						delta := current - task.Progress.Downloaded
+						delta := max(0, current-task.Progress.Downloaded)
+						if counter, ok := task.fetcher.(fetcher.DownloadCounter); ok && task.Status == base.DownloadStatusRunning {
+							received := counter.ReceivedBytes()
+							delta = received - task.receivedBytes
+							task.receivedBytes = received
+						}
 						if task.Meta != nil && task.Meta.Req != nil {
 							if producer := d.blob.Production(task.Meta.Req.URL); producer != nil {
 								downloaded, received := producer.Progress()
@@ -353,19 +358,20 @@ func (d *Downloader) Setup() error {
 									task.producer = producer
 									task.producerReceived = 0
 									task.speedArr = nil
+									task.downloadSampledAt = time.Time{}
 								}
 								current = downloaded
 								delta = max(0, received-task.producerReceived)
 								task.producerReceived = received
 							}
 						}
-						tick := float64(d.cfg.RefreshInterval) / 1000
 						downloadDataChanged := false
 						uploading := task.Uploading
 						if task.Status == base.DownloadStatusRunning {
 							downloadDataChanged = current != task.Progress.Downloaded
 							task.Progress.Used = task.timer.Used()
-							task.Progress.Speed = task.updateSpeed(max(0, delta), tick)
+							elapsed := sampleSeconds(&task.downloadSampledAt, time.Now())
+							task.Progress.Speed = task.updateSpeed(delta, elapsed)
 							task.Progress.Downloaded = current
 						}
 
@@ -374,7 +380,8 @@ func (d *Downloader) Setup() error {
 							uploader := task.fetcher.(fetcher.Uploader)
 							currentUploaded := uploader.UploadedBytes()
 							uploadDataChanged = currentUploaded != task.Progress.Uploaded
-							task.Progress.UploadSpeed = task.updateUploadSpeed(currentUploaded-task.Progress.Uploaded, tick)
+							elapsed := sampleSeconds(&task.uploadSampledAt, time.Now())
+							task.Progress.UploadSpeed = task.updateUploadSpeed(currentUploaded-task.Progress.Uploaded, elapsed)
 							task.Progress.Uploaded = currentUploaded
 						}
 						task.statusLock.Unlock()
@@ -2441,7 +2448,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
 		}
@@ -2498,6 +2507,8 @@ func (d *Downloader) doStart(task *Task) (err error) {
 			return nil
 		}
 		task.Progress.Speed = 0
+		task.speedArr = nil
+		task.downloadSampledAt = time.Time{}
 		task.timer.Start()
 		task.statusLock.Unlock()
 		startErr := task.fetcher.Start()
@@ -2697,11 +2708,12 @@ func logPanic(logDir string) {
 	if err := util.CreateDirIfNotExist(logDir); err != nil {
 		return
 	}
-	f, err := os.Create(filepath.Join(logDir, "crash.log"))
+	f, err := os.OpenFile(filepath.Join(logDir, "crash.log"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600)
 	if err != nil {
 		return
 	}
-	debug.SetCrashOutput(f, debug.CrashOptions{})
+	defer f.Close() // SetCrashOutput duplicates the descriptor.
+	_ = debug.SetCrashOutput(f, debug.CrashOptions{})
 }
 
 func (d *Downloader) assignFetcherManager(task *Task) error {
@@ -3111,8 +3123,10 @@ func initTask(task *Task) {
 	task.lock = &sync.Mutex{}
 	task.statsLock = &sync.Mutex{}
 	task.blobRefLock = &sync.Mutex{}
-	task.speedArr = make([]int64, 0)
-	task.uploadSpeedArr = make([]int64, 0)
+	task.speedArr = nil
+	task.uploadSpeedArr = nil
+	task.downloadSampledAt = time.Time{}
+	task.uploadSampledAt = time.Time{}
 }
 
 func (d *Downloader) releaseBlobTask(task *Task) {

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/engine/ffmpeg/README.md

@@ -174,3 +174,12 @@ compilation memory, host buffers and JS allocations are additional to the
 linear-memory limit. Playback support depends on the selected codecs, not just
 the `.mp4` suffix. Platform execution and performance must be tested on target
 devices, especially mobile/32-bit devices.
+
+On 64-bit Windows, instance linear memory uses a reserved virtual address range with
+pages committed as it grows. Module completion, errors and cancellation release
+this range directly through `VirtualFree`, rather than waiting for Go's garbage
+collector to reclaim large buffers. Growing an instance preserves its address
+without copying the previous buffer. The 512 MiB limit is unchanged; reservation
+does not commit that entire amount. Shared compilation remains resident after
+the first merge, so process memory is not expected to return exactly to the
+pre-FFmpeg baseline. This does not diagnose unrelated idle memory growth.

pkg/download/engine/ffmpeg/memory.go

@@ -0,0 +1,32 @@
+package ffmpeg
+
+import (
+	"context"
+	"sync"
+
+	"github.com/tetratelabs/wazero/experimental"
+)
+
+// The runtime does not always invoke Free when a module's start function traps.
+// Keep host ownership until execution returns, including failed instantiation.
+func withOwnedMemory(ctx context.Context, allocator experimental.MemoryAllocator) (context.Context, func()) {
+	var memories []*ownedLinearMemory
+	ctx = experimental.WithMemoryAllocator(ctx, experimental.MemoryAllocatorFunc(
+		func(cap, max uint64) experimental.LinearMemory {
+			memory := &ownedLinearMemory{LinearMemory: allocator.Allocate(cap, max)}
+			memories = append(memories, memory)
+			return memory
+		}))
+	return ctx, func() {
+		for _, memory := range memories {
+			memory.Free()
+		}
+	}
+}
+
+type ownedLinearMemory struct {
+	experimental.LinearMemory
+	once sync.Once
+}
+
+func (m *ownedLinearMemory) Free() { m.once.Do(m.LinearMemory.Free) }

pkg/download/engine/ffmpeg/memory_other.go

@@ -0,0 +1,7 @@
+//go:build !windows || (!amd64 && !arm64)
+
+package ffmpeg
+
+import "context"
+
+func withLinearMemory(ctx context.Context) (context.Context, func()) { return ctx, func() {} }

pkg/download/engine/ffmpeg/memory_test.go

@@ -0,0 +1,88 @@
+package ffmpeg
+
+import (
+	"context"
+	"encoding/hex"
+	"testing"
+	"time"
+
+	"github.com/tetratelabs/wazero"
+	"github.com/tetratelabs/wazero/experimental"
+)
+
+// Verify the pinned runtime's allocator lifecycle on every development host:
+// host-owned buffers must not be released before execution ends, or retained
+// after failed instantiation, traps or cancellation.
+func TestLinearMemoryModuleLifecycle(t *testing.T) {
+	testLinearMemoryModuleLifecycle(t, func(_, _ uint64) experimental.LinearMemory {
+		return &testLinearMemory{}
+	})
+}
+
+type testLinearMemory struct{ buffer []byte }
+
+func (m *testLinearMemory) Reallocate(size uint64) []byte {
+	m.buffer = append(m.buffer, make([]byte, int(size)-len(m.buffer))...)
+	return m.buffer
+}
+func (m *testLinearMemory) Free() { m.buffer = nil }
+
+type trackedLinearMemory struct {
+	experimental.LinearMemory
+	freeCount int
+}
+
+func (m *trackedLinearMemory) Free() {
+	m.freeCount++
+	m.LinearMemory.Free()
+}
+
+func testLinearMemoryModuleLifecycle(t *testing.T, allocate experimental.MemoryAllocatorFunc) {
+	t.Helper()
+	for _, tt := range []struct {
+		name string
+		// A module with one memory (min=1/max=2 pages), optionally a start
+		// function that traps or loops until its context is cancelled.
+		wasm    string
+		wantErr bool
+	}{
+		{"close", "0061736d01000000050401010102", false},
+		{"trap", "0061736d01000000010401600000030201000504010101020801000a05010300000b", true},
+		{"cancel", "0061736d01000000010401600000030201000504010101020801000a0901070003400c000b0b", true},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
+			var memory *trackedLinearMemory
+			ctx, release := withOwnedMemory(ctx, experimental.MemoryAllocatorFunc(
+				func(_, max uint64) experimental.LinearMemory {
+					memory = &trackedLinearMemory{LinearMemory: allocate(0, max)}
+					return memory
+				}))
+			defer release()
+			rt := wazero.NewRuntimeWithConfig(ctx, wazero.NewRuntimeConfig().WithCloseOnContextDone(true))
+			defer rt.Close(context.Background())
+			binary, err := hex.DecodeString(tt.wasm)
+			if err != nil {
+				t.Fatal(err)
+			}
+			module, err := rt.Instantiate(ctx, binary)
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("instantiate: %v", err)
+			}
+			if !tt.wantErr {
+				if memory == nil || memory.freeCount != 0 {
+					t.Fatal("module did not allocate native memory")
+				}
+				if err := module.Close(ctx); err != nil {
+					t.Fatal(err)
+				}
+			}
+			release()
+			release()
+			if memory == nil || memory.freeCount != 1 {
+				t.Fatal("host did not release memory exactly once")
+			}
+		})
+	}
+}

pkg/download/engine/ffmpeg/memory_windows.go

@@ -0,0 +1,70 @@
+//go:build windows && (amd64 || arm64)
+
+package ffmpeg
+
+import (
+	"context"
+	"fmt"
+	"unsafe"
+
+	"github.com/tetratelabs/wazero/experimental"
+	"golang.org/x/sys/windows"
+)
+
+// Keep large, short-lived WASM buffers out of the Go heap. Closing a module
+// releases its committed pages immediately, without waiting for GC/scavenging.
+func withLinearMemory(ctx context.Context) (context.Context, func()) {
+	return withOwnedMemory(ctx, experimental.MemoryAllocatorFunc(
+		func(_, max uint64) experimental.LinearMemory {
+			return &windowsLinearMemory{max: max}
+		}))
+}
+
+// Reserve the address range once, committing zero-filled pages on demand. This
+// avoids both copying the previous buffer and temporarily retaining two large
+// Go allocations when memory.grow runs. The backing address never changes.
+type windowsLinearMemory struct {
+	address uintptr
+	max     uint64
+	size    uint64
+	freed   bool
+}
+
+func (m *windowsLinearMemory) Reallocate(size uint64) []byte {
+	if m.freed || size > m.max || m.max > uint64(^uint(0)>>1) {
+		return nil
+	}
+	if size == 0 {
+		return []byte{}
+	}
+	if m.address == 0 {
+		address, err := windows.VirtualAlloc(0, uintptr(m.max), windows.MEM_RESERVE, windows.PAGE_READWRITE)
+		if err != nil {
+			panic(fmt.Errorf("ffmpeg: reserve WASM memory: %w", err))
+		}
+		m.address = address
+	}
+	if size > m.size {
+		_, err := windows.VirtualAlloc(m.address, uintptr(size), windows.MEM_COMMIT, windows.PAGE_READWRITE)
+		if err != nil {
+			if m.size == 0 {
+				// Instantiation cannot proceed without the minimum memory. Wazero
+				// has not taken ownership yet, so release the reservation here.
+				m.Free()
+				panic(fmt.Errorf("ffmpeg: commit WASM memory: %w", err))
+			}
+			return nil // memory.grow reports failure to the guest.
+		}
+		m.size = size
+	}
+	return unsafe.Slice((*byte)(unsafe.Pointer(m.address)), int(size))
+}
+
+func (m *windowsLinearMemory) Free() {
+	if m.address != 0 {
+		_ = windows.VirtualFree(m.address, 0, windows.MEM_RELEASE)
+		m.address = 0
+	}
+	m.size = 0
+	m.freed = true
+}

pkg/download/engine/ffmpeg/memory_windows_test.go

@@ -0,0 +1,54 @@
+//go:build windows && (amd64 || arm64)
+
+package ffmpeg
+
+import (
+	"testing"
+	"unsafe"
+
+	"github.com/tetratelabs/wazero/experimental"
+	"golang.org/x/sys/windows"
+)
+
+func TestWindowsLinearMemoryGrowthAndRelease(t *testing.T) {
+	const page = 65536
+	m := &windowsLinearMemory{max: 4 * page}
+	defer m.Free()
+	first := m.Reallocate(page)
+	if len(first) != page {
+		t.Fatalf("initial allocation: %d", len(first))
+	}
+	first[0], first[page-1] = 17, 29
+	address := m.address
+	grown := m.Reallocate(3 * page)
+	if unsafe.SliceData(first) != unsafe.SliceData(grown) || grown[0] != 17 || grown[page-1] != 29 {
+		t.Fatal("growing memory moved or lost existing bytes")
+	}
+	for _, b := range grown[page:] {
+		if b != 0 {
+			t.Fatal("newly committed memory is not zero-filled")
+		}
+	}
+	if m.Reallocate(5*page) != nil || m.address != address || m.size != 3*page {
+		t.Fatal("failed growth changed the existing allocation")
+	}
+	m.Free()
+	m.Free()
+	if m.address != 0 || m.Reallocate(page) != nil {
+		t.Fatal("freed memory can still be used")
+	}
+	var info windows.MemoryBasicInformation
+	if err := windows.VirtualQuery(address, &info, unsafe.Sizeof(info)); err != nil {
+		t.Fatal(err)
+	}
+	const memFree = 0x10000 // MEMORY_BASIC_INFORMATION.State: MEM_FREE.
+	if info.State != memFree {
+		t.Fatalf("pages remain reserved after Free: state=%x", info.State)
+	}
+}
+
+func TestWindowsLinearMemoryModuleLifecycle(t *testing.T) {
+	testLinearMemoryModuleLifecycle(t, func(_, max uint64) experimental.LinearMemory {
+		return &windowsLinearMemory{max: max}
+	})
+}

pkg/download/engine/ffmpeg/runtime.go

@@ -151,7 +151,9 @@ func RunAcquired(ctx context.Context, video, audio Input, out io.Writer, format
 	files := &inputFS{files: map[string]Input{"video": video, "audio": audio}}
 	fsconfig := wazero.NewFSConfig().(sysfs.FSConfig).WithSysFSMount(files, "/inputs")
 	logs := &tailLog{}
-	rc, err := wasm.Run(ctx, shared.runtime, shared.module, wasm.Args{
+	runCtx, freeMemory := withLinearMemory(ctx)
+	defer freeMemory()
+	rc, err := wasm.Run(runCtx, shared.runtime, shared.module, wasm.Args{
 		Name: "ffmpeg", Args: args, Stdout: out, Stderr: logs,
 		Config: func(c wazero.ModuleConfig) wazero.ModuleConfig { return c.WithFSConfig(fsconfig) },
 	})

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

@@ -84,21 +84,24 @@ type Task struct {
 	producer         production.Source
 	producerReceived int64
 
-	fetcherManager fetcher.FetcherManager
-	fetcher        fetcher.Fetcher
-	timer          *util.Timer
-	statusLock     *sync.Mutex
-	lock           *sync.Mutex
-	statsLock      *sync.Mutex
-	statsLoaded    bool
-	stats          *fetcher.Stats
-	statsSnapshot  json.RawMessage
-	blobRefLock    *sync.Mutex
-	blobURL        string
-	runGeneration  uint64
-	lifetime       *taskLifetime
-	speedArr       []int64
-	uploadSpeedArr []int64
+	fetcherManager    fetcher.FetcherManager
+	fetcher           fetcher.Fetcher
+	timer             *util.Timer
+	statusLock        *sync.Mutex
+	lock              *sync.Mutex
+	statsLock         *sync.Mutex
+	statsLoaded       bool
+	stats             *fetcher.Stats
+	statsSnapshot     json.RawMessage
+	blobRefLock       *sync.Mutex
+	blobURL           string
+	runGeneration     uint64
+	lifetime          *taskLifetime
+	speedArr          []speedSample
+	uploadSpeedArr    []speedSample
+	downloadSampledAt time.Time
+	uploadSampledAt   time.Time
+	receivedBytes     int64
 }
 
 // TaskRuntimeStatus is the lightweight, frequently refreshed view of a task.
@@ -206,7 +209,21 @@ func (t *Task) updateUploadSpeed(downloaded int64, usedTime float64) int64 {
 	return calcSpeed(&t.uploadSpeedArr, downloaded, usedTime)
 }
 
-func calcSpeed(speedArr *[]int64, downloaded int64, usedTime float64) int64 {
+type speedSample struct {
+	bytes   int64
+	seconds float64
+}
+
+func sampleSeconds(lastSample *time.Time, now time.Time) float64 {
+	previous := *lastSample
+	*lastSample = now
+	if previous.IsZero() {
+		return 0
+	}
+	return now.Sub(previous).Seconds()
+}
+
+func calcSpeed(speedArr *[]speedSample, downloaded int64, usedTime float64) int64 {
 	if usedTime <= 0 {
 		return 0
 	}
@@ -215,18 +232,18 @@ func calcSpeed(speedArr *[]int64, downloaded int64, usedTime float64) int64 {
 		return 0
 	}
 
-	*speedArr = append(*speedArr, downloaded)
-	// Record last 5 seconds of download speed to calculate the average speed
-	if len(*speedArr) > int(5.0/usedTime) {
-		*speedArr = (*speedArr)[1:]
-	}
-
+	*speedArr = append(*speedArr, speedSample{bytes: downloaded, seconds: usedTime})
 	var total int64
-	for _, v := range *speedArr {
-		total += v
+	var elapsed float64
+	first := len(*speedArr)
+	for first > 0 && elapsed < 5 {
+		first--
+		sample := (*speedArr)[first]
+		total += sample.bytes
+		elapsed += sample.seconds
 	}
-
-	return int64(float64(total) / float64(len(*speedArr)) / usedTime)
+	*speedArr = (*speedArr)[first:]
+	return int64(float64(total) / elapsed)
 }
 
 type TaskFilter struct {
@@ -243,12 +260,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 +291,9 @@ func (cfg *DownloaderConfig) Init() *DownloaderConfig {
 			new(ed2k.FetcherManager),
 		}
 	}
+	if cfg.ExtensionEventTimeout <= 0 {
+		cfg.ExtensionEventTimeout = time.Minute
+	}
 	if cfg.RefreshInterval == 0 {
 		cfg.RefreshInterval = 350
 	}

pkg/download/model_test.go

@@ -1,9 +1,32 @@
 package download
 
-import "testing"
+import (
+	"testing"
+	"time"
+)
+
+func TestTransferSpeedWithUnevenSamples(t *testing.T) {
+	var task Task
+	for _, sample := range []struct {
+		bytes   int64
+		seconds float64
+	}{
+		{1000, 1},
+		{2000, 2},
+		{4000, 4},
+		{6000, 6},
+	} {
+		if got := task.updateSpeed(sample.bytes, sample.seconds); got != 1000 {
+			t.Errorf("download speed after %v seconds = %d, want 1000", sample.seconds, got)
+		}
+		if got := task.updateUploadSpeed(sample.bytes, sample.seconds); got != 1000 {
+			t.Errorf("upload speed after %v seconds = %d, want 1000", sample.seconds, got)
+		}
+	}
+}
 
 func TestCalcSpeedResetOnRollback(t *testing.T) {
-	speedArr := []int64{1024, 2048, 4096}
+	speedArr := []speedSample{{1024, 1}, {2048, 1}, {4096, 1}}
 
 	if got := calcSpeed(&speedArr, -512, 1); got != 0 {
 		t.Fatalf("calcSpeed() = %d, want 0 after rollback", got)
@@ -16,3 +39,17 @@ func TestCalcSpeedResetOnRollback(t *testing.T) {
 		t.Fatalf("calcSpeed() = %d, want 1024 after reset", got)
 	}
 }
+
+func TestSampleSecondsUsesThePreviousSuccessfulSample(t *testing.T) {
+	start := time.Now()
+	var last time.Time
+	if got := sampleSeconds(&last, start); got != 0 {
+		t.Fatalf("first sample elapsed = %v, want 0", got)
+	}
+	if got := sampleSeconds(&last, start.Add(1500*time.Millisecond)); got != 1.5 {
+		t.Fatalf("elapsed after a delayed sample = %v, want 1.5", got)
+	}
+	if got := sampleSeconds(&last, start.Add(2*time.Second)); got != 0.5 {
+		t.Fatalf("elapsed after the next sample = %v, want 0.5", got)
+	}
+}