From e0ced260e2743a1b56aef9a303793bf687042d4d Mon Sep 17 00:00:00 2001 From: Levi Date: Wed, 16 Sep 2026 23:16:19 +0800 Subject: [PATCH 1/2] feat: spool media inputs to disk and report producer progress --- cmd/web/flags.go | 7 + cmd/web/main.go | 5 + internal/blob/registry.go | 44 +++ internal/controller/controller.go | 8 +- internal/production/progress.go | 35 +++ internal/protocol/http/fetcher.go | 41 ++- internal/tempfiles/scope.go | 63 ++++ internal/tempfiles/scope_test.go | 76 +++++ pkg/download/downloader.go | 63 +++- pkg/download/engine/engine.go | 15 +- pkg/download/engine/ffmpeg/README.md | 58 +++- pkg/download/engine/ffmpeg/http_spool.go | 248 +++++++++++++++ pkg/download/engine/ffmpeg/http_spool_test.go | 198 ++++++++++++ pkg/download/engine/ffmpeg/input.go | 29 +- pkg/download/engine/ffmpeg/input_test.go | 91 ------ pkg/download/engine/ffmpeg/runtime.go | 3 +- pkg/download/engine/ffmpeg/runtime_test.go | 12 +- pkg/download/engine/ffmpeg/spool.go | 284 ++++++++++++++++++ pkg/download/engine/ffmpeg/spool_test.go | 101 +++++++ pkg/download/engine/ffmpeg/stream.go | 163 ---------- .../engine/ffmpeg_integration_test.go | 122 +++++++- pkg/download/engine/inject/ffmpeg/ffmpeg.js | 23 +- pkg/download/engine/inject/ffmpeg/module.go | 103 ++++--- pkg/download/engine/inject/stream/module.go | 53 +++- pkg/download/engine/inject/stream/stream.js | 6 +- pkg/download/extension.go | 10 +- pkg/download/extension_engine_session.go | 2 + pkg/download/media_spooling_test.go | 150 +++++++++ pkg/download/model.go | 5 + pkg/download/temp.go | 26 ++ pkg/download/temp_test.go | 57 ++++ pkg/rest/model/server.go | 1 + pkg/rest/server.go | 1 + ui/flutter/lib/core/common/start_config.dart | 1 + .../lib/core/common/start_config.g.dart | 4 +- .../lib/core/entry/libgopeed_boot_native.dart | 5 + 36 files changed, 1747 insertions(+), 366 deletions(-) create mode 100644 internal/production/progress.go create mode 100644 internal/tempfiles/scope.go create mode 100644 internal/tempfiles/scope_test.go create mode 100644 pkg/download/engine/ffmpeg/http_spool.go create mode 100644 pkg/download/engine/ffmpeg/http_spool_test.go create mode 100644 pkg/download/engine/ffmpeg/spool.go create mode 100644 pkg/download/engine/ffmpeg/spool_test.go delete mode 100644 pkg/download/engine/ffmpeg/stream.go create mode 100644 pkg/download/media_spooling_test.go create mode 100644 pkg/download/temp.go create mode 100644 pkg/download/temp_test.go diff --git a/cmd/web/flags.go b/cmd/web/flags.go index 981860bb..98d4a8aa 100644 --- a/cmd/web/flags.go +++ b/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 } diff --git a/cmd/web/main.go b/cmd/web/main.go index 5a31909d..2b604d01 100644 --- a/cmd/web/main.go +++ b/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, diff --git a/internal/blob/registry.go b/internal/blob/registry.go index 0da060f9..4956767b 100644 --- a/internal/blob/registry.go +++ b/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() +} diff --git a/internal/controller/controller.go b/internal/controller/controller.go index dbaf2e6f..a352a876 100644 --- a/internal/controller/controller.go +++ b/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) } diff --git a/internal/production/progress.go b/internal/production/progress.go new file mode 100644 index 00000000..1c73c2f4 --- /dev/null +++ b/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) } diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index 6dfd9855..e6bd085e 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -513,11 +513,27 @@ func (f *Fetcher) asyncPrefetch() { } // Create temporary file for prefetch data - tmpFile, err := os.CreateTemp("", "gopeed-prefetch-*") + 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() @@ -532,7 +548,7 @@ func (f *Fetcher) asyncPrefetch() { }() buf := make([]byte, 32*1024) // 32KB buffer - reader := NewTimeoutReader(resp.Body, readTimeout) + reader := f.responseReader(resp) for { select { @@ -617,7 +633,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 +1349,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 +1731,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 +1848,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 +2471,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) +} diff --git a/internal/tempfiles/scope.go b/internal/tempfiles/scope.go new file mode 100644 index 00000000..b5d755da --- /dev/null +++ b/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 +} diff --git a/internal/tempfiles/scope_test.go b/internal/tempfiles/scope_test.go new file mode 100644 index 00000000..9d72c8e6 --- /dev/null +++ b/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) + } +} diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index 72953a8e..9b6313c8 100644 --- a/pkg/download/downloader.go +++ b/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" "github.com/GopeedLab/gopeed/pkg/base" "github.com/GopeedLab/gopeed/pkg/protocol/http" "github.com/GopeedLab/gopeed/pkg/util" @@ -97,6 +98,7 @@ type Progress struct { } type Downloader struct { + tempFiles *tempfiles.Scope Logger *logger.Logger ExtensionLogger *logger.Logger @@ -152,6 +154,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 @@ -263,24 +268,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 } @@ -299,6 +325,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) @@ -389,6 +416,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) } @@ -1124,7 +1154,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 { @@ -1399,17 +1429,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 } @@ -1808,7 +1827,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 } @@ -1962,11 +1983,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 diff --git a/pkg/download/engine/engine.go b/pkg/download/engine/engine.go index 147f1f98..6a8ea3f2 100644 --- a/pkg/download/engine/engine.go +++ b/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, diff --git a/pkg/download/engine/ffmpeg/README.md b/pkg/download/engine/ffmpeg/README.md index c7ef41bd..6e1ec99e 100644 --- a/pkg/download/engine/ffmpeg/README.md +++ b/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-`. 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 diff --git a/pkg/download/engine/ffmpeg/http_spool.go b/pkg/download/engine/ffmpeg/http_spool.go new file mode 100644 index 00000000..e5334de3 --- /dev/null +++ b/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() } diff --git a/pkg/download/engine/ffmpeg/http_spool_test.go b/pkg/download/engine/ffmpeg/http_spool_test.go new file mode 100644 index 00000000..f97b8a0d --- /dev/null +++ b/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) +} diff --git a/pkg/download/engine/ffmpeg/input.go b/pkg/download/engine/ffmpeg/input.go index 74e9c453..7182f6b9 100644 --- a/pkg/download/engine/ffmpeg/input.go +++ b/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() } diff --git a/pkg/download/engine/ffmpeg/input_test.go b/pkg/download/engine/ffmpeg/input_test.go index 63e86c21..9a12549e 100644 --- a/pkg/download/engine/ffmpeg/input_test.go +++ b/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") - } -} diff --git a/pkg/download/engine/ffmpeg/runtime.go b/pkg/download/engine/ffmpeg/runtime.go index 2d6b192f..5b60cc36 100644 --- a/pkg/download/engine/ffmpeg/runtime.go +++ b/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{}{}: diff --git a/pkg/download/engine/ffmpeg/runtime_test.go b/pkg/download/engine/ffmpeg/runtime_test.go index ee57c180..e5298dd3 100644 --- a/pkg/download/engine/ffmpeg/runtime_test.go +++ b/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) diff --git a/pkg/download/engine/ffmpeg/spool.go b/pkg/download/engine/ffmpeg/spool.go new file mode 100644 index 00000000..abf2965f --- /dev/null +++ b/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 } diff --git a/pkg/download/engine/ffmpeg/spool_test.go b/pkg/download/engine/ffmpeg/spool_test.go new file mode 100644 index 00000000..70fb2313 --- /dev/null +++ b/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") + } +} diff --git a/pkg/download/engine/ffmpeg/stream.go b/pkg/download/engine/ffmpeg/stream.go deleted file mode 100644 index 7db44f4f..00000000 --- a/pkg/download/engine/ffmpeg/stream.go +++ /dev/null @@ -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 } diff --git a/pkg/download/engine/ffmpeg_integration_test.go b/pkg/download/engine/ffmpeg_integration_test.go index 29c988ce..e505cde2 100644 --- a/pkg/download/engine/ffmpeg_integration_test.go +++ b/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) + } +} diff --git a/pkg/download/engine/inject/ffmpeg/ffmpeg.js b/pkg/download/engine/inject/ffmpeg/ffmpeg.js index 49b42f25..c54c5ff5 100644 --- a/pkg/download/engine/inject/ffmpeg/ffmpeg.js +++ b/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 }); })(); diff --git a/pkg/download/engine/inject/ffmpeg/module.go b/pkg/download/engine/inject/ffmpeg/module.go index cc73e990..f67dbf00 100644 --- a/pkg/download/engine/inject/ffmpeg/module.go +++ b/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 diff --git a/pkg/download/engine/inject/stream/module.go b/pkg/download/engine/inject/stream/module.go index 89db11fc..539d7cb9 100644 --- a/pkg/download/engine/inject/stream/module.go +++ b/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() + } +} diff --git a/pkg/download/engine/inject/stream/stream.js b/pkg/download/engine/inject/stream/stream.js index 829a9344..ca9d13af 100644 --- a/pkg/download/engine/inject/stream/stream.js +++ b/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; diff --git a/pkg/download/extension.go b/pkg/download/extension.go index c9bb7832..66b8b4c9 100644 --- a/pkg/download/extension.go +++ b/pkg/download/extension.go @@ -26,7 +26,6 @@ import ( var ( gitSuffix = ".git" - tempExtensionsDir = ".extensions" extensionsDir = "extensions" extensionIgnoreDirs = []string{gitSuffix, "node_modules"} @@ -204,11 +203,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() diff --git a/pkg/download/extension_engine_session.go b/pkg/download/extension_engine_session.go index 4f248c88..24e49c86 100644 --- a/pkg/download/extension_engine_session.go +++ b/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, diff --git a/pkg/download/media_spooling_test.go b/pkg/download/media_spooling_test.go new file mode 100644 index 00000000..c70f55a0 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/pkg/download/model.go b/pkg/download/model.go index 1a3b7ae8..c838cf31 100644 --- a/pkg/download/model.go +++ b/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 diff --git a/pkg/download/temp.go b/pkg/download/temp.go new file mode 100644 index 00000000..e3d63441 --- /dev/null +++ b/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 +} diff --git a/pkg/download/temp_test.go b/pkg/download/temp_test.go new file mode 100644 index 00000000..3f9328fb --- /dev/null +++ b/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) + } + }) + } +} diff --git a/pkg/rest/model/server.go b/pkg/rest/model/server.go index 441a0ffd..f54a2122 100644 --- a/pkg/rest/model/server.go +++ b/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"` diff --git a/pkg/rest/server.go b/pkg/rest/server.go index 69cbf2c8..c3cf82fc 100644 --- a/pkg/rest/server.go +++ b/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 { diff --git a/ui/flutter/lib/core/common/start_config.dart b/ui/flutter/lib/core/common/start_config.dart index ed442eab..6c362f2f 100644 --- a/ui/flutter/lib/core/common/start_config.dart +++ b/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; diff --git a/ui/flutter/lib/core/common/start_config.g.dart b/ui/flutter/lib/core/common/start_config.g.dart index 5d64420d..cb43efe2 100644 --- a/ui/flutter/lib/core/common/start_config.g.dart +++ b/ui/flutter/lib/core/common/start_config.g.dart @@ -23,9 +23,10 @@ StartConfig _$StartConfigFromJson(Map 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 _$StartConfigToJson(StartConfig instance) => 'mcpEnable': instance.mcpEnable, 'storage': instance.storage, 'storageDir': instance.storageDir, + 'tempDir': ?instance.tempDir, 'refreshInterval': instance.refreshInterval, 'apiToken': instance.apiToken, 'webViewRpcConfig': ?instance.webViewRpcConfig, diff --git a/ui/flutter/lib/core/entry/libgopeed_boot_native.dart b/ui/flutter/lib/core/entry/libgopeed_boot_native.dart index 6c3a9e4e..018d0a65 100644 --- a/ui/flutter/lib/core/entry/libgopeed_boot_native.dart +++ b/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 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); } From e27c7dddb48e613164a388eaf9c6a8180177766f Mon Sep 17 00:00:00 2001 From: Levi Date: Wed, 16 Sep 2026 23:43:43 +0800 Subject: [PATCH 2/2] fix(http): close prefetch response when temporary-file setup fails --- internal/protocol/http/fetcher.go | 21 +++--- internal/protocol/http/prefetch_setup_test.go | 74 +++++++++++++++++++ 2 files changed, 85 insertions(+), 10 deletions(-) create mode 100644 internal/protocol/http/prefetch_setup_test.go diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index e6bd085e..59aefc19 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -512,6 +512,17 @@ func (f *Fetcher) asyncPrefetch() { return } + defer func() { + // 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() + f.resolveResp = nil + } + f.resolveRespLock.Unlock() + }() + // Create temporary file for prefetch data tempDir := "" if f.ctl != nil && f.ctl.TempDir != "" { @@ -537,16 +548,6 @@ func (f *Fetcher) asyncPrefetch() { f.prefetchFile = tmpFile f.prefetchFilePath = tmpFile.Name() - defer func() { - // Close response body when prefetch stops - f.resolveRespLock.Lock() - if f.resolveResp != nil { - f.resolveResp.Body.Close() - f.resolveResp = nil - } - f.resolveRespLock.Unlock() - }() - buf := make([]byte, 32*1024) // 32KB buffer reader := f.responseReader(resp) diff --git a/internal/protocol/http/prefetch_setup_test.go b/internal/protocol/http/prefetch_setup_test.go new file mode 100644 index 00000000..b7d3187a --- /dev/null +++ b/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) + } + } + }) + } +}