fix(windows): release FFmpeg memory and capture native crashes

Co-authored-by: xunguangzlj-cloud <[email protected]>

.github/workflows/test.yml

@@ -134,6 +134,9 @@ jobs:
         if: runner.os == 'macOS'
         with:
           flutter-version: ${{ env.FLUTTER_VERSION }}
+      - name: Test Windows FFmpeg memory and stream lifecycle
+        if: runner.os == 'Windows'
+        run: go test -v -count=1 -timeout 180s ./pkg/download/engine/ffmpeg ./pkg/download/engine/inject/stream ./pkg/download/engine
       - name: Install Linux WebView dependencies
         if: runner.os == 'Linux'
         run: |

internal/fetcher/fetcher.go

@@ -46,6 +46,10 @@ type Uploader interface {
 	WaitUpload() error
 }
 
+type DownloadCounter interface {
+	ReceivedBytes() int64
+}
+
 // FetcherMeta defines the meta information of a fetcher.
 type FetcherMeta struct {
 	Req  *base.Request  `json:"req"`

internal/protocol/bt/download_test.go

@@ -290,6 +290,82 @@ func TestBTDownloadSelectedFiles(t *testing.T) {
 	if f.Meta().Res.Size != 48*1024 {
 		t.Fatal("selected resource size is incorrect")
 	}
+	if downloaded := f.Progress().TotalDownloaded(); downloaded != 48*1024 || f.ReceivedBytes() <= downloaded {
+		t.Fatalf("selected progress = %d, received traffic = %d; want 48 KiB of progress and traffic across the shared piece", downloaded, f.ReceivedBytes())
+	}
+}
+
+func TestBTProgressOnlyCountsVerifiedFileBytes(t *testing.T) {
+	const fileSize = 56 * 1024
+	fixture := makeBTFixtureWithPieceSize(t, map[string][]byte{
+		"a.bin": btPayload(fileSize, "verified-a"),
+		"b.bin": btPayload(fileSize, "verified-b"),
+	}, 16*1024, false, true, 64*1024)
+	f := btDownloadFetcher(t, nil)
+	btResolve(t, f, fixture.Magnet)
+	dir := t.TempDir()
+	f.Meta().Opts = &base.Options{Path: dir}
+	btStart(t, f)
+	deadline := time.Now().Add(8 * time.Second)
+	for f.torrent.Files()[0].BytesCompleted() == 0 && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+	btPause(t, f)
+	if f.torrent.Files()[0].BytesCompleted() == 0 || f.torrent.Piece(0).State().Complete {
+		t.Fatal("expected some received bytes in a piece that is still incomplete")
+	}
+	if f.ReceivedBytes() == 0 {
+		t.Fatal("received traffic must be visible before a piece is verified")
+	}
+	if progress := f.Progress(); len(progress) != 2 || progress[0] != 0 {
+		t.Errorf("progress before the first file's piece is verified = %v, want zero for the first file", progress)
+	}
+	fixture.Limiter.SetLimit(rate.Inf)
+	btStart(t, f)
+	btAwaitDone(t, f)
+	fixture.AssertFiles(t, dir)
+	progress := f.Progress()
+	if len(progress) != 2 || progress[0] != fileSize || progress[1] != fileSize {
+		t.Fatalf("verified progress at completion = %v, want %d for each file", progress, fileSize)
+	}
+}
+
+func TestBTUploadedBytesSamplesLiveTraffic(t *testing.T) {
+	payload := btPayload(48*1024, "upload-counter")
+	fixture := newBTFixture(t, map[string][]byte{"": payload}, 0)
+	const saved = 1024
+	f := &Fetcher{
+		torrent:       fixture.Torrent,
+		data:          &fetcherData{SeedBytes: saved},
+		seedBaseBytes: saved,
+	}
+	f.torrentReady.Store(true)
+	f.torrentUpload.Store(true)
+
+	cfg := torrent.TestingConfig(t)
+	cfg.DataDir = t.TempDir()
+	cfg.DisableTrackers = true
+	cfg.DisableUTP = true
+	cfg.DisableIPv6 = true
+	peer, err := torrent.NewClient(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { peer.Close() })
+	download, _, err := peer.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(fixture.MetaInfo))
+	if err != nil {
+		t.Fatal(err)
+	}
+	download.AddPeers([]torrent.PeerInfo{{Addr: fixture.Seeder.ListenAddrs()[0]}})
+	download.DownloadAll()
+	select {
+	case <-download.Complete().On():
+	case <-time.After(10 * time.Second):
+		t.Fatal("loopback peer did not complete")
+	}
+	if got, want := f.UploadedBytes(), int64(saved+len(payload)); got != want {
+		t.Fatalf("uploaded bytes = %d, want %d including the saved bytes", got, want)
+	}
 }
 
 func TestBTPauseAndResume(t *testing.T) {

internal/protocol/bt/fetcher.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"sort"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -42,9 +43,10 @@ type Fetcher struct {
 	ctl    *controller.Controller
 	config *config
 
-	torrent *torrent.Torrent
+	torrent       *torrent.Torrent
-	meta    *fetcher.FetcherMeta
+	meta          *fetcher.FetcherMeta
-	data    *fetcherData
+	data          *fetcherData
+	seedBaseBytes int64
 
 	torrentReady    atomic.Bool
 	torrentUpload   atomic.Bool
@@ -463,14 +465,46 @@ func (f *Fetcher) Progress() fetcher.Progress {
 	if !f.torrentReady.Load() {
 		return append(fetcher.Progress(nil), f.data.Progress...)
 	}
+	pieceSize := f.torrent.Info().PieceLength
+	var offset int64
+	var verified []verifiedByteRange
+	for _, run := range f.torrent.PieceStateRuns() {
+		end := offset + int64(run.Length)*pieceSize
+		if pieceVerified(run.PieceState) {
+			verified = append(verified, verifiedByteRange{start: offset, end: end})
+		}
+		offset = end
+	}
+	files := f.torrent.Files()
 	for i := range f.data.Progress {
 		selectIndex := f.meta.Opts.SelectFiles[i]
-		file := f.torrent.Files()[selectIndex]
+		file := files[selectIndex]
-		f.data.Progress[i] = file.BytesCompleted()
+		f.data.Progress[i] = verifiedFileBytes(verified, file.Offset(), file.Length())
 	}
 	return append(fetcher.Progress(nil), f.data.Progress...)
 }
 
+type verifiedByteRange struct {
+	start, end int64
+}
+
+func verifiedFileBytes(verified []verifiedByteRange, offset, length int64) int64 {
+	first := sort.Search(len(verified), func(i int) bool { return verified[i].end > offset })
+	fileEnd := offset + length
+	var completed int64
+	for _, run := range verified[first:] {
+		if run.start >= fileEnd {
+			break
+		}
+		completed += min(fileEnd, run.end) - max(offset, run.start)
+	}
+	return completed
+}
+
+func pieceVerified(state torrent.PieceState) bool {
+	return state.Ok && state.Complete && !state.Hashing && !state.QueuedForHash
+}
+
 func (f *Fetcher) Wait() error {
 	f.mu.Lock()
 	var closed <-chan struct{}
@@ -587,7 +621,7 @@ func (f *Fetcher) isDoneLocked() bool {
 		file := f.torrent.Files()[index]
 		for i := file.BeginPieceIndex(); i < file.EndPieceIndex(); i++ {
 			state := f.torrent.Piece(i).State()
-			if !state.Ok || !state.Complete || state.Hashing || state.QueuedForHash {
+			if !pieceVerified(state) {
 				return false
 			}
 		}
@@ -685,11 +719,13 @@ func (f *Fetcher) Upload() (err error) {
 }
 
 func (f *Fetcher) doUpload(fromUpload bool) {
+	f.mu.Lock()
 	if !f.torrentUpload.CompareAndSwap(false, true) {
+		f.mu.Unlock()
 		return
 	}
-	f.mu.Lock()
+	f.seedBaseBytes = f.data.SeedBytes
-	initialBytes, initialTime := f.data.SeedBytes, f.data.SeedTime
+	initialTime := f.data.SeedTime
 	torrentClosed := f.torrent.Closed()
 	f.mu.Unlock()
 	var doneTime int64
@@ -707,8 +743,7 @@ func (f *Fetcher) doUpload(fromUpload bool) {
 				f.mu.Unlock()
 				continue
 			}
-			stats := f.torrentStats()
+			f.refreshSeedBytesLocked()
-			f.data.SeedBytes = initialBytes + stats.BytesWrittenData.Int64()
 			if !fromUpload && !f.isDoneLocked() {
 				f.mu.Unlock()
 				continue
@@ -741,9 +776,27 @@ func (f *Fetcher) torrentStats() torrent.TorrentStats {
 func (f *Fetcher) UploadedBytes() int64 {
 	f.mu.Lock()
 	defer f.mu.Unlock()
+	if f.torrentReady.Load() && f.torrentUpload.Load() {
+		f.refreshSeedBytesLocked()
+	}
 	return f.data.SeedBytes
 }
 
+func (f *Fetcher) refreshSeedBytesLocked() {
+	stats := f.torrentStats()
+	f.data.SeedBytes = max(f.data.SeedBytes, f.seedBaseBytes+stats.BytesWrittenData.Int64())
+}
+
+func (f *Fetcher) ReceivedBytes() int64 {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if !f.torrentReady.Load() {
+		return 0
+	}
+	stats := f.torrentStats()
+	return stats.BytesReadUsefulData.Int64()
+}
+
 func (f *Fetcher) WaitUpload() (err error) {
 	<-f.uploadDoneCh
 	return nil

internal/protocol/bt/fixture_test.go

@@ -46,6 +46,11 @@ func newBTHybridFixture(t testing.TB, payload []byte) *btFixture {
 }
 
 func makeBTFixture(t testing.TB, files map[string][]byte, bytesPerSecond int, v2, private bool) *btFixture {
+	t.Helper()
+	return makeBTFixtureWithPieceSize(t, files, bytesPerSecond, v2, private, 16*1024)
+}
+
+func makeBTFixtureWithPieceSize(t testing.TB, files map[string][]byte, bytesPerSecond int, v2, private bool, pieceSize int64) *btFixture {
 	t.Helper()
 	source := t.TempDir()
 	name := "payload.bin"
@@ -64,7 +69,7 @@ func makeBTFixture(t testing.TB, files map[string][]byte, bytesPerSecond int, v2
 		}
 		expected[filepath.Join(name, filepath.FromSlash(path))] = append([]byte(nil), data...)
 	}
-	info := metainfo.Info{PieceLength: 16 * 1024}
+	info := metainfo.Info{PieceLength: pieceSize}
 	if err := info.BuildFromFilePath(root); err != nil {
 		t.Fatal(err)
 	}

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/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

@@ -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()
@@ -508,31 +515,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 +1227,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 +1419,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 +1525,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 +1882,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 +1937,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 +2080,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 +2208,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 +2375,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
 		}
@@ -2459,9 +2430,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 +2454,18 @@ 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.speedArr = nil
+		task.downloadSampledAt = time.Time{}
 		task.timer.Start()
 		task.statusLock.Unlock()
-		startErr := task.fetcher.Start()
+		if err := task.fetcher.Start(); err != nil {
-		task.lifetime.mu.Unlock()
+			return err
-		if startErr != nil {
-			return startErr
 		}
 		started = true
 		if err := d.captureTaskStats(task, true, true); err != nil {
@@ -2614,7 +2576,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 +2589,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 +2598,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 +2618,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 +2646,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)
@@ -2697,11 +2659,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 {
@@ -2745,7 +2708,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 +2737,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 +2753,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 +2849,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 +2869,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 +3013,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,23 +3059,22 @@ 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{}
 	task.lock = &sync.Mutex{}
 	task.statsLock = &sync.Mutex{}
 	task.blobRefLock = &sync.Mutex{}
-	task.speedArr = make([]int64, 0)
+	task.speedArr = nil
-	task.uploadSpeedArr = make([]int64, 0)
+	task.uploadSpeedArr = nil
+	task.downloadSampledAt = time.Time{}
+	task.uploadSampledAt = time.Time{}
 }
 
 func (d *Downloader) releaseBlobTask(task *Task) {

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/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"`
@@ -84,21 +75,23 @@ type Task struct {
 	producer         production.Source
 	producerReceived int64
 
-	fetcherManager fetcher.FetcherManager
+	fetcherManager    fetcher.FetcherManager
-	fetcher        fetcher.Fetcher
+	fetcher           fetcher.Fetcher
-	timer          *util.Timer
+	timer             *util.Timer
-	statusLock     *sync.Mutex
+	statusLock        *sync.Mutex
-	lock           *sync.Mutex
+	lock              *sync.Mutex
-	statsLock      *sync.Mutex
+	statsLock         *sync.Mutex
-	statsLoaded    bool
+	statsLoaded       bool
-	stats          *fetcher.Stats
+	stats             *fetcher.Stats
-	statsSnapshot  json.RawMessage
+	statsSnapshot     json.RawMessage
-	blobRefLock    *sync.Mutex
+	blobRefLock       *sync.Mutex
-	blobURL        string
+	blobURL           string
-	runGeneration  uint64
+	runGeneration     uint64
-	lifetime       *taskLifetime
+	speedArr          []speedSample
-	speedArr       []int64
+	uploadSpeedArr    []speedSample
-	uploadSpeedArr []int64
+	downloadSampledAt time.Time
+	uploadSampledAt   time.Time
+	receivedBytes     int64
 }
 
 // TaskRuntimeStatus is the lightweight, frequently refreshed view of a task.
@@ -133,7 +126,6 @@ func NewTask() *Task {
 	}
 	return &Task{
 		ID:        id,
-		lifetime:  &taskLifetime{},
 		Status:    base.DownloadStatusReady,
 		CreatedAt: time.Now(),
 		UpdatedAt: time.Now(),
@@ -206,7 +198,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 +221,18 @@ func calcSpeed(speedArr *[]int64, downloaded int64, usedTime float64) int64 {
 		return 0
 	}
 
-	*speedArr = append(*speedArr, downloaded)
+	*speedArr = append(*speedArr, speedSample{bytes: downloaded, seconds: usedTime})
-	// Record last 5 seconds of download speed to calculate the average speed
-	if len(*speedArr) > int(5.0/usedTime) {
-		*speedArr = (*speedArr)[1:]
-	}
-
 	var total int64
-	for _, v := range *speedArr {
+	var elapsed float64
-		total += v
+	first := len(*speedArr)
+	for first > 0 && elapsed < 5 {
+		first--
+		sample := (*speedArr)[first]
+		total += sample.bytes
+		elapsed += sample.seconds
 	}
-
+	*speedArr = (*speedArr)[first:]
-	return int64(float64(total) / float64(len(*speedArr)) / usedTime)
+	return int64(float64(total) / elapsed)
 }
 
 type TaskFilter struct {

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

ui/flutter/lib/app/application/app_runtime_controller.dart

@@ -53,6 +53,7 @@ class AppRuntimeState {
 class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
   static StartConfig? _defaultStartConfig;
   Future<void> _preferenceWrites = Future<void>.value();
+  bool _analyticsStarted = false;
 
   @override
   Future<AppRuntimeState> build() async {
@@ -73,6 +74,7 @@ class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
       startConfig
         ..apiEnable = true
         ..mcpEnable = config.api.mcpEnable;
+      await _initAnalytics(config);
     } else {
       startConfig
         ..apiEnable = config.api.enable
@@ -253,7 +255,8 @@ class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
   }
 
   Future<void> _initAnalytics(DownloaderConfig config) async {
-    if (!AnalyticsConfig.isConfigured || !config.extra.analyticsEnabled) return;
+    if (_analyticsStarted || !AnalyticsConfig.isConfigured || !config.extra.analyticsEnabled) return;
+    _analyticsStarted = true;
     try {
       final generatedClientId = await Analytics.instance.init(fallbackClientId: config.extra.deviceId);
       if (generatedClientId != null) {
@@ -262,6 +265,7 @@ class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
       }
       unawaited(Analytics.instance.logAppOpen());
     } catch (error, stackTrace) {
+      _analyticsStarted = false;
       logger.w('GA4 init failed', error, stackTrace);
     }
   }

ui/flutter/lib/app/application/downloader_config_init.dart

@@ -19,7 +19,7 @@ Future<DownloaderConfig> loadDownloaderConfig({
     config = await load();
   } catch (error, stackTrace) {
     onLoadError?.call(error, stackTrace);
-    config = DownloaderConfig();
+    config = DownloaderConfig()..extra.analyticsEnabled = false;
     loaded = false;
   }
   final changed = await initDownloaderConfig(config, defaultDownloadDir: defaultDownloadDir);

ui/flutter/test/app/application/downloader_config_init_test.dart

@@ -32,8 +32,22 @@ void main() {
       // The in-memory fallback still gets usable defaults.
       expect(config.downloadDir, '/downloads');
       expect(config.categories, hasLength(4));
+      expect(config.extra.analyticsEnabled, isFalse);
     });
 
+    for (final analyticsEnabled in [false, true]) {
+      test('preserves the saved analytics preference ($analyticsEnabled)', () async {
+        final stored = configWith()..extra.analyticsEnabled = analyticsEnabled;
+        final config = await loadDownloaderConfig(
+          load: () async => stored,
+          save: (_) async {},
+          defaultDownloadDir: () async => '/downloads',
+        );
+
+        expect(config.extra.analyticsEnabled, analyticsEnabled);
+      });
+    }
+
     test('skips persist when init changed nothing', () async {
       var saved = 0;
       await loadDownloaderConfig(

ui/flutter/windows/runner/CMakeLists.txt

@@ -7,6 +7,7 @@ project(runner LANGUAGES CXX)
 #
 # Any new source files that you add to the application should be added here.
 add_executable(${BINARY_NAME} WIN32
+  "crash_handler.cpp"
   "flutter_window.cpp"
   "main.cpp"
   "utils.cpp"
@@ -34,6 +35,7 @@ target_compile_definitions(${BINARY_NAME} PRIVATE "NOMINMAX")
 # dependencies here.
 target_link_libraries(${BINARY_NAME} PRIVATE flutter flutter_wrapper_app)
 target_link_libraries(${BINARY_NAME} PRIVATE "dwmapi.lib")
+target_link_libraries(${BINARY_NAME} PRIVATE "dbghelp.lib")
 target_include_directories(${BINARY_NAME} PRIVATE "${CMAKE_SOURCE_DIR}")
 
 # Run the Flutter tool portions of the build. This must not be removed.

ui/flutter/windows/runner/CRASH_DIAGNOSTICS.md

@@ -0,0 +1,31 @@
+# Windows native crash diagnostics
+
+The runner installs diagnostics before starting Flutter and plugins. Files are
+written beside the executable under `storage/logs`, matching Windows app storage:
+
+- `native-crash.log`: UTC timestamp, app version, process/thread, exception code,
+  faulting module/address/offset, and minidump success or the Windows error code.
+- `native-crash.dmp`: the most recent native crash, including thread stacks,
+  loaded modules and unloaded module information.
+- `crash.log`: Go runtime crash output, kept separately from native diagnostics.
+
+Logs are appended and existing dumps survive normal restarts. A subsequent native
+crash replaces the previous dump. Dump generation is best effort, particularly
+for stack exhaustion, heap corruption or unavailable disk space. Process kills
+and fail-fast termination can bypass exception handlers.
+
+The top-level unhandled filter covers native exceptions on process threads. An
+SEH boundary inside the runner's window callback records the original escaping exception
+before Windows wraps a window callback failure as `0xC000041D`. A vectored handler
+also observes that fatal callback status, since it can bypass the unhandled
+filter. All handlers continue exception processing; they do not attempt recovery
+or log ordinary handled first-chance exceptions.
+
+On Windows, verify a release build by exercising tray clicks, Explorer restart,
+system font changes and child-window open/close. In a disposable test build,
+raise an unhandled access violation inside a window callback and on a worker
+thread, then raise `0xC000041D`; check that the log and dump are written and
+survive relaunch. A debugger may intercept these before application handlers.
+Open the dump with WinDbg and matching runner/Flutter/plugin PDBs to locate the
+faulting stack. Minidumps can contain application data; inspect them before
+sharing.

ui/flutter/windows/runner/crash_handler.cpp

@@ -0,0 +1,119 @@
+#include "crash_handler.h"
+
+#include <dbghelp.h>
+#include <strsafe.h>
+
+#include <cstdint>
+#include <cstring>
+#include <filesystem>
+#include <string>
+#include <system_error>
+
+namespace {
+
+HANDLE crash_log = INVALID_HANDLE_VALUE;
+std::wstring dump_path;
+volatile LONG recording = 0;
+LPTOP_LEVEL_EXCEPTION_FILTER previous_filter = nullptr;
+
+void WriteLog(const char* text) {
+  if (crash_log == INVALID_HANDLE_VALUE) return;
+  DWORD written = 0;
+  WriteFile(crash_log, text, static_cast<DWORD>(strlen(text)), &written,
+            nullptr);
+  FlushFileBuffers(crash_log);
+}
+
+LONG WINAPI UnhandledException(EXCEPTION_POINTERS* exception) {
+  RecordNativeCrash(exception);
+  return previous_filter ? previous_filter(exception)
+                         : EXCEPTION_CONTINUE_SEARCH;
+}
+
+LONG CALLBACK FatalCallbackException(EXCEPTION_POINTERS* exception) {
+  // Windows can terminate on this callback exception without invoking the
+  // process's unhandled filter. Ignore ordinary first-chance exceptions (Go,
+  // Dart and plugins may handle those) and observe only this fatal status.
+  if (exception->ExceptionRecord->ExceptionCode == 0xC000041D) {
+    RecordNativeCrash(exception);
+  }
+  return EXCEPTION_CONTINUE_SEARCH;
+}
+
+}  // namespace
+
+LONG RecordNativeCrash(EXCEPTION_POINTERS* exception) {
+  if (!exception || InterlockedCompareExchange(&recording, 1, 0) != 0) {
+    return EXCEPTION_CONTINUE_SEARCH;
+  }
+
+  SYSTEMTIME time;
+  GetSystemTime(&time);
+  const auto* record = exception->ExceptionRecord;
+  HMODULE module = nullptr;
+  wchar_t module_path[MAX_PATH] = L"unknown";
+  GetModuleHandleExW(GET_MODULE_HANDLE_EX_FLAG_FROM_ADDRESS |
+                        GET_MODULE_HANDLE_EX_FLAG_UNCHANGED_REFCOUNT,
+                    reinterpret_cast<LPCWSTR>(record->ExceptionAddress),
+                    &module);
+  if (module) GetModuleFileNameW(module, module_path, MAX_PATH);
+
+  char line[2048] = {};
+  StringCchPrintfA(
+      line, ARRAYSIZE(line),
+      "\r\n%04u-%02u-%02uT%02u:%02u:%02uZ Gopeed %s pid=%lu tid=%lu "
+      "exception=0x%08lX address=%p module=%ls offset=0x%llX\r\n",
+      time.wYear, time.wMonth, time.wDay, time.wHour, time.wMinute,
+      time.wSecond, FLUTTER_VERSION, GetCurrentProcessId(), GetCurrentThreadId(),
+      record->ExceptionCode, record->ExceptionAddress, module_path,
+      static_cast<unsigned long long>(
+          reinterpret_cast<uintptr_t>(record->ExceptionAddress) -
+          reinterpret_cast<uintptr_t>(module)));
+  WriteLog(line);
+
+  HANDLE dump = CreateFileW(dump_path.c_str(), GENERIC_WRITE, 0, nullptr,
+                            CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL, nullptr);
+  if (dump != INVALID_HANDLE_VALUE) {
+    MINIDUMP_EXCEPTION_INFORMATION info = {};
+    info.ThreadId = GetCurrentThreadId();
+    info.ExceptionPointers = exception;
+    info.ClientPointers = FALSE;
+    const BOOL ok = MiniDumpWriteDump(
+        GetCurrentProcess(), GetCurrentProcessId(), dump,
+        static_cast<MINIDUMP_TYPE>(MiniDumpNormal | MiniDumpWithThreadInfo |
+                                   MiniDumpWithUnloadedModules),
+        &info, nullptr, nullptr);
+    const DWORD error = ok ? ERROR_SUCCESS : GetLastError();
+    FlushFileBuffers(dump);
+    CloseHandle(dump);
+    StringCchPrintfA(line, ARRAYSIZE(line), "minidump success=%d error=%lu\r\n",
+                     ok, error);
+    WriteLog(line);
+  } else {
+    StringCchPrintfA(line, ARRAYSIZE(line), "minidump open error=%lu\r\n",
+                     GetLastError());
+    WriteLog(line);
+  }
+
+  // Do not swallow an exception or resume a potentially corrupted process.
+  return EXCEPTION_CONTINUE_SEARCH;
+}
+
+void InstallCrashHandler() {
+  // Match Dart's Windows storage path independently of the working directory.
+  wchar_t executable[32768] = {};
+  const DWORD length = GetModuleFileNameW(nullptr, executable, ARRAYSIZE(executable));
+  if (length == 0 || length >= ARRAYSIZE(executable)) return;
+  const auto directory = std::filesystem::path(executable).parent_path() /
+                         L"storage" / L"logs";
+  std::error_code error;
+  std::filesystem::create_directories(directory, error);
+  if (error) return;
+
+  dump_path = (directory / L"native-crash.dmp").wstring();
+  crash_log = CreateFileW((directory / L"native-crash.log").c_str(),
+                         FILE_APPEND_DATA, FILE_SHARE_READ, nullptr, OPEN_ALWAYS,
+                         FILE_ATTRIBUTE_NORMAL, nullptr);
+  previous_filter = SetUnhandledExceptionFilter(UnhandledException);
+  AddVectoredExceptionHandler(1, FatalCallbackException);
+}

ui/flutter/windows/runner/crash_handler.h

@@ -0,0 +1,12 @@
+#ifndef RUNNER_CRASH_HANDLER_H_
+#define RUNNER_CRASH_HANDLER_H_
+
+#include <windows.h>
+
+// Installs best-effort native diagnostics before Flutter or plugins start.
+void InstallCrashHandler();
+
+// Records the exception and continues normal Windows exception processing.
+LONG RecordNativeCrash(EXCEPTION_POINTERS* exception);
+
+#endif  // RUNNER_CRASH_HANDLER_H_

ui/flutter/windows/runner/flutter_window.cpp

@@ -63,7 +63,9 @@ FlutterWindow::MessageHandler(HWND hwnd, UINT const message,
 
   switch (message) {
     case WM_FONTCHANGE:
-      flutter_controller_->engine()->ReloadSystemFonts();
+      if (flutter_controller_ && flutter_controller_->engine()) {
+        flutter_controller_->engine()->ReloadSystemFonts();
+      }
       break;
   }
 

ui/flutter/windows/runner/main.cpp

@@ -5,6 +5,7 @@
 #include <desktop_multi_window/desktop_multi_window_plugin.h>
 
 #include "app_links/app_links_plugin_c_api.h"
+#include "crash_handler.h"
 #include "flutter_window.h"
 #include "flutter/generated_plugin_registrant.h"
 #include "utils.h"
@@ -43,6 +44,7 @@ bool IsMultiWindowLaunch(const std::vector<std::string>& arguments) {
 
 int APIENTRY wWinMain(_In_ HINSTANCE instance, _In_opt_ HINSTANCE prev,
                       _In_ wchar_t *command_line, _In_ int show_command) {
+  InstallCrashHandler();
   // Attach to console when present (e.g., 'flutter run') or create a
   // new console when running with a debugger.
   if (!::AttachConsole(ATTACH_PARENT_PROCESS) && ::IsDebuggerPresent()) {
@@ -77,11 +79,12 @@ int APIENTRY wWinMain(_In_ HINSTANCE instance, _In_opt_ HINSTANCE prev,
   window.SetQuitOnClose(true);
 
   ::MSG msg;
-  while (::GetMessage(&msg, nullptr, 0, 0)) {
+  BOOL message_result;
+  while ((message_result = ::GetMessage(&msg, nullptr, 0, 0)) > 0) {
     ::TranslateMessage(&msg);
     ::DispatchMessage(&msg);
   }
 
   ::CoUninitialize();
-  return EXIT_SUCCESS;
+  return message_result == -1 ? EXIT_FAILURE : static_cast<int>(msg.wParam);
 }

ui/flutter/windows/runner/win32_window.cpp

@@ -3,6 +3,7 @@
 #include <dwmapi.h>
 #include <flutter_windows.h>
 
+#include "crash_handler.h"
 #include "resource.h"
 
 namespace {
@@ -158,19 +159,26 @@ LRESULT CALLBACK Win32Window::WndProc(HWND const window,
                                       UINT const message,
                                       WPARAM const wparam,
                                       LPARAM const lparam) noexcept {
-  if (message == WM_NCCREATE) {
+  // Capture inside the callback, before user32 wraps or suppresses an escaping
-    auto window_struct = reinterpret_cast<CREATESTRUCT*>(lparam);
+  // exception. Keep C++ objects requiring unwinding out of this SEH boundary.
-    SetWindowLongPtr(window, GWLP_USERDATA,
+  __try {
-                     reinterpret_cast<LONG_PTR>(window_struct->lpCreateParams));
+    if (message == WM_NCCREATE) {
+      auto window_struct = reinterpret_cast<CREATESTRUCT*>(lparam);
+      SetWindowLongPtr(window, GWLP_USERDATA,
+                       reinterpret_cast<LONG_PTR>(window_struct->lpCreateParams));
 
-    auto that = static_cast<Win32Window*>(window_struct->lpCreateParams);
+      auto that = static_cast<Win32Window*>(window_struct->lpCreateParams);
-    EnableFullDpiSupportIfAvailable(window);
+      EnableFullDpiSupportIfAvailable(window);
-    that->window_handle_ = window;
+      that->window_handle_ = window;
-  } else if (Win32Window* that = GetThisFromHandle(window)) {
+    } else if (Win32Window* that = GetThisFromHandle(window)) {
-    return that->MessageHandler(window, message, wparam, lparam);
+      return that->MessageHandler(window, message, wparam, lparam);
+    }
+
+    return DefWindowProc(window, message, wparam, lparam);
+  } __except (RecordNativeCrash(GetExceptionInformation())) {
+    // The filter always continues the exception search; this is unreachable.
+    return 0;
   }
-
-  return DefWindowProc(window, message, wparam, lparam);
 }
 
 LRESULT