feat(http): support advisory extension download sizes (#1599)

pkg/base/model.go

@@ -108,6 +108,9 @@ type FileInfo struct {
 
 // Options for download
 type Options struct {
+	// Size is an optional HTTP download size estimate. HTTP metadata and clean EOF take precedence.
+	// Other protocols ignore it.
+	Size int64 `json:"size,omitempty"`
 	// Download file name
 	Name string `json:"name"`
 	// Download file path

pkg/download/downloader.go

@@ -607,7 +607,7 @@ func (d *Downloader) createDirect(req *base.Request, opts *base.Options) (taskId
 		return
 	}
 	fetcher.Meta().Req = req
-	initOpt, err := d.initOptions(opts)
+	initOpt, err := d.initOptions(opts, req.URL)
 	if err != nil {
 		return
 	}
@@ -654,7 +654,7 @@ func (d *Downloader) CreateWithOptions(rrId string, opts *base.Options) (taskId
 	if opts == nil {
 		opts = &base.Options{}
 	}
-	opts, err = d.initOptions(opts.Clone())
+	opts, err = d.initOptions(opts.Clone(), fetcher.Meta().Req.URL)
 	if err != nil {
 		return "", err
 	}
@@ -670,6 +670,13 @@ func (d *Downloader) CreateWithOptions(rrId string, opts *base.Options) (taskId
 	if res := fetcher.Meta().Res; res != nil {
 		res.CalcSize(fetcher.Meta().Opts.SelectFiles)
 	}
+	// Resolve has no creation options. Publish a submitted HTTP estimate now
+	// while preserving any size already confirmed by the response.
+	if opts.Size > 0 {
+		if err := fetcher.Patch(nil, &base.Options{Size: opts.Size}); err != nil {
+			return "", err
+		}
+	}
 	defer func() {
 		d.fetcherMapLock.Lock()
 		delete(d.fetcherCache, rrId)
@@ -712,7 +719,7 @@ func (d *Downloader) Patch(id string, req *base.Request, opts *base.Options) err
 		}
 
 		// Call the fetcher's Patch method
-		if err := task.fetcher.Patch(req, opts); err != nil {
+		if err := task.fetcher.Patch(req, httpSizeOptions(opts, task.Protocol)); err != nil {
 			return err
 		}
 
@@ -1123,6 +1130,9 @@ func (d *Downloader) RuntimeStatus(id string) (*TaskRuntimeStatus, error) {
 		result.ExtractProgress = task.Progress.ExtractProgress
 	}
 	if task.Meta == nil || task.Meta.Res == nil {
+		if task.Meta != nil && task.Meta.Opts != nil && task.Protocol == "http" {
+			result.Total = max(0, task.Meta.Opts.Size)
+		}
 		return result, nil
 	}
 
@@ -1803,10 +1813,27 @@ func (d *Downloader) doCreate(f fetcher.Fetcher, opts *base.Options) (taskId str
 	return
 }
 
-func (d *Downloader) initOptions(opts *base.Options) (*base.Options, error) {
+// httpSizeOptions keeps estimates out of other protocols, including HLS URLs
+// served over HTTP. Clone before filtering because batch tasks may share options.
+func httpSizeOptions(opts *base.Options, protocol string) *base.Options {
+	if opts != nil && opts.Size != 0 && (protocol != "http" || opts.Size < 0) {
+		opts = opts.Clone()
+		opts.Size = 0
+	}
+	return opts
+}
+
+func (d *Downloader) initOptions(opts *base.Options, requestURL string) (*base.Options, error) {
 	if opts == nil {
 		opts = &base.Options{}
 	}
+	if opts.Size != 0 {
+		fm, err := d.parseFm(requestURL)
+		if err != nil {
+			return nil, err
+		}
+		opts = httpSizeOptions(opts, fm.Name())
+	}
 	if opts.SelectFiles == nil {
 		opts.SelectFiles = make([]int, 0)
 	}

pkg/download/engine/ffmpeg/README.md

@@ -30,6 +30,21 @@ support must remain disabled. Do not declare the sum of the input sizes as the
 output size. The downloader reads the merged output as one file; FFmpeg errors
 propagate through the stream instead of completing a partial download.
 
+Extension resolve results can set `files[].size` to an estimated output size,
+for example the sum of the selected video and audio input lengths. Clients forward
+this value as the optional task creation `opts.size` for HTTP downloads only
+(including Blob URLs served over HTTP). Other protocols, including HLS, ignore
+this option. The estimate is persisted
+and displayed while queued and downloading. A valid HTTP total length (including
+zero) always overrides it; otherwise a clean EOF records the actual output length.
+Estimates never determine byte ranges, file allocation, integrity checks, or task
+completion. Size authority is persisted alongside HTTP connection state for
+pause/retry and process restart.
+
+Blob options are unchanged: `createObjectURL`'s `size` is an exact source length,
+not an estimate. Leave it unset for generated merge output and pass any estimate
+through the extension resolve file instead.
+
 ## JS inputs, including SABR
 
 Stream inputs **must** be created by an `inputs({ signal })` callback. It returns
@@ -130,7 +145,8 @@ 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;
+The task stays downloading while queued or merging, displaying the optional
+`files[].size` estimate (or an unknown total when omitted);
 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

pkg/download/extension_resolve_size_test.go

@@ -0,0 +1,158 @@
+package download
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/GopeedLab/gopeed/pkg/base"
+)
+
+func TestExtensionResolveSizeThroughTaskCreation(t *testing.T) {
+	for _, mode := range []string{"http", "blob"} {
+		for _, known := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/known=%v", mode, known), func(t *testing.T) {
+				payload := []byte("0123456789")
+				server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					body := payload
+					if r.URL.Path == "/empty" {
+						body = nil
+					}
+					if known {
+						w.Header().Set("Content-Length", fmt.Sprint(len(body)))
+					}
+					w.(http.Flusher).Flush()
+					w.Write(body)
+				}))
+				defer server.Close()
+				d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir()})
+				if err := d.Setup(); err != nil {
+					t.Fatal(err)
+				}
+				defer d.Clear()
+				d.cfg.MaxRunning = 0 // Inspect persisted queued tasks before HTTP resolution.
+				extDir := t.TempDir()
+				manifest := `{"name":"size-hint","author":"test","title":"Size hint","version":"1.0.0","scripts":[{"event":"onResolve","match":{"urls":["https://example.com/size-hint"]},"entry":"index.js"}]}`
+				script := fmt.Sprintf(`gopeed.events.onResolve(async ctx=>{
+  const mode=%q, origin=%q;
+  const file=async(name,size,empty)=>({name,size,req:{url:mode==='http'?origin+(empty?'/empty':'/data'):
+    await gopeed.runtime.blob.createObjectURL(()=>new ReadableStream({start(c){if(!empty)c.enqueue(new Uint8Array([48,49,50,51,52,53,54,55,56,57]));c.close();}}))}});
+  ctx.res={files:[await file('under.bin',3,false),await file('over.bin',1048576,false),await file('empty.bin',99,true)]};
+});`, mode, server.URL)
+				for name, contents := range map[string]string{"manifest.json": manifest, "index.js": script} {
+					if err := os.WriteFile(filepath.Join(extDir, name), []byte(contents), 0600); err != nil {
+						t.Fatal(err)
+					}
+				}
+				if _, err := d.InstallExtensionByFolder(extDir, true); err != nil {
+					t.Fatal(err)
+				}
+				rr, err := d.Resolve(&base.Request{URL: "https://example.com/size-hint"})
+				if err != nil {
+					t.Fatal(err)
+				}
+				output := t.TempDir()
+				batch := &base.CreateTaskBatch{}
+				for _, file := range rr.Res.Files {
+					batch.Reqs = append(batch.Reqs, &base.CreateTaskBatchItem{Req: file.Req, Opts: &base.Options{Size: file.Size, Name: file.Name, Path: output}})
+				}
+				wire, err := json.Marshal(batch)
+				if err != nil {
+					t.Fatal(err)
+				}
+				var submitted base.CreateTaskBatch
+				if err := json.Unmarshal(wire, &submitted); err != nil {
+					t.Fatal(err)
+				}
+				ids, err := d.CreateDirectBatch(&submitted)
+				if err != nil {
+					t.Fatal(err)
+				}
+				for i, id := range ids {
+					task := d.GetTask(id)
+					if task.Meta.Res != nil {
+						t.Fatal("a size estimate must not mark the task resolved")
+					}
+					status, err := d.RuntimeStatus(id)
+					if err != nil || status.Total != rr.Res.Files[i].Size {
+						t.Fatalf("queued size lost: status=%+v err=%v", status, err)
+					}
+					var saved Task
+					if _, err := d.storage.Get(bucketTask, id, &saved); err != nil || saved.Meta.Opts.Size != status.Total {
+						t.Fatal("queued estimate was not persisted", err)
+					}
+					if err := d.doStart(task); err != nil {
+						t.Fatal(err)
+					}
+					waitForTaskTerminal(t, d, id, 5*time.Second)
+					body, err := os.ReadFile(filepath.Join(output, rr.Res.Files[i].Name))
+					if err != nil {
+						t.Fatal(err)
+					}
+					want := payload
+					if i == 2 {
+						want = nil
+					}
+					if !bytes.Equal(body, want) {
+						t.Fatalf("estimate changed output: %q", body)
+					}
+					status, err = d.RuntimeStatus(id)
+					if err != nil || status.Total != int64(len(want)) || status.Downloaded != int64(len(want)) || task.Meta.Res.Files[0].Size != int64(len(want)) {
+						t.Fatalf("final size not updated: %+v err=%v", status, err)
+					}
+				}
+			})
+		}
+	}
+}
+
+func TestResolvedHTTPSizeEstimateAtCreation(t *testing.T) {
+	for _, known := range []bool{false, true} {
+		t.Run(fmt.Sprintf("known=%v", known), func(t *testing.T) {
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+				if known {
+					w.Header().Set("Content-Length", "10")
+				}
+				w.(http.Flusher).Flush()
+				w.Write([]byte("0123456789"))
+			}))
+			defer server.Close()
+			d := NewDownloader(&DownloaderConfig{StorageDir: t.TempDir(), TempDir: t.TempDir()})
+			if err := d.Setup(); err != nil {
+				t.Fatal(err)
+			}
+			defer d.Clear()
+			d.cfg.MaxRunning = 0
+			rr, err := d.Resolve(&base.Request{URL: server.URL})
+			if err != nil {
+				t.Fatal(err)
+			}
+			id, err := d.CreateWithOptions(rr.ID, &base.Options{Size: 3, Path: t.TempDir(), Name: "output.bin"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			want := int64(3)
+			if known {
+				want = 10
+			}
+			status, err := d.RuntimeStatus(id)
+			if err != nil || status.Total != want || d.GetTask(id).Meta.Res.Files[0].Size != want {
+				t.Fatalf("resolve-before-create size=%+v want=%d err=%v", status, want, err)
+			}
+			if err := d.doStart(d.GetTask(id)); err != nil {
+				t.Fatal(err)
+			}
+			waitForTaskTerminal(t, d, id, 5*time.Second)
+			task := d.GetTask(id)
+			if task.Status != base.DownloadStatusDone || task.Meta.Res.Size != 10 {
+				t.Fatalf("estimate changed completion: status=%s size=%d", task.Status, task.Meta.Res.Size)
+			}
+		})
+	}
+}

