Merge PR #1520 into 2.0.0-beta.2

cmd/web/flags.go

@@ -13,6 +13,7 @@ import (
 )
 
 type args struct {
+	TempDir           *string  `json:"tempDir"`
 	Address           *string  `json:"address"`
 	Port              *int     `json:"port"`
 	Username          *string  `json:"username"`
@@ -54,6 +55,7 @@ func loadCliArgs() *args {
 	cfg.ApiToken = flag.String("T", "", "API token, it must be configured when using HTTP API in the case of enabling web authentication")
 	flag.StringVar(cfg.ApiToken, "api-token", "", "API token, it must be configured when using HTTP API in the case of enabling web authentication")
 	flag.BoolVar(&cfg.MCPEnable, "mcp-enable", false, "Enable MCP endpoint")
+	cfg.TempDir = flag.String("temp-dir", "", "Temporary directory (defaults to the OS temporary directory)")
 	cfg.StorageDir = flag.String("d", "", "Storage directory")
 	flag.StringVar(cfg.StorageDir, "storage-dir", "", "Storage directory")
 	whiteDownloadDirs := flag.String("white-download-dirs", "", "White download directories, comma-separated")
@@ -89,6 +91,8 @@ func overrideWithCliArgs(cfg *args, cliConfig *args) {
 			cfg.ApiToken = cliConfig.ApiToken
 		case "mcp-enable":
 			cfg.MCPEnable = cliConfig.MCPEnable
+		case "temp-dir":
+			cfg.TempDir = cliConfig.TempDir
 		case "d", "storage-dir":
 			cfg.StorageDir = cliConfig.StorageDir
 		case "white-download-dirs":
@@ -101,6 +105,9 @@ func overrideWithCliArgs(cfg *args, cliConfig *args) {
 
 // setDefaults sets default values for any unset configuration fields
 func setDefaults(cfg *args, cliConfig *args) {
+	if cfg.TempDir == nil {
+		cfg.TempDir = cliConfig.TempDir
+	}
 	if cfg.Address == nil {
 		cfg.Address = cliConfig.Address
 	}

cmd/web/main.go

@@ -43,11 +43,16 @@ func main() {
 		storageDir = filepath.Join(filepath.Dir(exe), "storage")
 	}
 
+	tempDir := ""
+	if args.TempDir != nil {
+		tempDir = *args.TempDir
+	}
 	cfg := &model.StartConfig{
 		Network:           "tcp",
 		Address:           fmt.Sprintf("%s:%d", *args.Address, *args.Port),
 		Storage:           model.StorageBolt,
 		StorageDir:        storageDir,
+		TempDir:           tempDir,
 		WhiteDownloadDirs: args.WhiteDownloadDirs,
 		ApiToken:          *args.ApiToken,
 		MCPEnable:         args.MCPEnable,

internal/blob/registry.go

@@ -7,6 +7,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"github.com/GopeedLab/gopeed/internal/production"
 	"io"
 	"net"
 	"net/http"
@@ -61,6 +62,7 @@ type Metadata struct {
 }
 
 type Source struct {
+	producer    production.Reader
 	ID          string
 	URL         string
 	ContentType string
@@ -359,11 +361,35 @@ func (r *Registry) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 	defer reader.Close()
+	if producer, ok := reader.(production.Reader); ok {
+		src.mu.Lock()
+		src.producer = producer
+		src.mu.Unlock()
+		if err := producer.WaitProduction(req.Context()); err != nil {
+			return
+		}
+		if producer.Production() != nil {
+			meta.Size = 0
+			meta.Range = false
+			start = 0
+			end = -1
+			ranged = false
+			src.mu.Lock()
+			src.size = 0
+			src.rangeEnabled = false
+			src.mu.Unlock()
+		}
+	}
 
 	writeHeaders(w, meta, start, end, ranged)
 	if ranged {
 		w.WriteHeader(http.StatusPartialContent)
 	}
+	if producer, ok := reader.(production.Reader); ok && producer.Production() != nil {
+		if flusher, ok := w.(http.Flusher); ok {
+			flusher.Flush()
+		}
+	}
 	limit := int64(-1)
 	if end >= start {
 		limit = end - start + 1
@@ -675,3 +701,21 @@ func randomID(size int) (string, error) {
 	}
 	return hex.EncodeToString(buf), nil
 }
+
+// Production returns the active opener's input counter, if it generates output.
+func (r *Registry) Production(raw string) production.Source {
+	if r == nil {
+		return nil
+	}
+	src, err := r.get(raw)
+	if err != nil {
+		return nil
+	}
+	src.mu.Lock()
+	reader := src.producer
+	src.mu.Unlock()
+	if reader == nil {
+		return nil
+	}
+	return reader.Production()
+}

internal/controller/controller.go

@@ -1,6 +1,7 @@
 package controller
 
 import (
+	"github.com/GopeedLab/gopeed/internal/tempfiles"
 	"github.com/GopeedLab/gopeed/pkg/base"
 	"net/http"
 	"net/url"
@@ -9,8 +10,11 @@ import (
 )
 
 type Controller struct {
-	GetConfig func(v any)
-	GetProxy  func(requestProxy *base.RequestProxy) func(*http.Request) (*url.URL, error)
+	TempFiles         *tempfiles.Scope
+	ManagedProduction func(string) bool
+	TempDir           string
+	GetConfig         func(v any)
+	GetProxy          func(requestProxy *base.RequestProxy) func(*http.Request) (*url.URL, error)
 	FileController
 	//ContextDialer() (proxy.Dialer, error)
 }

internal/production/progress.go

@@ -0,0 +1,35 @@
+// Package production associates generated resources with their input progress.
+package production
+
+import (
+	"context"
+	"sync"
+)
+
+// Source counts unique input bytes and total received bytes (including refetches).
+// Implementations must support concurrent snapshots.
+type Source interface {
+	Progress() (downloaded, received int64)
+}
+
+// Reader resolves its optional producer before response headers are sent.
+// A nil producer preserves ordinary Blob behavior.
+type Reader interface {
+	Production() Source
+	WaitProduction(context.Context) error
+}
+type Registry struct {
+	mu      sync.Mutex
+	sources map[string]Source
+}
+
+func (r *Registry) Set(id string, source Source) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.sources == nil {
+		r.sources = make(map[string]Source)
+	}
+	r.sources[id] = source
+}
+func (r *Registry) Get(id string) Source { r.mu.Lock(); defer r.mu.Unlock(); return r.sources[id] }
+func (r *Registry) Remove(id string)     { r.mu.Lock(); defer r.mu.Unlock(); delete(r.sources, id) }

internal/protocol/http/fetcher.go

@@ -512,17 +512,9 @@ func (f *Fetcher) asyncPrefetch() {
 		return
 	}
 
-	// Create temporary file for prefetch data
-	tmpFile, err := os.CreateTemp("", "gopeed-prefetch-*")
-	if err != nil {
-		f.prefetchErr = err
-		return
-	}
-	f.prefetchFile = tmpFile
-	f.prefetchFilePath = tmpFile.Name()
-
 	defer func() {
-		// Close response body when prefetch stops
+		// Close the response even if temporary-file setup fails or shutdown
+		// rejects a late prefetch. Otherwise the server can remain blocked writing.
 		f.resolveRespLock.Lock()
 		if f.resolveResp != nil {
 			f.resolveResp.Body.Close()
@@ -531,8 +523,33 @@ func (f *Fetcher) asyncPrefetch() {
 		f.resolveRespLock.Unlock()
 	}()
 
+	// Create temporary file for prefetch data
+	tempDir := ""
+	if f.ctl != nil && f.ctl.TempDir != "" {
+		tempDir = f.ctl.TempDir
+		if err := os.MkdirAll(tempDir, 0700); err != nil {
+			f.prefetchErr = err
+			return
+		}
+	}
+	tmpFile, err := os.CreateTemp(tempDir, "gopeed-prefetch-*")
+	if err != nil {
+		f.prefetchErr = err
+		return
+	}
+	if f.ctl != nil {
+		if err := f.ctl.TempFiles.Track(tmpFile.Name()); err != nil {
+			tmpFile.Close()
+			os.Remove(tmpFile.Name())
+			f.prefetchErr = err
+			return
+		}
+	}
+	f.prefetchFile = tmpFile
+	f.prefetchFilePath = tmpFile.Name()
+
 	buf := make([]byte, 32*1024) // 32KB buffer
-	reader := NewTimeoutReader(resp.Body, readTimeout)
+	reader := f.responseReader(resp)
 
 	for {
 		select {
@@ -617,7 +634,11 @@ func (f *Fetcher) cleanupPrefetchFile() {
 		f.prefetchFile = nil
 	}
 	if f.prefetchFilePath != "" {
-		os.Remove(f.prefetchFilePath)
+		if f.ctl != nil {
+			f.ctl.TempFiles.Remove(f.prefetchFilePath)
+		} else {
+			os.Remove(f.prefetchFilePath)
+		}
 		f.prefetchFilePath = ""
 	}
 	// The byte count is only reusable while the backing prefetch file exists.
@@ -1329,7 +1350,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [
 		f.slowStart.onConnectSuccess()
 	}
 
-	reader := NewTimeoutReader(resp.Body, readTimeout)
+	reader := f.responseReader(resp)
 	var responseBytesRead int64
 	for {
 		if conn.ctx.Err() != nil {
@@ -1711,7 +1732,7 @@ func (f *Fetcher) runConnectionWithResolveResp(conn *connection) {
 	}
 
 	// Download data from resolve response
-	reader := NewTimeoutReader(resp.Body, readTimeout)
+	reader := f.responseReader(resp)
 	for {
 		if conn.ctx.Err() != nil {
 			return
@@ -1828,7 +1849,7 @@ func (f *Fetcher) runConnectionFallback(conn *connection) {
 				f.slowStart.onConnectSuccess()
 			}
 
-			reader := NewTimeoutReader(resp.Body, readTimeout)
+			reader := f.responseReader(resp)
 			for {
 				if conn.ctx.Err() != nil {
 					return conn.ctx.Err()
@@ -2451,3 +2472,12 @@ func (f *Fetcher) Progress() fetcher.Progress {
 func (f *Fetcher) Wait() error {
 	return <-f.doneCh
 }
+
+// Generated output can pause while its producer waits for processing capacity.
+// Its producer owns upstream deadlines; cancellation still closes the response.
+func (f *Fetcher) responseReader(resp *http.Response) io.Reader {
+	if f.ctl != nil && f.ctl.ManagedProduction != nil && resp.Request != nil && f.ctl.ManagedProduction(resp.Request.URL.String()) {
+		return resp.Body
+	}
+	return NewTimeoutReader(resp.Body, readTimeout)
+}

internal/protocol/http/prefetch_setup_test.go

@@ -0,0 +1,74 @@
+package http
+
+import (
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/GopeedLab/gopeed/internal/controller"
+	"github.com/GopeedLab/gopeed/internal/tempfiles"
+)
+
+type prefetchSetupBody struct {
+	io.Reader
+	closed bool
+}
+
+func (b *prefetchSetupBody) Close() error {
+	b.closed = true
+	return nil
+}
+
+func TestPrefetchSetupFailureClosesResponse(t *testing.T) {
+	for _, closedScope := range []bool{false, true} {
+		name := "invalid_temp_directory"
+		if closedScope {
+			name = "downloader_already_closed"
+		}
+		t.Run(name, func(t *testing.T) {
+			root := t.TempDir()
+			scope := &tempfiles.Scope{}
+			dir := root
+			if closedScope {
+				if err := scope.Close(); err != nil {
+					t.Fatal(err)
+				}
+			} else {
+				dir = filepath.Join(root, "file")
+				if err := os.WriteFile(dir, nil, 0600); err != nil {
+					t.Fatal(err)
+				}
+			}
+			body := &prefetchSetupBody{Reader: strings.NewReader("unused response")}
+			f := &Fetcher{
+				ctl:         &controller.Controller{TempDir: dir, TempFiles: scope},
+				resolveResp: &http.Response{Body: body},
+			}
+			f.asyncPrefetch()
+			if f.prefetchErr == nil {
+				t.Fatal("expected temporary-file initialization to fail")
+			}
+			if !body.closed {
+				t.Error("prefetch setup failure left the response body open")
+			}
+			if f.resolveResp != nil {
+				t.Error("prefetch retained its failed resolve response")
+			}
+			if !f.prefetchDone.Load() {
+				t.Error("prefetch did not report completion")
+			}
+			if closedScope {
+				entries, err := os.ReadDir(root)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if len(entries) != 0 {
+					t.Errorf("late prefetch left temporary files: %v", entries)
+				}
+			}
+		})
+	}
+}

internal/tempfiles/scope.go

@@ -0,0 +1,63 @@
+// Package tempfiles tracks temporary paths owned by one downloader, without
+// requiring a private parent directory or scanning the shared system temp root.
+package tempfiles
+
+import (
+	"errors"
+	"io"
+	"os"
+	"sync"
+)
+
+type Scope struct {
+	mu     sync.Mutex
+	paths  map[string]struct{}
+	closed bool
+}
+
+// Track adopts a newly created unique path. Late creations are removed when
+// shutdown has already begun. A nil scope leaves cleanup to the resource owner.
+func (s *Scope) Track(path string) error {
+	if s == nil {
+		return nil
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.closed {
+		return errors.Join(io.ErrClosedPipe, os.RemoveAll(path))
+	}
+	if s.paths == nil {
+		s.paths = make(map[string]struct{})
+	}
+	s.paths[path] = struct{}{}
+	return nil
+}
+func (s *Scope) Remove(path string) error {
+	if s == nil {
+		return os.RemoveAll(path)
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if err := os.RemoveAll(path); err != nil {
+		return err
+	}
+	delete(s.paths, path)
+	return nil
+}
+func (s *Scope) Close() error {
+	if s == nil {
+		return nil
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.closed = true
+	var result error
+	for path := range s.paths {
+		if err := os.RemoveAll(path); err != nil {
+			result = errors.Join(result, err)
+		} else {
+			delete(s.paths, path)
+		}
+	}
+	return result
+}

internal/tempfiles/scope_test.go

@@ -0,0 +1,76 @@
+package tempfiles
+
+import (
+	"errors"
+	"io"
+	"os"
+	"sync"
+	"testing"
+)
+
+func TestSharedDirectoryOwnersAndLateCreation(t *testing.T) {
+	root := t.TempDir()
+	a, b := &Scope{}, &Scope{}
+	own := func(scope *Scope) string {
+		t.Helper()
+		dir, err := os.MkdirTemp(root, "gopeed-media-")
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := scope.Track(dir); err != nil {
+			t.Fatal(err)
+		}
+		return dir
+	}
+	first, second := own(a), own(b)
+	if err := a.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := os.Stat(first); !os.IsNotExist(err) {
+		t.Fatal(err)
+	}
+	if _, err := os.Stat(second); err != nil {
+		t.Fatal("removed other owner's path", err)
+	}
+	late, err := os.MkdirTemp(root, "gopeed-media-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := a.Track(late); !errors.Is(err, io.ErrClosedPipe) {
+		t.Fatal(err)
+	}
+	if _, err := os.Stat(late); !os.IsNotExist(err) {
+		t.Fatal("late creation leaked", err)
+	}
+	if err := b.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestConcurrentCreationAndShutdown(t *testing.T) {
+	root := t.TempDir()
+	scope := &Scope{}
+	var wg sync.WaitGroup
+	for i := 0; i < 20; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			dir, err := os.MkdirTemp(root, "gopeed-media-")
+			if err != nil {
+				t.Error(err)
+				return
+			}
+			if err := scope.Track(dir); err != nil && !errors.Is(err, io.ErrClosedPipe) {
+				t.Error(err)
+			}
+		}()
+	}
+	if err := scope.Close(); err != nil {
+		t.Fatal(err)
+	}
+	wg.Wait()
+	entries, err := os.ReadDir(root)
+	if err != nil || len(entries) != 0 {
+		t.Fatal("shutdown leaked paths", entries, err)
+	}
+}

pkg/download/downloader.go

@@ -21,6 +21,7 @@ import (
 	"github.com/GopeedLab/gopeed/internal/controller"
 	"github.com/GopeedLab/gopeed/internal/fetcher"
 	"github.com/GopeedLab/gopeed/internal/logger"
+	"github.com/GopeedLab/gopeed/internal/tempfiles"
 	webviewproxy "github.com/GopeedLab/gopeed/internal/webview/proxy"
 	"github.com/GopeedLab/gopeed/pkg/base"
 	"github.com/GopeedLab/gopeed/pkg/protocol/http"
@@ -98,6 +99,7 @@ type Progress struct {
 }
 
 type Downloader struct {
+	tempFiles           *tempfiles.Scope
 	webviewProfilesLock sync.Mutex
 	webviewProfiles     map[string]*extensionWebViewProfile
 	webviewProxyLock    sync.Mutex
@@ -157,6 +159,9 @@ func NewDownloader(cfg *DownloaderConfig) *Downloader {
 }
 
 func (d *Downloader) Setup() error {
+	if err := d.setupTempDir(); err != nil {
+		return err
+	}
 	d.blob = internalblob.NewRegistry("")
 
 	// setup storage
@@ -268,24 +273,45 @@ func (d *Downloader) Setup() error {
 						if d.GetTask(task.ID) == nil {
 							return
 						}
+						// Startup can replace resource metadata while resolving a Blob.
+						// Skip that short transition instead of reading half-initialized state.
+						if !task.lock.TryLock() {
+							return
+						}
 						task.statusLock.Lock()
 						if task.Status != base.DownloadStatusRunning && !task.Uploading {
 							task.statusLock.Unlock()
+							task.lock.Unlock()
 							return
 						}
 						if task.fetcher == nil {
 							task.statusLock.Unlock()
+							task.lock.Unlock()
 							return
 						}
 
 						current := task.fetcher.Progress().TotalDownloaded()
+						delta := current - task.Progress.Downloaded
+						if task.Meta != nil && task.Meta.Req != nil {
+							if producer := d.blob.Production(task.Meta.Req.URL); producer != nil {
+								downloaded, received := producer.Progress()
+								if task.producer != producer {
+									task.producer = producer
+									task.producerReceived = 0
+									task.speedArr = nil
+								}
+								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(current-task.Progress.Downloaded, tick)
+							task.Progress.Speed = task.updateSpeed(max(0, delta), tick)
 							task.Progress.Downloaded = current
 						}
 
@@ -304,6 +330,7 @@ func (d *Downloader) Setup() error {
 						if err := d.captureTaskStats(task, true, false); err != nil {
 							d.Logger.Warn().Err(err).Msgf("persist task stats failed, task id: %s", task.ID)
 						}
+						task.lock.Unlock()
 						// Listener callbacks may Pause/Continue and acquire statusLock.
 						d.emit(EventKeyProgress, task)
 
@@ -394,6 +421,9 @@ func (d *Downloader) parseFm(url string) (fetcher.FetcherManager, error) {
 
 func (d *Downloader) setupFetcher(fm fetcher.FetcherManager, fetcher fetcher.Fetcher) {
 	ctl := controller.NewController()
+	ctl.TempDir = d.cfg.TempDir
+	ctl.TempFiles = d.tempFiles
+	ctl.ManagedProduction = func(raw string) bool { return d.blob.Production(raw) != nil }
 	ctl.GetConfig = func(v any) {
 		d.getProtocolConfig(fm.Name(), v)
 	}
@@ -1134,7 +1164,7 @@ func (d *Downloader) Close() error {
 	if d.blob != nil {
 		closeArr = append(closeArr, d.blob.Close)
 	}
-	closeArr = append(closeArr, d.storage.Close)
+	closeArr = append(closeArr, d.storage.Close, d.tempFiles.Close)
 	// Make sure all resources are released, if had error, return the last error
 	var lastErr error
 	for i, close := range closeArr {
@@ -1409,17 +1439,6 @@ func (d *Downloader) watch(task *Task) {
 			}
 		}
 
-		task.Progress.Used = task.timer.Used()
-		if task.Meta.Res.Size == 0 {
-			task.Meta.Res.Size = task.fetcher.Progress().TotalDownloaded()
-		}
-		used := task.Progress.Used / int64(time.Second)
-		if used == 0 {
-			used = 1
-		}
-		totalSize := task.Meta.Res.Size
-		task.Progress.Speed = totalSize / used
-		task.Progress.Downloaded = totalSize
 		if !d.markTaskDone(task) {
 			return
 		}
@@ -1818,7 +1837,9 @@ func (d *Downloader) doStart(task *Task) (err error) {
 			task.Meta.Res.CalcSize(task.Meta.Opts.SelectFiles)
 		}
 
+		task.statusLock.Lock()
 		task.Progress.Speed = 0
+		task.statusLock.Unlock()
 		if !d.taskIsRunningGeneration(task, generation) {
 			return nil
 		}
@@ -1972,11 +1993,27 @@ func (d *Downloader) markTaskDone(task *Task) bool {
 	if task == nil || task.statusLock == nil {
 		return false
 	}
+	task.lock.Lock()
+	defer task.lock.Unlock()
 	task.statusLock.Lock()
 	defer task.statusLock.Unlock()
 	if task.Status != base.DownloadStatusRunning {
 		return false
 	}
+	task.Progress.Used = task.timer.Used()
+	if task.Meta.Res.Size == 0 {
+		task.Meta.Res.Size = task.fetcher.Progress().TotalDownloaded()
+	}
+	used := task.Progress.Used / int64(time.Second)
+	if used == 0 {
+		used = 1
+	}
+	totalSize := task.Meta.Res.Size
+	task.Progress.Speed = totalSize / used
+	task.Progress.Downloaded = totalSize
+
+	task.producer = nil
+	task.producerReceived = 0
 	task.updateStatus(base.DownloadStatusDone)
 	task.runGeneration++
 	return true

pkg/download/engine/engine.go

@@ -6,6 +6,8 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/GopeedLab/gopeed/internal/production"
+	"github.com/GopeedLab/gopeed/internal/tempfiles"
 	"github.com/GopeedLab/gopeed/pkg/base"
 	gojaerror "github.com/GopeedLab/gopeed/pkg/download/engine/inject/error"
 	fetchapi "github.com/GopeedLab/gopeed/pkg/download/engine/inject/fetch"
@@ -160,7 +162,9 @@ func (e *Engine) addCleanup(cleanup func()) {
 }
 
 type Config struct {
+	TempFiles *tempfiles.Scope
 	// HTTPUserAgent is a snapshot of the downloader HTTP default. Nil leaves standalone engines unchanged.
+	TempDir       string
 	HTTPUserAgent *string
 	ProxyConfig   *base.DownloaderProxyConfig
 	StreamConfig  *stream.Config
@@ -203,10 +207,19 @@ func NewEngine(cfg *Config) *Engine {
 		if _, err := runtime.RunString("global.location = new URL('http://localhost');"); err != nil {
 			return
 		}
-		if err := stream.Enable(runtime, loop, cfg.StreamConfig); err != nil {
+		producers := &production.Registry{}
+		streamCfg := stream.Config{}
+		if cfg.StreamConfig != nil {
+			streamCfg = *cfg.StreamConfig
+		}
+		streamCfg.Producers = producers
+		if err := stream.Enable(runtime, loop, &streamCfg); err != nil {
 			return
 		}
 		if err := ffmpegapi.Enable(runtime, loop, &ffmpegapi.Config{
+			Producers:        producers,
+			TempDir:          cfg.TempDir,
+			TempFiles:        cfg.TempFiles,
 			ProxyHandler:     cfg.ProxyConfig.ToHandler(),
 			DefaultUserAgent: cfg.HTTPUserAgent,
 			RegisterCleanup:  engine.addCleanup,

pkg/download/engine/ffmpeg/README.md

@@ -54,8 +54,8 @@ const output = gopeed.runtime.ffmpeg.merge({
 ```
 
 Creating the output does not start requests or acquire input readers until the
-output is pulled and a global execution slot is acquired. Queued jobs do not
-open HTTP inputs, invoke input factories, or allocate media ring buffers.
+output is pulled. Inputs then download to temporary disk chunks even while the
+job waits for a global FFmpeg execution slot. Only merging occupies a slot.
 The callback runs once per output; its signal aborts on cancellation, engine
 shutdown, failure, and completion. A callback must cooperate with cancellation;
 returned streams are cancelled even if the callback finishes after cancellation.
@@ -63,8 +63,10 @@ A stream input is single-use. To retry a Blob-backed task,
 its opener must create a new SABR session and fresh input streams. Arbitrary
 output byte ranges / resumable merged output are not supported.
 
-`gopeed.runtime.ffmpeg.supportsInputFactory === true` identifies this API.
-Older builds accept eager streams and must be upgraded before using callbacks.
+`gopeed.runtime.ffmpeg.supportsInputFactory === true` identifies the callback API.
+`gopeed.runtime.ffmpeg.supportsProducerProgress === true` identifies disk-backed
+inputs and automatic Blob production progress. Require the latter before using
+one shared SABR session for both tracks; older bounded buffers can deadlock.
 
 ## Options
 
@@ -109,15 +111,45 @@ server validator, same-size remote changes cannot be reliably detected.
 Content encoding is forced to identity. Authentication/status failures and
 malformed range responses are errors. There is no full-file seek fallback.
 
-Sequential JS inputs are drained independently into memory ring buffers that
-grow on demand up to **32 MiB per input**, independent of total media size.
-No temporary files are used. This does not enable seeking. Full buffers apply
-async backpressure until space is available or the operation is cancelled.
-Extensions must propagate backpressure through their upstream queues and
-network reads. Independent upstream track requests avoid blocking one track
-behind another in a shared response. Buffers are released on close.
-Blob output writes also wait asynchronously for bounded downstream capacity,
-so a slow consumer does not block the extension event loop.
+HTTP and sequential JS inputs are drained into **8 MiB disk chunks** beneath
+`TempDir/gopeed-media-<random>`. Range downloads stream each response into as many chunks as
+needed, tracking valid byte intervals independently of sparse file length.
+Verified HTTP Range inputs can seek, skip missing blocks, and re-fetch deleted
+blocks. Sequential inputs cannot seek even though their bytes are on disk.
+Consumed chunks are deleted; cancellation, errors and completion remove the
+remaining input directories. Disk space is not capped; write failures stop the
+merge. Media size does not determine the size of in-memory input queues.
+
+JS writes await disk writes, and Blob output writes await bounded downstream
+capacity. Extensions must still propagate backpressure through upstream queues
+and network reads. Both tracks can share a SABR response because disk spooling
+continues independently of which track FFmpeg currently reads.
+
+When a Blob opener returns the merge stream directly, the runtime automatically
+associates its input progress with that Blob. No public progress callback or
+extra `createObjectURL` argument is required. Return the merge stream itself;
+wrapping it in another stream loses this association. Downloaded bytes count
+unique input offsets; speed counts received input bytes, including re-fetches.
+The task stays downloading with unknown total size while queued or merging;
+once input traffic stops the displayed speed settles to zero. Completion uses
+the actual merged output size. Generated output waits are governed by producer
+cancellation/errors rather than the HTTP body idle timeout; HTTP media inputs
+still enforce a 15-second network idle deadline.
+
+`tempDir` defaults to Go's `os.TempDir()` when omitted. Flutter only supplies
+an explicit directory on Android/iOS, using `getTemporaryDirectory()` directly;
+no extra `gopeed` parent directory is added. Desktop/server callers can override
+it via startup JSON or `--temp-dir`.
+
+Initialization never clears this directory. Media inputs, HTTP prefetch and
+extension installation staging create unique `gopeed-media-*`,
+`gopeed-prefetch-*`, and `gopeed-extension-*` paths directly beneath it. The
+downloader tracks only its own paths and removes remaining ones on normal
+shutdown, without scanning or deleting other files in a shared temporary
+directory. Resources still clean up on completion/cancellation, and consumed
+media chunks are deleted promptly. Forced process termination can leave files
+behind; the next startup does not delete them. Persistent HLS resume data and
+WebView profiles are unchanged.
 
 At most two WASM merges execute concurrently per process, with a 512 MiB
 linear-memory limit per instance. Compilation is shared and lazy; the upstream

pkg/download/engine/ffmpeg/http_spool.go

@@ -0,0 +1,248 @@
+package ffmpeg
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"sync"
+
+	"github.com/GopeedLab/gopeed/internal/tempfiles"
+)
+
+// HTTPSpool downloads ahead independently of the FFmpeg semaphore. Range
+// sources can fetch a missing block on demand, including blocks already evicted.
+type HTTPSpool struct {
+	demand     chan int64
+	demandDone chan struct{}
+	ctx        context.Context
+	cancel     context.CancelFunc
+	source     Input
+	disk       *DiskInput
+	network    sync.Mutex
+	readMu     sync.Mutex
+	pos        int64
+	done       chan struct{}
+}
+
+func OpenSpoolingHTTP(ctx context.Context, client *http.Client, source HTTPSource, root string, owners ...*tempfiles.Scope) (Input, error) {
+	ctx, cancel := context.WithCancel(ctx)
+	in, err := OpenHTTP(ctx, client, source)
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+	disk, err := NewDiskInput(ctx, root, owners...)
+	if err != nil {
+		cancel()
+		in.Close()
+		return nil, err
+	}
+	if ranged := in.(*httpInput); ranged.ranged {
+		if _, err = disk.spool.WriteAt(ranged.block, 0); err != nil {
+			cancel()
+			in.Close()
+			disk.Close()
+			return nil, err
+		}
+		ranged.block = nil
+	}
+	h := &HTTPSpool{ctx: ctx, cancel: cancel, source: in, disk: disk, done: make(chan struct{}), demand: make(chan int64, 1), demandDone: make(chan struct{})}
+	go func() {
+		defer close(h.demandDone)
+		for {
+			select {
+			case offset := <-h.demand:
+				// A prefetch may have filled this position while demand was queued.
+				var probe [1]byte
+				if n, err, _ := h.disk.spool.readAt(probe[:], offset); n > 0 || err != nil {
+					continue
+				}
+				if err := h.fetch(offset / SpoolBlockSize * SpoolBlockSize); err != nil {
+					h.disk.End(err)
+					return
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	go func() {
+		defer close(h.done)
+		var err error
+		if !in.Seekable() {
+			_, err = io.CopyBuffer(disk, in, make([]byte, 64*1024))
+		} else {
+			for offset := int64(0); offset < in.Size(); offset += 3 * SpoolBlockSize {
+				if err = h.fetch(offset); err != nil {
+					break
+				}
+			}
+		}
+		disk.End(err)
+	}()
+	return h, nil
+}
+func (h *HTTPSpool) fetch(offset int64) error {
+	h.network.Lock()
+	defer h.network.Unlock()
+	if err := h.ctx.Err(); err != nil {
+		return err
+	}
+	source := h.source.(*httpInput)
+	end := offset + min(3*SpoolBlockSize, h.source.Size()-offset)
+	buffer := make([]byte, 64*1024)
+	for offset < end {
+		n, err, _ := h.disk.spool.readAt(buffer[:min(int64(len(buffer)), end-offset)], offset)
+		if err != nil {
+			return err
+		}
+		if n > 0 {
+			offset += int64(n)
+			continue
+		}
+		resp, err := source.request(offset, end-1)
+		if err != nil {
+			return err
+		}
+		err = func() error {
+			defer resp.Body.Close()
+			if resp.StatusCode != http.StatusPartialContent {
+				return fmt.Errorf("ffmpeg: Range request returned %d; source changed or stopped supporting Range", resp.StatusCode)
+			}
+			start, last, size, err := contentRange(resp.Header.Get("Content-Range"))
+			if err != nil || start != offset || last >= end || size != source.size {
+				return errors.New("ffmpeg: inconsistent Content-Range")
+			}
+			if source.etag != "" && resp.Header.Get("ETag") != source.etag {
+				return errors.New("ffmpeg: HTTP input ETag changed")
+			}
+			if source.etag == "" && source.modified != "" && resp.Header.Get("Last-Modified") != source.modified {
+				return errors.New("ffmpeg: HTTP input Last-Modified changed")
+			}
+			if err := identityEncoding(resp); err != nil {
+				return err
+			}
+			remaining := last - start + 1
+			if resp.ContentLength >= 0 && resp.ContentLength != remaining {
+				return errors.New("ffmpeg: Range response length mismatch")
+			}
+			for remaining > 0 {
+				n, err := resp.Body.Read(buffer[:min(int64(len(buffer)), remaining)])
+				if n > 0 {
+					if _, e := h.disk.spool.WriteAt(buffer[:n], offset); e != nil {
+						return e
+					}
+					offset += int64(n)
+					remaining -= int64(n)
+				}
+				if err != nil {
+					if err == io.EOF && remaining == 0 {
+						return nil
+					}
+					if err == io.EOF {
+						return io.ErrUnexpectedEOF
+					}
+					return err
+				}
+				if n == 0 {
+					return io.ErrNoProgress
+				}
+			}
+			n, err := resp.Body.Read(buffer[:1])
+			if n != 0 || err != io.EOF {
+				if err != nil && err != io.EOF {
+					return err
+				}
+				return errors.New("ffmpeg: oversized Range response")
+			}
+			return nil
+		}()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+func (h *HTTPSpool) Read(p []byte) (int, error) {
+	if !h.Seekable() {
+		return h.disk.Read(p)
+	}
+	h.readMu.Lock()
+	defer h.readMu.Unlock()
+	if len(p) == 0 {
+		return 0, nil
+	}
+	if h.pos >= h.Size() {
+		if err := h.disk.spool.discard((h.Size() - 1) / SpoolBlockSize); err != nil {
+			return 0, err
+		}
+		return 0, io.EOF
+	}
+	for {
+		if err := h.ctx.Err(); err != nil {
+			return 0, err
+		}
+		n, err, wake := h.disk.spool.readAt(p, h.pos)
+		if n > 0 {
+			block := h.pos / SpoolBlockSize
+			h.pos += int64(n)
+			if h.pos/SpoolBlockSize > block {
+				if e := h.disk.spool.discard(block); err == nil {
+					err = e
+				}
+			}
+			return n, err
+		}
+		if err != nil {
+			return 0, err
+		}
+		select {
+		case h.demand <- h.pos:
+		default:
+		}
+		// Read as soon as a partial response supplies these bytes, rather than
+		// waiting for its entire multi-block Range request to finish.
+		select {
+		case <-wake:
+		case <-h.ctx.Done():
+			return 0, h.ctx.Err()
+		}
+	}
+}
+func (h *HTTPSpool) Seek(offset int64, whence int) (int64, error) {
+	if !h.Seekable() {
+		return 0, ErrSeek
+	}
+	h.readMu.Lock()
+	defer h.readMu.Unlock()
+	base := int64(0)
+	switch whence {
+	case io.SeekStart:
+	case io.SeekCurrent:
+		base = h.pos
+	case io.SeekEnd:
+		base = h.Size()
+	default:
+		return 0, errors.New("invalid seek origin")
+	}
+	next := base + offset
+	if next < 0 || (offset > 0 && next < base) {
+		return 0, errors.New("invalid seek offset")
+	}
+	h.pos = next
+	return next, nil
+}
+func (h *HTTPSpool) Size() int64              { return h.source.Size() }
+func (h *HTTPSpool) Seekable() bool           { return h.source.Seekable() }
+func (h *HTTPSpool) Progress() (int64, int64) { return h.disk.Progress() }
+func (h *HTTPSpool) Close() error {
+	h.cancel()
+	h.source.Close()
+	<-h.done
+	<-h.demandDone
+	return h.disk.Close()
+}
+
+func (h *HTTPSpool) Failures() <-chan error { return h.disk.Failures() }

pkg/download/engine/ffmpeg/http_spool_test.go

@@ -0,0 +1,198 @@
+package ffmpeg
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestHTTPSpoolRangeCrossBlocksAndRefetch(t *testing.T) {
+	data := make([]byte, 4*SpoolBlockSize+57)
+	for i := range data {
+		data[i] = byte(i*31 + i/257)
+	}
+	var large atomic.Bool
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var start, end int64
+		fmt.Sscanf(r.Header.Get("Range"), "bytes=%d-%d", &start, &end)
+		if end-start+1 > SpoolBlockSize {
+			large.Store(true)
+		}
+		w.Header().Set("ETag", `"stable"`)
+		http.ServeContent(w, r, "media", time.Time{}, bytes.NewReader(data))
+	}))
+	defer srv.Close()
+	input, err := OpenSpoolingHTTP(context.Background(), srv.Client(), HTTPSource{URL: srv.URL}, t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer input.Close()
+	h := input.(*HTTPSpool)
+	select {
+	case <-h.done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("prefetch did not finish")
+	}
+	if !large.Load() {
+		t.Fatal("did not fetch multiple blocks in one response")
+	}
+	for _, offset := range []int64{3*SpoolBlockSize - 123, 0, 2*SpoolBlockSize + 41, SpoolBlockSize - 5} {
+		if _, err := h.Seek(offset, io.SeekStart); err != nil {
+			t.Fatal(err)
+		}
+		got := make([]byte, min(int64(2*SpoolBlockSize), int64(len(data))-offset))
+		if _, err := io.ReadFull(h, got); err != nil {
+			t.Fatal(err)
+		}
+		if !bytes.Equal(got, data[offset:offset+int64(len(got))]) {
+			t.Fatal("incorrect random/cross-block read", offset)
+		}
+	}
+	unique, received := h.Progress()
+	if unique != int64(len(data)) || received <= unique {
+		t.Fatal("refetch progress must not double-count stored bytes", unique, received)
+	}
+}
+
+func TestHTTPSpoolSequentialAndCancellation(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	entered := make(chan struct{})
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte("hello"))
+		w.(http.Flusher).Flush()
+		close(entered)
+		<-r.Context().Done()
+	}))
+	defer srv.Close()
+	in, err := OpenSpoolingHTTP(ctx, srv.Client(), HTTPSource{URL: srv.URL}, t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	<-entered
+	if in.Seekable() {
+		t.Fatal("200 response advertised seek")
+	}
+	got := make([]byte, 5)
+	if _, err := io.ReadFull(in, got); err != nil || string(got) != "hello" {
+		t.Fatal(string(got), err)
+	}
+	cancel()
+	if err := in.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestHTTPSpoolRejectsBadRanges(t *testing.T) {
+	for _, mode := range []string{"etag", "length", "offset", "status", "truncated", "encoding"} {
+		t.Run(mode, func(t *testing.T) {
+			var calls atomic.Int32
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				first := calls.Add(1) == 1
+				start, end := int64(0), int64(httpBlockSize-1)
+				if !first {
+					fmt.Sscanf(r.Header.Get("Range"), "bytes=%d-%d", &start, &end)
+				}
+				tag := `"stable"`
+				if !first && mode == "etag" {
+					tag = `"changed"`
+				}
+				w.Header().Set("ETag", tag)
+				if !first && mode == "status" {
+					w.WriteHeader(200)
+					return
+				}
+				if !first && mode == "offset" {
+					start++
+				}
+				if !first && mode == "encoding" {
+					w.Header().Set("Content-Encoding", "gzip")
+				}
+				w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, 2*SpoolBlockSize))
+				length := end - start + 1
+				if !first && mode == "length" {
+					length++
+				}
+				w.Header().Set("Content-Length", fmt.Sprint(length))
+				w.WriteHeader(206)
+				if !first && mode == "truncated" {
+					length /= 2
+				}
+				chunk := make([]byte, 64*1024)
+				for length > 0 {
+					n := min(int64(len(chunk)), length)
+					if _, err := w.Write(chunk[:n]); err != nil {
+						return
+					}
+					length -= n
+				}
+			}))
+			defer srv.Close()
+			in, err := OpenSpoolingHTTP(context.Background(), srv.Client(), HTTPSource{URL: srv.URL}, t.TempDir())
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer in.Close()
+			h := in.(*HTTPSpool)
+			select {
+			case err := <-h.Failures():
+				if err == nil {
+					t.Fatal("nil failure")
+				}
+			case <-time.After(5 * time.Second):
+				t.Fatal("bad range did not fail")
+			}
+		})
+	}
+}
+
+func TestHTTPSpoolReadsPartialRangeBeforeResponseFinishes(t *testing.T) {
+	ready, finish := make(chan struct{}), make(chan struct{})
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var start, end int64
+		fmt.Sscanf(r.Header.Get("Range"), "bytes=%d-%d", &start, &end)
+		end = min(end, 4*1024*1024-1)
+		w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, 4*1024*1024))
+		w.Header().Set("Content-Length", fmt.Sprint(end-start+1))
+		w.WriteHeader(206)
+		if start == 0 {
+			w.Write(make([]byte, end+1))
+			return
+		}
+		w.Write(bytes.Repeat([]byte{93}, 64*1024))
+		w.(http.Flusher).Flush()
+		close(ready)
+		select {
+		case <-finish:
+			w.Write(make([]byte, end-start+1-64*1024))
+		case <-r.Context().Done():
+		}
+	}))
+	defer srv.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	input, err := OpenSpoolingHTTP(ctx, srv.Client(), HTTPSource{URL: srv.URL}, t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer input.Close()
+	select {
+	case <-ready:
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	}
+	if _, err := input.Seek(1024*1024, io.SeekStart); err != nil {
+		t.Fatal(err)
+	}
+	b := make([]byte, 1)
+	if _, err := io.ReadFull(input, b); err != nil || b[0] != 93 {
+		t.Fatal("waited for complete response", b, err)
+	}
+	close(finish)
+}

pkg/download/engine/ffmpeg/input.go

@@ -10,6 +10,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"time"
 )
 
 const httpBlockSize = 1024 * 1024
@@ -79,8 +80,10 @@ func OpenHTTP(ctx context.Context, client *http.Client, source HTTPSource) (Inpu
 }
 
 func (h *httpInput) request(start, end int64) (*http.Response, error) {
-	req, err := http.NewRequestWithContext(h.ctx, http.MethodGet, h.source.URL, nil)
+	ctx, cancel := context.WithCancel(h.ctx)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, h.source.URL, nil)
 	if err != nil {
+		cancel()
 		return nil, err
 	}
 	for k, v := range h.source.Headers {
@@ -94,7 +97,15 @@ func (h *httpInput) request(start, end int64) (*http.Response, error) {
 	} else if h.modified != "" {
 		req.Header.Set("If-Range", h.modified)
 	}
-	return h.client.Do(req)
+	timer := time.AfterFunc(15*time.Second, cancel)
+	resp, err := h.client.Do(req)
+	timer.Stop()
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+	resp.Body = &idleMediaBody{ReadCloser: resp.Body, cancel: cancel}
+	return resp, nil
 }
 
 func identityEncoding(resp *http.Response) error {
@@ -225,3 +236,17 @@ func (h *httpInput) Close() error {
 	}
 	return nil
 }
+
+// Only time spent waiting on the network counts toward this idle deadline.
+// Pauses between reads (disk writes or FFmpeg scheduling) do not count.
+type idleMediaBody struct {
+	io.ReadCloser
+	cancel context.CancelFunc
+}
+
+func (b *idleMediaBody) Read(p []byte) (int, error) {
+	timer := time.AfterFunc(15*time.Second, b.cancel)
+	defer timer.Stop()
+	return b.ReadCloser.Read(p)
+}
+func (b *idleMediaBody) Close() error { b.cancel(); return b.ReadCloser.Close() }

pkg/download/engine/ffmpeg/input_test.go

@@ -115,29 +115,6 @@ func TestHTTPSequentialAndValidation(t *testing.T) {
 	}
 }
 
-func TestStreamRing(t *testing.T) {
-	s := NewStreamInput(context.Background())
-	defer s.Close()
-	// Start near the boundary to exercise wrap-around without a huge fixture.
-	s.buffer = make([]byte, 256*1024)
-	s.read = int64(len(s.buffer)) - 3
-	s.written = s.read
-	if _, err := s.Write([]byte("abcdefgh")); err != nil {
-		t.Fatal(err)
-	}
-	s.End(nil)
-	b, err := io.ReadAll(s)
-	if err != nil || string(b) != "abcdefgh" {
-		t.Fatalf("%q %v", b, err)
-	}
-	if err = s.Close(); err != nil {
-		t.Fatal(err)
-	}
-	if s.buffer != nil {
-		t.Fatal("buffer retained after close")
-	}
-}
-
 func TestOutputArguments(t *testing.T) {
 	a, err := Arguments("", []string{"-shortest", "-metadata", "title=test"})
 	if err != nil {
@@ -152,71 +129,3 @@ func TestOutputArguments(t *testing.T) {
 		}
 	}
 }
-
-func TestStreamBackpressureAndCancel(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	s := NewStreamInput(ctx)
-	defer s.Close()
-	s.buffer = make([]byte, StreamBufferLimit)
-	s.written = StreamBufferLimit
-	finished := make(chan error, 1)
-	go func() { _, err := s.Write([]byte{42}); finished <- err }()
-	select {
-	case err := <-finished:
-		t.Fatalf("did not apply backpressure: %v", err)
-	case <-time.After(10 * time.Millisecond):
-	}
-	if _, err := s.Read(make([]byte, 1)); err != nil {
-		t.Fatal(err)
-	}
-	select {
-	case err := <-finished:
-		if err != nil {
-			t.Fatal(err)
-		}
-	case <-time.After(time.Second):
-		t.Fatal("writer did not resume")
-	}
-	go func() { _, err := s.Write([]byte{43}); finished <- err }()
-	cancel()
-	select {
-	case err := <-finished:
-		if err == nil {
-			t.Fatal("cancelled write succeeded")
-		}
-	case <-time.After(time.Second):
-		t.Fatal("writer did not cancel")
-	}
-}
-
-func TestStreamMemoryGrowthAndWrap(t *testing.T) {
-	s := NewStreamInput(context.Background())
-	defer s.Close()
-	if len(s.buffer) != 0 {
-		t.Fatal("buffer allocated before first write")
-	}
-	a := bytes.Repeat([]byte{1}, 200*1024)
-	b := bytes.Repeat([]byte{2}, 180*1024)
-	s.Write(a)
-	consumed := make([]byte, 180*1024)
-	if _, err := io.ReadFull(s, consumed); err != nil {
-		t.Fatal(err)
-	}
-	s.Write(b) // Wrap in the initial 256 KiB allocation.
-	c := bytes.Repeat([]byte{3}, 300*1024)
-	s.Write(c) // Grow while unread data wraps around the old ring.
-	s.End(nil)
-	got, err := io.ReadAll(s)
-	want := append(append(append([]byte{}, a[len(consumed):]...), b...), c...)
-	if err != nil || !bytes.Equal(got, want) {
-		t.Fatalf("growth corrupted data: %v", err)
-	}
-	if len(s.buffer) > StreamBufferLimit {
-		t.Fatal("buffer exceeded limit")
-	}
-	s.Close()
-	if s.buffer != nil {
-		t.Fatal("buffer retained on close")
-	}
-}

pkg/download/engine/ffmpeg/runtime.go

@@ -120,7 +120,8 @@ func Run(ctx context.Context, video, audio Input, out io.Writer, format string,
 	return RunAcquired(ctx, video, audio, out, format, extra)
 }
 
-// Acquire reserves execution capacity before input factories or HTTP requests start.
+// Acquire reserves WASM execution capacity. Disk-backed input downloads may
+// already be running while this call waits.
 func Acquire(ctx context.Context) (func(), error) {
 	select {
 	case slots <- struct{}{}:

pkg/download/engine/ffmpeg/runtime_test.go

@@ -122,13 +122,16 @@ func TestWebMToMP4(t *testing.T) {
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
-	inputs := make([]*StreamInput, 2)
+	inputs := make([]*DiskInput, 2)
 	for i, name := range []string{"video.webm", "audio.webm"} {
 		data, err := os.ReadFile(filepath.Join("..", "testdata", "ffmpeg", name))
 		if err != nil {
 			t.Fatal(err)
 		}
-		inputs[i] = NewStreamInput(ctx)
+		inputs[i], err = NewDiskInput(ctx, t.TempDir())
+		if err != nil {
+			t.Fatal(err)
+		}
 		defer inputs[i].Close()
 		if _, err = inputs[i].Write(data); err != nil {
 			t.Fatal(err)
@@ -205,7 +208,10 @@ func TestWASMMerge(t *testing.T) {
 		})
 	}
 	// A malformed input must produce an error, not successful empty output.
-	s := NewStreamInput(context.Background())
+	s, err := NewDiskInput(context.Background(), t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
 	defer s.Close()
 	s.Write([]byte("invalid"))
 	s.End(nil)

pkg/download/engine/ffmpeg/spool.go

@@ -0,0 +1,284 @@
+package ffmpeg
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"sync"
+
+	"github.com/GopeedLab/gopeed/internal/tempfiles"
+)
+
+const SpoolBlockSize int64 = 8 * 1024 * 1024
+
+type byteSpan struct{ start, end int64 }
+
+// blockSpool tracks valid byte intervals separately from file length: a sparse
+// file's unwritten holes are never valid media. No file handles survive a call.
+type blockSpool struct {
+	owner                *tempfiles.Scope
+	failures             chan error
+	mu                   sync.Mutex
+	dir                  string
+	blocks               map[int64][]byteSpan
+	seen                 []byteSpan
+	wake                 chan struct{}
+	ended, closed        bool
+	err                  error
+	downloaded, received int64
+}
+
+func newBlockSpool(root string, owners ...*tempfiles.Scope) (*blockSpool, error) {
+	if root != "" {
+		if err := os.MkdirAll(root, 0700); err != nil {
+			return nil, err
+		}
+	}
+	dir, err := os.MkdirTemp(root, "gopeed-media-")
+	if err != nil {
+		return nil, err
+	}
+	var owner *tempfiles.Scope
+	if len(owners) > 0 {
+		owner = owners[0]
+	}
+	if err := owner.Track(dir); err != nil {
+		return nil, err
+	}
+	return &blockSpool{owner: owner, failures: make(chan error, 1), dir: dir, blocks: make(map[int64][]byteSpan), wake: make(chan struct{})}, nil
+}
+func (s *blockSpool) notify()                 { close(s.wake); s.wake = make(chan struct{}) }
+func (s *blockSpool) path(index int64) string { return filepath.Join(s.dir, fmt.Sprintf("%d", index)) }
+func addSpan(spans []byteSpan, next byteSpan) ([]byteSpan, int64) {
+	before := int64(0)
+	for _, span := range spans {
+		before += span.end - span.start
+	}
+	out := make([]byteSpan, 0, len(spans)+1)
+	for _, span := range spans {
+		if span.end < next.start {
+			out = append(out, span)
+		} else if next.end < span.start {
+			out = append(out, next)
+			next = span
+		} else {
+			next.start = min(next.start, span.start)
+			next.end = max(next.end, span.end)
+		}
+	}
+	out = append(out, next)
+	after := int64(0)
+	for _, span := range out {
+		after += span.end - span.start
+	}
+	return out, after - before
+}
+func (s *blockSpool) WriteAt(p []byte, offset int64) (int, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if offset < 0 {
+		return 0, errors.New("negative media offset")
+	}
+	if s.closed {
+		return 0, io.ErrClosedPipe
+	}
+	if s.err != nil {
+		return 0, s.err
+	}
+	written := 0
+	for len(p) > 0 {
+		index, local := offset/SpoolBlockSize, offset%SpoolBlockSize
+		count := int(min(int64(len(p)), SpoolBlockSize-local))
+		f, err := os.OpenFile(s.path(index), os.O_CREATE|os.O_WRONLY, 0600)
+		if err != nil {
+			s.fail(err)
+			s.notify()
+			return written, err
+		}
+		n, err := f.WriteAt(p[:count], local)
+		closeErr := f.Close()
+		if err == nil {
+			err = closeErr
+		}
+		if n > 0 {
+			s.blocks[index], _ = addSpan(s.blocks[index], byteSpan{local, local + int64(n)})
+			var added int64
+			s.seen, added = addSpan(s.seen, byteSpan{offset, offset + int64(n)})
+			s.downloaded += added
+			s.received += int64(n)
+			written += n
+			offset += int64(n)
+			p = p[n:]
+		}
+		if err == nil && n != count {
+			err = io.ErrShortWrite
+		}
+		if err != nil {
+			s.fail(err)
+			s.notify()
+			return written, err
+		}
+	}
+	s.notify()
+	return written, nil
+}
+func (s *blockSpool) readAt(p []byte, offset int64) (int, error, <-chan struct{}) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.closed {
+		return 0, io.ErrClosedPipe, nil
+	}
+	if s.err != nil {
+		return 0, s.err, nil
+	}
+	index, local := offset/SpoolBlockSize, offset%SpoolBlockSize
+	for _, span := range s.blocks[index] {
+		if span.start <= local && local < span.end {
+			f, err := os.Open(s.path(index))
+			if err != nil {
+				return 0, err, nil
+			}
+			n, err := f.ReadAt(p[:min(int64(len(p)), span.end-local)], local)
+			f.Close()
+			return n, err, nil
+		}
+	}
+	return 0, nil, s.wake
+}
+func (s *blockSpool) discard(index int64) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if err := os.Remove(s.path(index)); err != nil && !os.IsNotExist(err) {
+		return err
+	}
+	delete(s.blocks, index)
+	return nil
+}
+func (s *blockSpool) end(err error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.ended = true
+	if s.err == nil {
+		s.fail(err)
+	}
+	s.notify()
+}
+func (s *blockSpool) Close() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.closed = true
+	clear(s.blocks)
+	s.seen = nil
+	s.notify()
+	return s.owner.Remove(s.dir)
+}
+func (s *blockSpool) progress() (int64, int64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.downloaded, s.received
+}
+
+// DiskInput consumes a sequential producer without retaining its payload in RAM.
+// Writes acknowledge durable-to-the-OS chunks, allowing both SABR tracks to drain
+// even while FFmpeg is queued or currently reading only the other track.
+type DiskInput struct {
+	ctx             context.Context
+	spool           *blockSpool
+	writeMu, readMu sync.Mutex
+	written, read   int64
+}
+
+func NewDiskInput(ctx context.Context, root string, owners ...*tempfiles.Scope) (*DiskInput, error) {
+	spool, err := newBlockSpool(root, owners...)
+	if err != nil {
+		return nil, err
+	}
+	return &DiskInput{ctx: ctx, spool: spool}, nil
+}
+func (s *DiskInput) Write(p []byte) (int, error) {
+	s.writeMu.Lock()
+	defer s.writeMu.Unlock()
+	if err := s.ctx.Err(); err != nil {
+		return 0, err
+	}
+	s.spool.mu.Lock()
+	ended := s.spool.ended
+	s.spool.mu.Unlock()
+	if ended {
+		return 0, io.ErrClosedPipe
+	}
+	n, err := s.spool.WriteAt(p, s.written)
+	s.written += int64(n)
+	return n, err
+}
+func (s *DiskInput) End(err error) { s.writeMu.Lock(); defer s.writeMu.Unlock(); s.spool.end(err) }
+func (s *DiskInput) Read(p []byte) (int, error) {
+	s.readMu.Lock()
+	defer s.readMu.Unlock()
+	if len(p) == 0 {
+		return 0, nil
+	}
+	for {
+		if err := s.ctx.Err(); err != nil {
+			return 0, err
+		}
+		n, err, wake := s.spool.readAt(p, s.read)
+		if n > 0 {
+			previous := s.read / SpoolBlockSize
+			s.read += int64(n)
+			if s.read/SpoolBlockSize > previous {
+				if e := s.spool.discard(previous); err == nil {
+					err = e
+				}
+			}
+			return n, err
+		}
+		if err != nil {
+			return 0, err
+		}
+		s.spool.mu.Lock()
+		ended := s.spool.ended
+		s.spool.mu.Unlock()
+		// End can race the missing-data check; recheck after observing it.
+		if ended {
+			n, err, _ = s.spool.readAt(p, s.read)
+			if n > 0 {
+				// Use the regular consumption path, including block eviction.
+				continue
+			}
+			if err != nil {
+				return 0, err
+			}
+			if err := s.spool.discard(s.read / SpoolBlockSize); err != nil {
+				return 0, err
+			}
+			return 0, io.EOF
+		}
+		select {
+		case <-wake:
+		case <-s.ctx.Done():
+			return 0, s.ctx.Err()
+		}
+	}
+}
+func (s *DiskInput) Close() error                 { return s.spool.Close() }
+func (*DiskInput) Seek(int64, int) (int64, error) { return 0, ErrSeek }
+func (*DiskInput) Size() int64                    { return -1 }
+func (*DiskInput) Seekable() bool                 { return false }
+func (s *DiskInput) Progress() (int64, int64)     { return s.spool.progress() }
+
+// fail is called with mu held. The first error wakes a queued FFmpeg job.
+func (s *blockSpool) fail(err error) {
+	if err == nil || s.err != nil {
+		return
+	}
+	s.err = err
+	select {
+	case s.failures <- err:
+	default:
+	}
+}
+func (s *DiskInput) Failures() <-chan error { return s.spool.failures }

pkg/download/engine/ffmpeg/spool_test.go

@@ -0,0 +1,101 @@
+package ffmpeg
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"os"
+	"testing"
+)
+
+func TestSpoolSparseCrossBlock(t *testing.T) {
+	s, err := newBlockSpool(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	offset := 7*SpoolBlockSize + 123
+	data := bytes.Repeat([]byte{91}, int(3*SpoolBlockSize+37))
+	if n, err := s.WriteAt(data, offset); err != nil || n != len(data) {
+		t.Fatal(n, err)
+	}
+	if n, err, wake := s.readAt(make([]byte, 16), 7*SpoolBlockSize); n != 0 || err != nil || wake == nil {
+		t.Fatal("sparse hole was exposed", n, err)
+	}
+	got := make([]byte, len(data))
+	pos := 0
+	for pos < len(got) {
+		n, err, _ := s.readAt(got[pos:], offset+int64(pos))
+		if err != nil || n == 0 {
+			t.Fatal(n, err)
+		}
+		pos += n
+	}
+	if !bytes.Equal(data, got) {
+		t.Fatal("cross-block contents differ")
+	}
+	if err := s.discard(8); err != nil {
+		t.Fatal(err)
+	}
+	if n, _, wake := s.readAt(got[:1], 8*SpoolBlockSize); n != 0 || wake == nil {
+		t.Fatal("deleted block remained readable")
+	}
+	if _, err := s.WriteAt(data, offset); err != nil {
+		t.Fatal(err)
+	}
+	unique, received := s.progress()
+	if unique != int64(len(data)) || received != 2*unique {
+		t.Fatal(unique, received)
+	}
+}
+func TestDiskInputDrainAndCleanup(t *testing.T) {
+	s, err := NewDiskInput(context.Background(), t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	data := bytes.Repeat([]byte{33}, int(3*SpoolBlockSize+99))
+	if _, err := s.Write(data); err != nil {
+		t.Fatal(err)
+	}
+	s.End(nil)
+	got, err := io.ReadAll(s)
+	if err != nil || !bytes.Equal(data, got) {
+		t.Fatal("drain", err)
+	}
+	files, err := os.ReadDir(s.spool.dir)
+	if err != nil || len(files) != 0 {
+		t.Fatal("consumed blocks retained", len(files), err)
+	}
+	if err := s.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := os.Stat(s.spool.dir); !os.IsNotExist(err) {
+		t.Fatal("spool not removed", err)
+	}
+}
+func TestDiskInputCancelAndWriteFailure(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	s, err := NewDiskInput(ctx, t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	cancel()
+	if _, err := s.Read(make([]byte, 1)); err != context.Canceled {
+		t.Fatal(err)
+	}
+	broken, err := NewDiskInput(context.Background(), t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer broken.Close()
+	if err := os.RemoveAll(broken.spool.dir); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := broken.Write([]byte{1}); err == nil {
+		t.Fatal("missing disk error")
+	}
+	if _, err := broken.Read(make([]byte, 1)); err == nil {
+		t.Fatal("disk error not propagated")
+	}
+}

pkg/download/engine/ffmpeg/stream.go

@@ -1,163 +0,0 @@
-package ffmpeg
-
-import (
-	"context"
-	"errors"
-	"io"
-	"sync"
-)
-
-// A bounded memory ring lets both JS inputs drain independently, including when
-// they share a SABR producer. It does not make a sequential input seekable.
-// A full ring waits for consumption or cancellation. Producers must propagate
-// backpressure and allow each track to make progress independently.
-const StreamBufferLimit = 32 * 1024 * 1024
-
-type StreamInput struct {
-	ctx           context.Context
-	mu            sync.Mutex
-	buffer        []byte
-	read, written int64
-	ended, closed bool
-	err           error
-	wake          chan struct{}
-	space         chan struct{}
-}
-
-func NewStreamInput(ctx context.Context) *StreamInput {
-	return &StreamInput{ctx: ctx, wake: make(chan struct{}, 1), space: make(chan struct{}, 1)}
-}
-
-func (s *StreamInput) Write(p []byte) (int, error) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if len(p) > StreamBufferLimit {
-		return 0, errors.New("ffmpeg: input chunk exceeds buffer capacity")
-	}
-	for int64(len(p)) > StreamBufferLimit-(s.written-s.read) && !s.closed && !s.ended {
-		s.mu.Unlock()
-		var err error
-		select {
-		case <-s.space:
-		case <-s.ctx.Done():
-			err = s.ctx.Err()
-		}
-		s.mu.Lock()
-		if err != nil {
-			return 0, err
-		}
-	}
-	if s.closed || s.ended {
-		return 0, io.ErrClosedPipe
-	}
-	if err := s.ctx.Err(); err != nil {
-		return 0, err
-	}
-	if len(p) == 0 {
-		return 0, nil
-	}
-	// Grow only as needed. Rebase unread bytes when the ring grows.
-	need := int(s.written-s.read) + len(p)
-	if need > len(s.buffer) {
-		capacity := min(StreamBufferLimit, max(256*1024, max(need, len(s.buffer)*2)))
-		buffer := make([]byte, capacity)
-		unread := int(s.written - s.read)
-		if unread > 0 {
-			off := int(s.read % int64(len(s.buffer)))
-			first := min(unread, len(s.buffer)-off)
-			copy(buffer, s.buffer[off:off+first])
-			copy(buffer[first:], s.buffer[:unread-first])
-		}
-		s.buffer = buffer
-		s.read, s.written = 0, int64(unread)
-	}
-	off := int(s.written % int64(len(s.buffer)))
-	first := min(len(p), len(s.buffer)-off)
-	copy(s.buffer[off:], p[:first])
-	copy(s.buffer, p[first:])
-	s.written += int64(len(p))
-	select {
-	case s.wake <- struct{}{}:
-	default:
-	}
-	return len(p), nil
-}
-
-func (s *StreamInput) End(err error) {
-	s.mu.Lock()
-	s.ended = true
-	if s.err == nil {
-		s.err = err
-	}
-	s.mu.Unlock()
-	select {
-	case s.space <- struct{}{}:
-	default:
-	}
-	select {
-	case s.wake <- struct{}{}:
-	default:
-	}
-}
-
-func (s *StreamInput) Read(p []byte) (int, error) {
-	if len(p) == 0 {
-		return 0, nil
-	}
-	for {
-		if err := s.ctx.Err(); err != nil {
-			return 0, err
-		}
-		s.mu.Lock()
-		if s.closed {
-			s.mu.Unlock()
-			return 0, io.ErrClosedPipe
-		}
-		if s.err != nil {
-			err := s.err
-			s.mu.Unlock()
-			return 0, err
-		}
-		if s.read < s.written {
-			off := s.read % int64(len(s.buffer))
-			n := min(int64(len(p)), s.written-s.read, int64(len(s.buffer))-off)
-			count := copy(p, s.buffer[off:off+n])
-			s.read += int64(count)
-			s.mu.Unlock()
-			select {
-			case s.space <- struct{}{}:
-			default:
-			}
-			return count, nil
-		}
-		ended := s.ended
-		s.mu.Unlock()
-		if ended {
-			return 0, io.EOF
-		}
-		select {
-		case <-s.wake:
-		case <-s.ctx.Done():
-			return 0, s.ctx.Err()
-		}
-	}
-}
-
-func (s *StreamInput) Close() error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	s.closed = true
-	select {
-	case s.space <- struct{}{}:
-	default:
-	}
-	select {
-	case s.wake <- struct{}{}:
-	default:
-	}
-	s.buffer = nil
-	return nil
-}
-func (*StreamInput) Seek(int64, int) (int64, error) { return 0, ErrSeek }
-func (*StreamInput) Size() int64                    { return -1 }
-func (*StreamInput) Seekable() bool                 { return false }

pkg/download/engine/ffmpeg_integration_test.go

@@ -4,12 +4,14 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"github.com/GopeedLab/gopeed/internal/production"
 	media "github.com/GopeedLab/gopeed/pkg/download/engine/ffmpeg"
 	"github.com/GopeedLab/gopeed/pkg/download/engine/inject/stream"
 	"io"
 	"net/http"
 	"net/http/httptest"
 	"os"
+	"path/filepath"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -198,7 +200,7 @@ func TestFFmpegEngineCloseCancelsRequest(t *testing.T) {
 	}
 }
 
-func TestFFmpegInputFactoriesWaitForCapacity(t *testing.T) {
+func TestFFmpegInputsStartBeforeCapacity(t *testing.T) {
 	for _, mode := range []string{"factory", "http", "cancel"} {
 		t.Run(mode, func(t *testing.T) {
 			release1, _ := media.Acquire(context.Background())
@@ -226,8 +228,12 @@ func TestFFmpegInputFactoriesWaitForCapacity(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			if fmt.Sprint(result) != "0" || requests.Load() != 0 {
-				t.Fatalf("queued inputs started: %v, requests=%d", result, requests.Load())
+			if mode == "http" {
+				if requests.Load() != 1 {
+					t.Fatalf("queued HTTP did not start: %d", requests.Load())
+				}
+			} else if fmt.Sprint(result) != "1" {
+				t.Fatalf("queued factory did not start: %v", result)
 			}
 			if mode == "cancel" {
 				if _, err = e.RunString(`work`); err != nil {
@@ -245,11 +251,11 @@ func TestFFmpegInputFactoriesWaitForCapacity(t *testing.T) {
 			if mode == "factory" && fmt.Sprint(calls) != "1" {
 				t.Fatal(calls)
 			}
-			if mode == "cancel" && fmt.Sprint(calls) != "0" {
+			if mode == "cancel" && fmt.Sprint(calls) != "1" {
 				t.Fatal("cancelled factory ran")
 			}
 			if mode == "http" && requests.Load() != 1 {
-				t.Fatal("HTTP did not start after admission")
+				t.Fatal("HTTP was restarted unnecessarily")
 			}
 			// Failure/cancellation must return the acquired slot even with the other held.
 			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
@@ -319,3 +325,109 @@ func TestFFmpegCancelPendingFactoryCleansLateInputs(t *testing.T) {
 	}
 	second()
 }
+
+func TestQueuedMergePrefetchFailureAndCancel(t *testing.T) {
+	for _, mode := range []string{"failure", "cancel"} {
+		t.Run(mode, func(t *testing.T) {
+			release1, _ := media.Acquire(context.Background())
+			defer release1()
+			release2, _ := media.Acquire(context.Background())
+			defer release2()
+			second := make(chan struct{})
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if r.Header.Get("Range") == "bytes=0-1048575" {
+					w.Header().Set("Content-Range", "bytes 0-1048575/33554432")
+					w.Header().Set("Content-Length", "1048576")
+					w.WriteHeader(206)
+					w.Write(make([]byte, 1048576))
+					return
+				}
+				close(second)
+				if mode == "cancel" {
+					<-r.Context().Done()
+					return
+				}
+				http.Error(w, "upstream failed", 503)
+			}))
+			defer srv.Close()
+			root := t.TempDir()
+			e := NewEngine(&Config{TempDir: root})
+			defer e.Close()
+			e.Runtime.Set("sourceURL", srv.URL)
+			_, err := e.RunString(`
+   globalThis.abortJob=new AbortController();
+   globalThis.result=__gopeed_ffmpeg.merge({signal:abortJob.signal,inputs:()=>({video:{url:sourceURL},audio:new ReadableStream({start(c){c.close();}})})})
+    .getReader().read().then(()=>'',e=>String(e.message||e));
+   'started';`)
+			if err != nil {
+				t.Fatal(err)
+			}
+			select {
+			case <-second:
+			case <-time.After(3 * time.Second):
+				t.Fatal("queued input did not download")
+			}
+			if mode == "cancel" {
+				if _, err := e.RunString(`abortJob.abort()`); err != nil {
+					t.Fatal(err)
+				}
+			}
+			result, err := e.RunString(`Promise.race([result,new Promise(r=>setTimeout(()=>r('test timeout'),3000))])`)
+			if err != nil || fmt.Sprint(result) == "test timeout" || fmt.Sprint(result) == "" {
+				t.Fatal(result, err)
+			}
+			if mode == "failure" && !strings.Contains(fmt.Sprint(result), "503") {
+				t.Fatal("lost upstream failure", result)
+			}
+			deadline := time.Now().Add(3 * time.Second)
+			for {
+				entries, err := os.ReadDir(root)
+				if err == nil && len(entries) == 0 {
+					break
+				}
+				if time.Now().After(deadline) {
+					t.Fatal("queued input files leaked", entries, err)
+				}
+				time.Sleep(10 * time.Millisecond)
+			}
+		})
+	}
+}
+
+func TestFFmpegUnavailableTempDirReportsBlobError(t *testing.T) {
+	root := filepath.Join(t.TempDir(), "blocked")
+	if err := os.WriteFile(root, []byte("not a directory"), 0600); err != nil {
+		t.Fatal(err)
+	}
+	var open stream.ObjectURLOpener
+	e := NewEngine(&Config{TempDir: root, StreamConfig: &stream.Config{CreateObjectURL: func(_ *stream.ObjectURLOptions, opener stream.ObjectURLOpener) (string, error) {
+		open = opener
+		return "blob:test", nil
+	}}})
+	defer e.Close()
+	_, err := e.RunString(`__gopeed_blob_create_object_url(()=>__gopeed_ffmpeg.merge({inputs:()=>({video:new ReadableStream(),audio:new ReadableStream()})}),{})`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	reader, err := open(ctx, stream.ObjectURLOpenRequest{End: -1})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer reader.Close()
+	produced := reader.(production.Reader)
+	if err := produced.WaitProduction(ctx); err != nil {
+		t.Fatal(err)
+	}
+	source := produced.Production()
+	if source == nil {
+		t.Fatal("missing producer")
+	}
+	if _, err := io.ReadAll(reader); err == nil || !strings.Contains(err.Error(), "blocked") {
+		t.Fatal("missing useful disk error", err)
+	}
+	if downloaded, received := source.Progress(); downloaded != 0 || received != 0 {
+		t.Fatal(downloaded, received)
+	}
+}

pkg/download/engine/inject/ffmpeg/ffmpeg.js

@@ -1,7 +1,6 @@
 (function () {
   const create = __gopeed_ffmpeg_create;
   const read = __gopeed_ffmpeg_read;
-  const acquire = __gopeed_ffmpeg_acquire;
   const start = __gopeed_ffmpeg_start;
   const push = __gopeed_ffmpeg_push;
   const end = __gopeed_ffmpeg_end;
@@ -37,7 +36,8 @@
     const format = options.format === undefined ? "mp4" : options.format;
     if (typeof format !== "string") throw new TypeError("format must be a string");
     const signal = options.signal;
-    let id, controller, stopped = false;
+    const id = create({ format, args });
+    let controller, initialized = false, stopped = false;
     const readers = [];
     const lifetime = new AbortController();
     let ownedInputs;
@@ -91,7 +91,7 @@
         }
       }
     }
-    return new ReadableStream({
+    const output = new ReadableStream({
       start(c) {
         controller = c;
         active.add(shutdown);
@@ -103,9 +103,8 @@
       async pull(c) {
         if (stopped) return;
         try {
-          if (!id) {
-            id = create({ format, args });
-            await acquire(id);
+          if (!initialized) {
+            initialized = true;
             if (stopped) return;
             if (factory) {
               const inputs = await factory({ signal: lifetime.signal });
@@ -129,11 +128,17 @@
           if (chunk === null) { stop(); c.close(); }
           else c.enqueue(new Uint8Array(chunk));
         } catch (error) {
-          if (!stopped) { c.error(error); stop(error); }
+          if (!stopped) {
+            const failure = typeof MessageError === "function" && !(error instanceof MessageError)
+              ? new MessageError(String(error && error.message || error)) : error;
+            c.error(failure); stop(failure);
+          }
         }
       },
       cancel(reason) { stop(reason); }
-    });
+    }, { highWaterMark: 0 });
+    if (globalThis.__gopeed_bind_producer) globalThis.__gopeed_bind_producer(output, id);
+    return output;
   }
-  globalThis.__gopeed_ffmpeg = Object.freeze({ merge, supportsInputFactory: true });
+  globalThis.__gopeed_ffmpeg = Object.freeze({ merge, supportsInputFactory: true, supportsProducerProgress: true });
 })();

pkg/download/engine/inject/ffmpeg/module.go

@@ -13,6 +13,8 @@ import (
 	"sync/atomic"
 
 	"github.com/GopeedLab/gopeed/internal/httpclient"
+	"github.com/GopeedLab/gopeed/internal/production"
+	"github.com/GopeedLab/gopeed/internal/tempfiles"
 	media "github.com/GopeedLab/gopeed/pkg/download/engine/ffmpeg"
 	"github.com/dop251/goja"
 	"github.com/dop251/goja_nodejs/eventloop"
@@ -22,6 +24,9 @@ import (
 var script string
 
 type Config struct {
+	TempFiles        *tempfiles.Scope
+	Producers        *production.Registry
+	TempDir          string
 	DefaultUserAgent *string
 	ProxyHandler     func(*http.Request) (*url.URL, error)
 	RegisterCleanup  func(func())
@@ -33,16 +38,29 @@ type options struct {
 	Args   []string          `json:"args"`
 }
 type job struct {
+	inputs  [2]media.Input
 	ctx     context.Context
 	cancel  context.CancelFunc
 	output  *io.PipeReader
-	streams [2]*media.StreamInput
+	streams [2]*media.DiskInput
 	mu      sync.Mutex
-	ready   chan struct{}
 	start   chan options
 	started bool
 }
 
+func (j *job) Progress() (downloaded, received int64) {
+	j.mu.Lock()
+	defer j.mu.Unlock()
+	for _, input := range j.inputs {
+		if p, ok := input.(production.Source); ok {
+			d, r := p.Progress()
+			downloaded += d
+			received += r
+		}
+	}
+	return
+}
+
 func (j *job) stop() {
 	j.cancel()
 	j.output.CloseWithError(context.Canceled)
@@ -59,6 +77,10 @@ func Enable(vm *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error {
 	if cfg == nil {
 		cfg = &Config{}
 	}
+	spoolDir := cfg.TempDir
+	if cfg.Producers == nil {
+		cfg.Producers = &production.Registry{}
+	}
 	client, err := httpclient.NewClient(httpclient.Options{Transport: httpclient.TransportOptions{Proxy: cfg.ProxyHandler}})
 	if err != nil {
 		return err
@@ -71,6 +93,7 @@ func Enable(vm *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error {
 		mu.Lock()
 		j := jobs[id]
 		delete(jobs, id)
+		cfg.Producers.Remove(id)
 		mu.Unlock()
 		if j != nil {
 			j.stop()
@@ -102,15 +125,14 @@ func Enable(vm *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error {
 		if err := vm.ExportTo(call.Argument(0), &opts); err != nil {
 			panic(vm.NewGoError(err))
 		}
-		if _, err := media.Arguments(opts.Format, opts.Args); err != nil {
-			panic(vm.NewGoError(err))
-		}
-		ctx, cancel := context.WithCancel(context.Background())
+		ctx, cancelCause := context.WithCancelCause(context.Background())
+		cancel := func() { cancelCause(context.Canceled) }
 		r, w := io.Pipe()
-		j := &job{ctx: ctx, cancel: cancel, output: r, ready: make(chan struct{}), start: make(chan options, 1)}
+		j := &job{ctx: ctx, cancel: cancel, output: r, start: make(chan options, 1)}
 		id := fmt.Sprintf("ffmpeg-%d", next.Add(1))
 		mu.Lock()
 		jobs[id] = j
+		cfg.Producers.Set(id, j)
 		mu.Unlock()
 		go func() {
 			var inputs [2]media.Input
@@ -119,6 +141,9 @@ func Enable(vm *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error {
 				if v := recover(); v != nil {
 					runErr = fmt.Errorf("ffmpeg panic: %v", v)
 				}
+				if errors.Is(runErr, context.Canceled) {
+					runErr = context.Cause(ctx)
+				}
 				cancel()
 				for _, in := range inputs {
 					if in != nil {
@@ -127,13 +152,6 @@ func Enable(vm *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error {
 				}
 				w.CloseWithError(runErr)
 			}()
-			release, err := media.Acquire(ctx)
-			if err != nil {
-				runErr = err
-				return
-			}
-			defer release()
-			close(j.ready)
 			var opts options
 			select {
 			case opts = <-j.start:
@@ -145,47 +163,36 @@ func Enable(vm *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error {
 				if source == nil {
 					inputs[i] = j.streams[i]
 				} else {
-					inputs[i], runErr = media.OpenHTTP(ctx, client, *source)
+					inputs[i], runErr = media.OpenSpoolingHTTP(ctx, client, *source, spoolDir, cfg.TempFiles)
 					if runErr != nil {
 						return
 					}
 				}
+				j.mu.Lock()
+				j.inputs[i] = inputs[i]
+				j.mu.Unlock()
+				if input, ok := inputs[i].(interface{ Failures() <-chan error }); ok {
+					go func() {
+						select {
+						case err := <-input.Failures():
+							cancelCause(err)
+						case <-ctx.Done():
+						}
+					}()
+				}
 			}
+			release, err := media.Acquire(ctx)
+			if err != nil {
+				runErr = err
+				return
+			}
+			defer release()
 			runErr = media.RunAcquired(ctx, inputs[0], inputs[1], w, opts.Format, opts.Args)
 		}()
 		return vm.ToValue(id)
 	}); err != nil {
 		return err
 	}
-	if err := vm.Set("__gopeed_ffmpeg_acquire", func(id string) goja.Value {
-		promise, resolve, reject := vm.NewPromise()
-		j := get(id)
-		go func() {
-			var err error
-			if j == nil {
-				err = io.ErrClosedPipe
-			} else {
-				select {
-				case <-j.ready:
-				case <-j.ctx.Done():
-					err = j.ctx.Err()
-				}
-				if j.ctx.Err() != nil {
-					err = j.ctx.Err()
-				}
-			}
-			loop.RunOnLoop(func(vm *goja.Runtime) {
-				if err != nil {
-					reject(vm.NewGoError(err))
-				} else {
-					resolve(true)
-				}
-			})
-		}()
-		return vm.ToValue(promise)
-	}); err != nil {
-		return err
-	}
 	if err := vm.Set("__gopeed_ffmpeg_start", func(call goja.FunctionCall) goja.Value {
 		id := call.Argument(0).String()
 		j := get(id)
@@ -226,7 +233,13 @@ func Enable(vm *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error {
 		j.started = true
 		for i, source := range []*media.HTTPSource{opts.Video, opts.Audio} {
 			if source == nil {
-				j.streams[i] = media.NewStreamInput(j.ctx)
+				var err error
+				j.streams[i], err = media.NewDiskInput(j.ctx, spoolDir, cfg.TempFiles)
+				if err != nil {
+					j.cancel()
+					panic(vm.NewGoError(err))
+				}
+				j.inputs[i] = j.streams[i]
 			}
 		}
 		j.start <- opts

pkg/download/engine/inject/stream/module.go

@@ -9,6 +9,7 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"github.com/GopeedLab/gopeed/internal/production"
 	"github.com/dop251/goja"
 	"github.com/dop251/goja_nodejs/eventloop"
 )
@@ -17,6 +18,7 @@ import (
 var script string
 
 type Config struct {
+	Producers       *production.Registry
 	CreateObjectURL func(opts *ObjectURLOptions, open ObjectURLOpener) (string, error)
 	RevokeObjectURL func(url string) error
 }
@@ -39,6 +41,23 @@ func Enable(runtime *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error
 		cfg = &Config{}
 	}
 	blobPipes := newBlobPipeRegistry()
+	if err := runtime.Set("__gopeed_blob_pipe_bind", func(pipeID, producerID string) {
+		var source production.Source
+		if cfg.Producers != nil {
+			source = cfg.Producers.Get(producerID)
+		}
+		blobPipes.mu.Lock()
+		pipe := blobPipes.pipes[pipeID]
+		blobPipes.mu.Unlock()
+		if pipe != nil {
+			pipe.mu.Lock()
+			pipe.producer = source
+			pipe.mu.Unlock()
+			pipe.readyOnce.Do(func() { close(pipe.ready) })
+		}
+	}); err != nil {
+		return err
+	}
 	if err := runtime.Set("__gopeed_create_blob_object_url", func(call goja.FunctionCall) goja.Value {
 		if cfg.CreateObjectURL == nil {
 			panic(runtime.NewGoError(fmt.Errorf("blob object url handler not configured")))
@@ -189,12 +208,15 @@ type blobPipeItem struct {
 }
 
 type blobPipe struct {
-	id       string
-	loop     *eventloop.EventLoop
-	registry *blobPipeRegistry
-	ctx      context.Context
-	cancel   context.CancelFunc
-	ch       chan blobPipeItem
+	ready     chan struct{}
+	readyOnce sync.Once
+	producer  production.Source
+	id        string
+	loop      *eventloop.EventLoop
+	registry  *blobPipeRegistry
+	ctx       context.Context
+	cancel    context.CancelFunc
+	ch        chan blobPipeItem
 
 	mu     sync.Mutex
 	buf    []byte
@@ -205,6 +227,7 @@ type blobPipe struct {
 func newBlobPipe(ctx context.Context, loop *eventloop.EventLoop, registry *blobPipeRegistry) *blobPipe {
 	pipeCtx, cancel := context.WithCancel(ctx)
 	return &blobPipe{
+		ready:    make(chan struct{}),
 		id:       registry.NewID(),
 		loop:     loop,
 		registry: registry,
@@ -228,6 +251,7 @@ func (p *blobPipe) Push(chunk []byte) bool {
 }
 
 func (p *blobPipe) CloseWithError(err error) {
+	p.readyOnce.Do(func() { close(p.ready) })
 	p.once.Do(func() {
 		p.registry.Remove(p.id)
 		select {
@@ -480,3 +504,20 @@ func exportJSError(value goja.Value) error {
 	}
 	return errors.New(stack)
 }
+
+func (p *blobPipe) Production() production.Source {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return p.producer
+}
+
+func (p *blobPipe) WaitProduction(ctx context.Context) error {
+	select {
+	case <-p.ready:
+		return nil
+	case <-p.ctx.Done():
+		return p.ctx.Err()
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}

pkg/download/engine/inject/stream/stream.js

@@ -734,6 +734,8 @@
     finishBlobPipeReader(state, true, reason).catch(function () {});
   };
 
+  const producers = new WeakMap();
+  globalThis.__gopeed_bind_producer = (stream, id) => producers.set(stream, id);
   globalThis.__gopeed_blob_pipe_source = function (openReadable, request, pipeId) {
     if (typeof openReadable !== "function") {
       throw new TypeError("blob opener must be callable");
@@ -752,6 +754,8 @@
       try {
         const source = await openReadable(normalizeOpenRequest(request));
         state.openSettled = true;
+        const producer = producers.get(source);
+        globalThis.__gopeed_blob_pipe_bind(pipeId, producer || "");
         state.reader = toReadableStreamReader(source, "gopeed.runtime.blob.createObjectURL opener function");
         if (state.cancelRequested) {
           await finishBlobPipeReader(state, true, state.cancelReason);
@@ -777,7 +781,7 @@
         }
       } catch (error) {
         if (!state.cancelRequested) {
-          globalThis.__gopeed_blob_pipe_error(pipeId, error && error.stack ? error.stack : String(error));
+          globalThis.__gopeed_blob_pipe_error(pipeId, error && (error.stack || error.message) ? (error.stack || error.message) : String(error));
         }
       } finally {
         state.openSettled = true;

pkg/download/extension.go

@@ -26,7 +26,6 @@ import (
 var (
 	gitSuffix = ".git"
 
-	tempExtensionsDir   = ".extensions"
 	extensionsDir       = "extensions"
 	extensionIgnoreDirs = []string{gitSuffix, "node_modules"}
 
@@ -213,11 +212,14 @@ func (d *Downloader) fetchExtensionByGit(url string, handler func(tempExtPath st
 
 	// Use unique suffix to avoid concurrent conflicts when multiple git operations
 	// target the same extension (e.g., install and update check happening simultaneously)
-	tempExtDir := filepath.Join(d.cfg.StorageDir, tempExtensionsDir, fmt.Sprintf("%s_%d", projectPath, time.Now().UnixNano()))
-	if err := os.MkdirAll(tempExtDir, 0755); err != nil {
+	tempExtDir, err := os.MkdirTemp(d.cfg.TempDir, "gopeed-extension-")
+	if err != nil {
 		return nil, err
 	}
-	defer os.RemoveAll(tempExtDir)
+	if err := d.tempFiles.Track(tempExtDir); err != nil {
+		return nil, err
+	}
+	defer d.tempFiles.Remove(tempExtDir)
 
 	proxyOptions := transport.ProxyOptions{}
 	proxyUrl := d.cfg.DownloaderStoreConfig.Proxy.ToUrl()

pkg/download/extension_engine_session.go

@@ -127,6 +127,8 @@ func (d *Downloader) newExtensionEngine() (*engine.Engine, *engineSession) {
 	d.getProtocolConfig("http", &httpConfig)
 	userAgent := strings.TrimSpace(httpConfig.UserAgent)
 	e := engine.NewEngine(&engine.Config{
+		TempDir:       d.cfg.TempDir,
+		TempFiles:     d.tempFiles,
 		HTTPUserAgent: &userAgent,
 		ProxyConfig:   d.cfg.Proxy,
 		StreamConfig:  engineCfg,

pkg/download/media_spooling_test.go

@@ -0,0 +1,150 @@
+package download
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/GopeedLab/gopeed/pkg/base"
+	media "github.com/GopeedLab/gopeed/pkg/download/engine/ffmpeg"
+)
+
+func TestQueuedMergeReportsInputProgress(t *testing.T) {
+	for _, mode := range []string{"streams", "http"} {
+		t.Run(mode, func(t *testing.T) {
+			release1, err := media.Acquire(context.Background())
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer release1()
+			release2, err := media.Acquire(context.Background())
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer release2()
+			video, err := os.ReadFile("engine/testdata/ffmpeg/video.mp4")
+			if err != nil {
+				t.Fatal(err)
+			}
+			audio, err := os.ReadFile("engine/testdata/ffmpeg/audio.mp4")
+			if err != nil {
+				t.Fatal(err)
+			}
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				b := video
+				if r.URL.Path == "/audio" {
+					b = audio
+				}
+				http.ServeContent(w, r, "media", time.Time{}, bytes.NewReader(b))
+			}))
+			defer server.Close()
+			d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir(), RefreshInterval: 50})
+			if err := d.Setup(); err != nil {
+				t.Fatal(err)
+			}
+			defer d.Clear()
+			e, _ := d.newExtensionEngine()
+			defer e.Close()
+			e.Runtime.Set("videoBytes", e.Runtime.NewArrayBuffer(video))
+			e.Runtime.Set("audioBytes", e.Runtime.NewArrayBuffer(audio))
+			e.Runtime.Set("mode", mode)
+			e.Runtime.Set("sourceURL", server.URL)
+			result, err := e.RunString(`__gopeed_blob_create_object_url(()=>{
+   const source=b=>new ReadableStream({start(c){c.enqueue(new Uint8Array(b));c.close();}});
+   return __gopeed_ffmpeg.merge(mode==='http'?{video:{url:sourceURL+'/video'},audio:{url:sourceURL+'/audio'}}:
+    {inputs:()=>({video:source(videoBytes),audio:source(audioBytes)})});
+  },{contentType:'video/mp4'})`)
+			if err != nil {
+				t.Fatal(err)
+			}
+			output := t.TempDir()
+			id, err := d.CreateDirect(&base.Request{URL: fmt.Sprint(result)}, &base.Options{Path: output, Name: "merged.mp4"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			deadline := time.Now().Add(5 * time.Second)
+			for {
+				task := d.GetTask(id)
+				task.lock.Lock()
+				task.statusLock.Lock()
+				status, downloaded := task.Status, task.Progress.Downloaded
+				total := int64(0)
+				if task.Meta.Res != nil {
+					total = task.Meta.Res.Size
+				}
+				task.statusLock.Unlock()
+				task.lock.Unlock()
+				if downloaded == int64(len(video)+len(audio)) {
+					if status != base.DownloadStatusRunning || total != 0 {
+						t.Fatalf("queued task: status=%v total=%d", status, total)
+					}
+					break
+				}
+				if time.Now().After(deadline) {
+					t.Fatalf("input progress not reported: status=%v bytes=%d", status, downloaded)
+				}
+				time.Sleep(20 * time.Millisecond)
+			}
+			task := d.GetTask(id)
+			if mode == "streams" {
+				time.Sleep(16 * time.Second)
+			} else {
+				time.Sleep(200 * time.Millisecond)
+			}
+			task.lock.Lock()
+			task.statusLock.Lock()
+			speed := task.Progress.Speed
+			task.statusLock.Unlock()
+			task.lock.Unlock()
+			// Once the upstream is drained, waiting for FFmpeg is still a running task.
+			if speed < 0 || (mode == "streams" && speed != 0) {
+				t.Fatal("negative queued speed", speed)
+			}
+			release1()
+			deadline = time.Now().Add(100 * time.Second)
+			for d.taskStatus(task) != base.DownloadStatusDone {
+				if time.Now().After(deadline) {
+					t.Fatal("merge did not complete")
+				}
+				time.Sleep(20 * time.Millisecond)
+			}
+			// Done is published before extension hooks finish. Let the watcher
+			// release its Blob/session before tearing down the downloader.
+			deadline = time.Now().Add(5 * time.Second)
+			for {
+				if _, active := d.watchedTasks.Load(id); !active {
+					break
+				}
+				if time.Now().After(deadline) {
+					t.Fatal("completion hooks did not finish")
+				}
+				time.Sleep(10 * time.Millisecond)
+			}
+			body, err := os.ReadFile(filepath.Join(output, "merged.mp4"))
+			if err != nil {
+				t.Fatal(err)
+			}
+			if !bytes.Contains(body, []byte("moof")) {
+				t.Fatal("invalid merged output")
+			}
+			task.lock.Lock()
+			task.statusLock.Lock()
+			downloaded, total := task.Progress.Downloaded, task.Meta.Res.Size
+			task.statusLock.Unlock()
+			task.lock.Unlock()
+			if downloaded != int64(len(body)) || total != int64(len(body)) {
+				t.Fatal("completion did not use output size", downloaded, total, len(body))
+			}
+			files, err := os.ReadDir(d.cfg.TempDir)
+			if err != nil || len(files) != 0 {
+				t.Fatal("temporary media leaked", files, err)
+			}
+		})
+	}
+}

pkg/download/model.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/GopeedLab/gopeed/internal/controller"
 	"github.com/GopeedLab/gopeed/internal/fetcher"
+	"github.com/GopeedLab/gopeed/internal/production"
 	"github.com/GopeedLab/gopeed/internal/protocol/bt"
 	"github.com/GopeedLab/gopeed/internal/protocol/ed2k"
 	"github.com/GopeedLab/gopeed/internal/protocol/hls"
@@ -32,6 +33,9 @@ type Task struct {
 	CreatedAt time.Time            `json:"createdAt"`
 	UpdatedAt time.Time            `json:"updatedAt"`
 
+	producer         production.Source
+	producerReceived int64
+
 	fetcherManager fetcher.FetcherManager
 	fetcher        fetcher.Fetcher
 	timer          *util.Timer
@@ -184,6 +188,7 @@ type DownloaderConfig struct {
 	RefreshInterval   int // RefreshInterval time duration to refresh task progress(ms)
 	Storage           Storage
 	StorageDir        string
+	TempDir           string
 	WhiteDownloadDirs []string
 	WebViewProvider   enginewebview.Provider
 

pkg/download/temp.go

@@ -0,0 +1,26 @@
+package download
+
+import (
+	"os"
+	"path/filepath"
+
+	"github.com/GopeedLab/gopeed/internal/tempfiles"
+)
+
+// TempDir may be shared with the OS or other applications. Initialization never
+// removes its contents; shutdown only removes paths created by this downloader.
+func (d *Downloader) setupTempDir() error {
+	if d.cfg.TempDir == "" {
+		d.cfg.TempDir = os.TempDir()
+	}
+	root, err := filepath.Abs(d.cfg.TempDir)
+	if err != nil {
+		return err
+	}
+	if err := os.MkdirAll(root, 0700); err != nil {
+		return err
+	}
+	d.cfg.TempDir = root
+	d.tempFiles = &tempfiles.Scope{}
+	return nil
+}

pkg/download/temp_test.go

@@ -0,0 +1,57 @@
+package download
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestTempDirInitialization(t *testing.T) {
+	for _, configured := range []bool{false, true} {
+		t.Run(map[bool]string{false: "system-default", true: "explicit"}[configured], func(t *testing.T) {
+			root := t.TempDir()
+			// os.TempDir uses different environment variables across supported platforms.
+			t.Setenv("TMPDIR", root)
+			t.Setenv("TMP", root)
+			t.Setenv("TEMP", root)
+			cfg := &DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir()}
+			if configured {
+				cfg.TempDir = root
+			}
+			existing := filepath.Join(root, "gopeed-media-existing")
+			if err := os.Mkdir(existing, 0700); err != nil {
+				t.Fatal(err)
+			}
+			marker := filepath.Join(existing, "keep")
+			if err := os.WriteFile(marker, []byte("other owner"), 0600); err != nil {
+				t.Fatal(err)
+			}
+			d := NewDownloader(cfg)
+			if err := d.Setup(); err != nil {
+				t.Fatal(err)
+			}
+			if d.cfg.TempDir != root {
+				t.Fatalf("TempDir=%q, want %q", d.cfg.TempDir, root)
+			}
+			if _, err := os.Stat(marker); err != nil {
+				t.Fatal("initialization removed existing data", err)
+			}
+			owned, err := os.MkdirTemp(d.cfg.TempDir, "gopeed-media-")
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err := d.tempFiles.Track(owned); err != nil {
+				t.Fatal(err)
+			}
+			if err := d.Close(); err != nil {
+				t.Fatal(err)
+			}
+			if _, err := os.Stat(owned); !os.IsNotExist(err) {
+				t.Fatal("shutdown retained owned data", err)
+			}
+			if _, err := os.Stat(marker); err != nil {
+				t.Fatal("shutdown removed another owner's data", err)
+			}
+		})
+	}
+}

pkg/rest/model/server.go

@@ -23,6 +23,7 @@ type StartConfig struct {
 	RefreshInterval   int                         `json:"refreshInterval"`
 	Storage           Storage                     `json:"storage"`
 	StorageDir        string                      `json:"storageDir"`
+	TempDir           string                      `json:"tempDir"`
 	WhiteDownloadDirs []string                    `json:"whiteDownloadDirs"`
 	ApiToken          string                      `json:"apiToken"`
 	DownloadConfig    *base.DownloaderStoreConfig `json:"downloadConfig"`

pkg/rest/server.go

@@ -403,6 +403,7 @@ func initializeCore(startCfg *model.StartConfig) error {
 		downloadCfg.Storage = download.NewMemStorage()
 	}
 	downloadCfg.StorageDir = startCfg.StorageDir
+	downloadCfg.TempDir = startCfg.TempDir
 	downloadCfg.Init()
 	Downloader = download.NewDownloader(downloadCfg)
 	if err := Downloader.Setup(); err != nil {

ui/flutter/lib/core/common/start_config.dart

@@ -23,6 +23,7 @@ class StartConfig {
   bool mcpEnable = false;
   late String storage;
   late String storageDir;
+  String? tempDir;
   late int refreshInterval;
   late String apiToken;
   WebViewRpcConfig? webViewRpcConfig;

ui/flutter/lib/core/common/start_config.g.dart

@@ -23,9 +23,10 @@ StartConfig _$StartConfigFromJson(Map<String, dynamic> json) => StartConfig()
   ..network = json['network'] as String
   ..address = json['address'] as String
   ..apiEnable = json['apiEnable'] as bool
-  ..mcpEnable = json['mcpEnable'] as bool? ?? false
+  ..mcpEnable = json['mcpEnable'] as bool
   ..storage = json['storage'] as String
   ..storageDir = json['storageDir'] as String
+  ..tempDir = json['tempDir'] as String?
   ..refreshInterval = (json['refreshInterval'] as num).toInt()
   ..apiToken = json['apiToken'] as String
   ..webViewRpcConfig = json['webViewRpcConfig'] == null
@@ -42,6 +43,7 @@ Map<String, dynamic> _$StartConfigToJson(StartConfig instance) =>
       'mcpEnable': instance.mcpEnable,
       'storage': instance.storage,
       'storageDir': instance.storageDir,
+      'tempDir': ?instance.tempDir,
       'refreshInterval': instance.refreshInterval,
       'apiToken': instance.apiToken,
       'webViewRpcConfig': ?instance.webViewRpcConfig,

ui/flutter/lib/core/entry/libgopeed_boot_native.dart

@@ -1,5 +1,7 @@
 import 'dart:async';
 
+import 'package:path_provider/path_provider.dart';
+
 import '../../util/util.dart';
 import '../common/libgopeed_channel.dart';
 import '../common/libgopeed_ffi.dart';
@@ -26,6 +28,9 @@ class LibgopeedBootNative implements LibgopeedBoot {
   Future<int> start(StartConfig cfg) async {
     cfg.storage = 'bolt';
     cfg.storageDir = Util.getStorageDir();
+    if (Util.isMobile()) {
+      cfg.tempDir = (await getTemporaryDirectory()).path;
+    }
     cfg.refreshInterval = 0;
     return _libgopeed.start(cfg);
   }