pkg/download/extension_size_hint_test.go

@@ -0,0 +1,212 @@
+package download
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/GopeedLab/gopeed/pkg/base"
+)
+
+func TestExtensionFileSizeIsAdvisory(t *testing.T) {
+	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)
+	}
+	for _, mode := range []string{"plain", "merge", "wrapped-merge"} {
+		var expected []byte
+		for _, size := range []int64{0, 3, 1 << 20} {
+			t.Run(fmt.Sprintf("%s/size=%d", mode, size), func(t *testing.T) {
+				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)
+				result, err := e.RunString(`__gopeed_blob_create_object_url(()=>{
+  const source=b=>new ReadableStream({start(c){c.enqueue(new Uint8Array(b));c.close();}});
+  if (mode==='plain') return source([0,1,2,3,4,5,6,7,8,9]);
+  const output=__gopeed_ffmpeg.merge({inputs:()=>({video:source(videoBytes),audio:source(audioBytes)})});
+  if (mode==='merge') return output;
+  const reader=output.getReader();
+  return new ReadableStream({async pull(c){const {done,value}=await reader.read();if(done)c.close();else c.enqueue(value);},cancel(r){return reader.cancel(r);}});
+},{contentType:'video/mp4',range:false})`)
+				if err != nil {
+					t.Fatal(err)
+				}
+				output := t.TempDir()
+				id, err := d.CreateDirect(&base.Request{URL: fmt.Sprint(result)}, &base.Options{Size: size, Path: output, Name: "output.bin"})
+				if err != nil {
+					t.Fatal(err)
+				}
+				waitForTaskTerminal(t, d, id, 30*time.Second)
+				body, err := os.ReadFile(filepath.Join(output, "output.bin"))
+				if err != nil {
+					t.Fatal(err)
+				}
+				if size == 0 {
+					expected = body
+					if mode == "plain" && !bytes.Equal(body, []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}) {
+						t.Fatalf("unexpected stream output: %v", body)
+					}
+					if mode != "plain" && !bytes.Contains(body, []byte("moof")) {
+						t.Fatal("invalid merged output")
+					}
+				} else if !bytes.Equal(body, expected) {
+					t.Fatalf("size hint changed output: got %d bytes, want %d", len(body), len(expected))
+				}
+				task := d.GetTask(id)
+				task.lock.Lock()
+				defer task.lock.Unlock()
+				if task.Meta.Res.Size != int64(len(body)) || task.Meta.Res.Files[0].Size != int64(len(body)) {
+					t.Fatalf("final metadata must use actual size: %#v", task.Meta.Res)
+				}
+			})
+		}
+	}
+}
+
+func TestExtensionFileSizeYieldsToBlobContentLength(t *testing.T) {
+	for _, materialized := range []bool{false, true} {
+		for _, size := range []int64{3, 1 << 20} {
+			t.Run(fmt.Sprintf("materialized=%v/estimate=%d", materialized, size), func(t *testing.T) {
+				d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir()})
+				if err := d.Setup(); err != nil {
+					t.Fatal(err)
+				}
+				defer d.Clear()
+				e, _ := d.newExtensionEngine()
+				defer e.Close()
+				e.Runtime.Set("materialized", materialized)
+				result, err := e.RunString(`(()=>{
+  const payload=new Uint8Array([0,1,2,3,4,5,6,7,8,9]);
+  const source=materialized?new Blob([payload]):()=>new ReadableStream({start(c){c.enqueue(payload);c.close();}});
+  return __gopeed_blob_create_object_url(source,{size:10});
+})()`)
+				if err != nil {
+					t.Fatal(err)
+				}
+				output := t.TempDir()
+				id, err := d.CreateDirect(&base.Request{URL: fmt.Sprint(result)}, &base.Options{Size: size, Path: output, Name: "output.bin"})
+				if err != nil {
+					t.Fatal(err)
+				}
+				waitForTaskTerminal(t, d, id, 5*time.Second)
+				body, err := os.ReadFile(filepath.Join(output, "output.bin"))
+				if err != nil || !bytes.Equal(body, []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}) {
+					t.Fatalf("unexpected body=%v err=%v", body, err)
+				}
+				task := d.GetTask(id)
+				if task.Meta.Res.Size != 10 || task.Meta.Res.Files[0].Size != 10 {
+					t.Fatalf("Blob Content-Length did not override estimate: %+v", task.Meta.Res)
+				}
+			})
+		}
+	}
+}
+
+func TestExtensionFileSizeHintWaitsForEOF(t *testing.T) {
+	for _, fail := range []bool{false, true} {
+		for _, size := range []int64{3, 1 << 20} {
+			t.Run(fmt.Sprintf("fail=%v/size=%d", fail, size), func(t *testing.T) {
+				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("fail", fail)
+				result, err := e.RunString(`__gopeed_blob_create_object_url(()=>new ReadableStream({
+  start(c){c.enqueue(new Uint8Array([0,1,2,3,4,5,6,7,8,9]));globalThis.finish=()=>{if(fail)c.error(new Error('upstream failed'));else c.close();};}
+}),{})`)
+				if err != nil {
+					t.Fatal(err)
+				}
+				output := t.TempDir()
+				id, err := d.CreateDirect(&base.Request{URL: fmt.Sprint(result)}, &base.Options{Size: size, Path: output, Name: "output.bin"})
+				if err != nil {
+					t.Fatal(err)
+				}
+				waitForTaskDownloadedBetween(t, d, id, 10, 10, 5*time.Second)
+				task := d.GetTask(id)
+				task.lock.Lock()
+				if task.Meta.Res.Size != size || d.taskStatus(task) != base.DownloadStatusRunning {
+					t.Errorf("hint must remain visible until EOF: size=%d status=%s", task.Meta.Res.Size, d.taskStatus(task))
+				}
+				task.lock.Unlock()
+				stat, err := os.Stat(filepath.Join(output, "output.bin"))
+				if err != nil || stat.Size() != 10 {
+					t.Fatalf("hint must not preallocate or truncate output: stat=%v err=%v", stat, err)
+				}
+				if _, err := e.RunString(`finish()`); err != nil {
+					t.Fatal(err)
+				}
+				if fail {
+					waitForTaskStatus(t, d, id, base.DownloadStatusError, 5*time.Second)
+				} else {
+					waitForTaskTerminal(t, d, id, 5*time.Second)
+				}
+			})
+		}
+	}
+}
+
+func TestExtensionFileSizeHintPauseContinue(t *testing.T) {
+	for _, size := range []int64{3, 1 << 20} {
+		t.Run(fmt.Sprint(size), func(t *testing.T) {
+			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()
+			result, err := e.RunString(`(()=>{
+  let opens=0;
+  return __gopeed_blob_create_object_url(()=>new ReadableStream({start(c){
+    if (++opens===1) c.enqueue(new Uint8Array([0,1,2,3,4,5,6,7,8,9]));
+    else {c.enqueue(new Uint8Array([5,6]));c.close();}
+  }}),{});
+})()`)
+			if err != nil {
+				t.Fatal(err)
+			}
+			output := t.TempDir()
+			id, err := d.CreateDirect(&base.Request{URL: fmt.Sprint(result)}, &base.Options{Size: size, Path: output, Name: "output.bin"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			waitForTaskDownloadedBetween(t, d, id, 10, 10, 5*time.Second)
+			filter := &TaskFilter{IDs: []string{id}}
+			if err := d.Pause(filter); err != nil {
+				t.Fatal(err)
+			}
+			if err := d.Continue(filter); err != nil {
+				t.Fatal(err)
+			}
+			waitForTaskTerminal(t, d, id, 5*time.Second)
+			body, err := os.ReadFile(filepath.Join(output, "output.bin"))
+			if err != nil {
+				t.Fatal(err)
+			}
+			task := d.GetTask(id)
+			task.lock.Lock()
+			defer task.lock.Unlock()
+			if !bytes.Equal(body, []byte{5, 6}) || task.Meta.Res.Size != 2 {
+				t.Fatalf("resumed stream kept an estimate or stale bytes: body=%v size=%d", body, task.Meta.Res.Size)
+			}
+		})
+	}
+}