diff --git a/pkg/download/engine/ffmpeg/README.md b/pkg/download/engine/ffmpeg/README.md index 7e97540c..c7ef41bd 100644 --- a/pkg/download/engine/ffmpeg/README.md +++ b/pkg/download/engine/ffmpeg/README.md @@ -32,29 +32,46 @@ propagate through the stream instead of completing a partial download. ## JS inputs, including SABR -Both `video` and `audio` accept either an HTTP descriptor or a -`ReadableStream`, including mixed inputs. SABR/UMP decoding belongs +Stream inputs **must** be created by an `inputs({ signal })` callback. It returns +`{ video, audio }`, each an HTTP descriptor or `ReadableStream`, +including mixed inputs. Do not start producers before this callback. SABR/UMP decoding belongs to the extension: pass the resulting media streams, with initialization data, not the SABR protocol bytes. Keep each selected track's codec/configuration stable for the duration of a merge. ```js const output = gopeed.runtime.ffmpeg.merge({ - video: videoStream, - audio: audioStream, + inputs: async ({ signal }) => { + const session = await prepareSession({ signal }); + const { videoStream, audioStream, abort } = await session.openStreams(); + if (signal.aborted) { abort(); throw new Error("Cancelled"); } + signal.addEventListener("abort", abort, { once: true }); + return { video: videoStream, audio: audioStream }; + }, signal: abortController.signal, args: ["-shortest", "-metadata", "title=Example"] }); ``` Creating the output does not start requests or acquire input readers until the -output is pulled. A stream input is single-use. To retry a Blob-backed task, +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. +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. +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. + ## Options -- `video`, `audio`: required inputs; HTTP descriptors have `url` and optional +- `inputs`: callback returning `{ video, audio }`, synchronously or asynchronously. + Required for stream inputs. Cannot be combined with top-level `video`/`audio`. +- `video`, `audio`: HTTP descriptors only when supplied at the top level; both + are required unless `inputs` is supplied. HTTP descriptors have `url` and optional `headers` (standard `HeadersInit`). HTTP requests use Gopeed's configured proxy. Supply required cookies/authorization explicitly; browser or `fetch` cookie jars are not implicitly shared. When `User-Agent` is absent, requests diff --git a/pkg/download/engine/ffmpeg/runtime.go b/pkg/download/engine/ffmpeg/runtime.go index cd6e49fc..2d6b192f 100644 --- a/pkg/download/engine/ffmpeg/runtime.go +++ b/pkg/download/engine/ffmpeg/runtime.go @@ -108,16 +108,35 @@ func Arguments(format string, extra []string) ([]string, error) { // Run shares compilation, but gives every merge its own memory and filesystem. // A clean stdout EOF is only delivered by the caller after this succeeds. func Run(ctx context.Context, video, audio Input, out io.Writer, format string, extra []string) error { + _, err := Arguments(format, extra) + if err != nil { + return err + } + release, err := Acquire(ctx) + if err != nil { + return err + } + defer release() + return RunAcquired(ctx, video, audio, out, format, extra) +} + +// Acquire reserves execution capacity before input factories or HTTP requests start. +func Acquire(ctx context.Context) (func(), error) { + select { + case slots <- struct{}{}: + case <-ctx.Done(): + return nil, ctx.Err() + } + var once sync.Once + return func() { once.Do(func() { <-slots }) }, nil +} + +// RunAcquired requires a reservation held until this call returns. +func RunAcquired(ctx context.Context, video, audio Input, out io.Writer, format string, extra []string) error { args, err := Arguments(format, extra) if err != nil { return err } - select { - case slots <- struct{}{}: - case <-ctx.Done(): - return ctx.Err() - } - defer func() { <-slots }() ready := make(chan struct{}) go func() { initialize(); close(ready) }() select { diff --git a/pkg/download/engine/ffmpeg_integration_test.go b/pkg/download/engine/ffmpeg_integration_test.go index 64e8f362..29c988ce 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" + 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" "strings" + "sync/atomic" "testing" "time" ) @@ -53,8 +55,8 @@ func TestFFmpegJSStreams(t *testing.T) { if(offset>=Math.max(v.length,a.length)){clearInterval(timer);vc.close();ac.close();} },1); const stream=__gopeed_ffmpeg.merge({ - video:inputMode==='http'?{url:sourceURL+'/video'}:video, - audio:inputMode==='streams'?audio:{url:sourceURL+'/audio'}, + inputs: () => ({ video:inputMode==='http'?{url:sourceURL+'/video'}:video, + audio:inputMode==='streams'?audio:{url:sourceURL+'/audio'} }), args:['-metadata','title=gopeed-test'] }); const bytes=await new Response(stream).arrayBuffer(); @@ -87,7 +89,7 @@ func TestFFmpegJSErrorAndCancel(t *testing.T) { cancel(){cancelled++;} }); const controller=new AbortController(); - const output=__gopeed_ffmpeg.merge({video:source(),audio:source(),signal:controller.signal, + const output=__gopeed_ffmpeg.merge({inputs:()=>({video:source(),audio:source()}),signal:controller.signal, args:scenario==='invalid-args'?['-i','/outside']:[]}); const reader=output.getReader(); const pending=reader.read().then(()=>'',e=>e.message); @@ -118,7 +120,7 @@ func TestFFmpegMalformedMediaFails(t *testing.T) { defer e.Close() _, err := e.RunString(`(async()=>{ const source=()=>new ReadableStream({start(c){c.enqueue(new Uint8Array([1,2,3]));c.close();}}); - return await new Response(__gopeed_ffmpeg.merge({video:source(),audio:source()})).arrayBuffer(); + return await new Response(__gopeed_ffmpeg.merge({inputs:()=>({video:source(),audio:source()})})).arrayBuffer(); })()`) if err == nil || !strings.Contains(err.Error(), "ffmpeg") { t.Fatalf("expected FFmpeg error: %v", err) @@ -141,7 +143,7 @@ func TestFFmpegBlobOpener(t *testing.T) { e.Runtime.Set("audioBytes", e.Runtime.NewArrayBuffer(audio)) _, 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({video:source(videoBytes),audio:source(audioBytes)}); + return __gopeed_ffmpeg.merge({inputs:()=>({video:source(videoBytes),audio:source(audioBytes)})}); },{contentType:'video/mp4'})`) if err != nil { t.Fatal(err) @@ -175,7 +177,7 @@ func TestFFmpegEngineCloseCancelsRequest(t *testing.T) { _, err := e.RunString(fmt.Sprintf(` globalThis.cancelCount=0; const audio=new ReadableStream({cancel(){cancelCount++;}}); - __gopeed_ffmpeg.merge({video:{url:%q},audio}).getReader().read().catch(()=>{}); + __gopeed_ffmpeg.merge({inputs:()=>({video:{url:%q},audio})}).getReader().read().catch(()=>{}); "started";`, srv.URL)) if err != nil { t.Fatal(err) @@ -195,3 +197,125 @@ func TestFFmpegEngineCloseCancelsRequest(t *testing.T) { t.Fatal("JS input not cancelled on shutdown") } } + +func TestFFmpegInputFactoriesWaitForCapacity(t *testing.T) { + for _, mode := range []string{"factory", "http", "cancel"} { + t.Run(mode, func(t *testing.T) { + release1, _ := media.Acquire(context.Background()) + defer release1() + release2, _ := media.Acquire(context.Background()) + defer release2() + var requests atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { requests.Add(1); http.Error(w, "test", 500) })) + defer srv.Close() + e := NewEngine(nil) + defer e.Close() + e.Runtime.Set("mode", mode) + e.Runtime.Set("sourceURL", srv.URL) + result, err := e.RunString(`(async()=>{ + globalThis.factoryCalls=0; + globalThis.abortQueue=new AbortController(); + const options=mode==='http'?{video:{url:sourceURL},audio:{url:sourceURL}}: + {inputs:()=>{factoryCalls++;throw new Error('factory failed');}}; + options.signal=abortQueue.signal; + globalThis.work=__gopeed_ffmpeg.merge(options).getReader().read().then(()=>'',e=>e.message); + await new Promise(r=>setTimeout(r,50)); + if(mode==='cancel')abortQueue.abort(); + return factoryCalls; + })()`) + 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 == "cancel" { + if _, err = e.RunString(`work`); err != nil { + t.Fatal(err) + } + } + release1() + if _, err = e.RunString(`work`); err != nil { + t.Fatal(err) + } + calls, err := e.RunString(`factoryCalls`) + if err != nil { + t.Fatal(err) + } + if mode == "factory" && fmt.Sprint(calls) != "1" { + t.Fatal(calls) + } + if mode == "cancel" && fmt.Sprint(calls) != "0" { + t.Fatal("cancelled factory ran") + } + if mode == "http" && requests.Load() != 1 { + t.Fatal("HTTP did not start after admission") + } + // Failure/cancellation must return the acquired slot even with the other held. + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + release, err := media.Acquire(ctx) + if err != nil { + t.Fatalf("slot leaked: %v", err) + } + release() + }) + } +} + +func TestFFmpegRequiresFactoryForStreams(t *testing.T) { + e := NewEngine(nil) + defer e.Close() + result, err := e.RunString(`(()=>{ + const source=()=>new ReadableStream({}); + let rejected=0; + for(const opts of [ + {video:source(),audio:source()}, + {video:{url:'https://example.test'},audio:source()}, + {inputs:()=>({}),video:{url:'https://example.test'}}, + {inputs:{video:source(),audio:source()}} + ]) {try{__gopeed_ffmpeg.merge(opts);}catch(e){if(e instanceof TypeError)rejected++;}} + return rejected; + })()`) + if err != nil || fmt.Sprint(result) != "4" { + t.Fatalf("validation: %v %v", result, err) + } +} + +func TestFFmpegCancelPendingFactoryCleansLateInputs(t *testing.T) { + e := NewEngine(nil) + defer e.Close() + result, err := e.RunString(`(async()=>{ + let ready,finish,cancelled=0,signal; + const entered=new Promise(r=>ready=r); + const abort=new AbortController(); + const output=__gopeed_ffmpeg.merge({signal:abort.signal,inputs:async opts=>{ + signal=opts.signal;ready();await new Promise(r=>finish=r); + const source=()=>new ReadableStream({cancel(){cancelled++;}}); + return {video:source(),audio:source()}; + }}); + const work=output.getReader().read().catch(e=>e.name); + await entered;abort.abort();await work;finish(); + await new Promise(r=>setTimeout(r,30)); + return {cancelled,aborted:signal.aborted}; + })()`) + if err != nil { + t.Fatal(err) + } + state := result.(map[string]any) + if fmt.Sprint(state["cancelled"]) != "2" || state["aborted"] != true { + t.Fatal(state) + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + first, err := media.Acquire(ctx) + if err != nil { + t.Fatal(err) + } + defer first() + second, err := media.Acquire(ctx) + if err != nil { + t.Fatal("factory cancellation leaked capacity") + } + second() +} diff --git a/pkg/download/engine/inject/ffmpeg/ffmpeg.js b/pkg/download/engine/inject/ffmpeg/ffmpeg.js index 1441813f..49b42f25 100644 --- a/pkg/download/engine/inject/ffmpeg/ffmpeg.js +++ b/pkg/download/engine/inject/ffmpeg/ffmpeg.js @@ -1,6 +1,8 @@ (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; const cancel = __gopeed_ffmpeg_cancel; @@ -9,8 +11,9 @@ for (const stop of Array.from(active)) stop(new Error("extension engine closed")); }; - function input(value, name) { + function input(value, name, allowStream) { if (value && typeof value.getReader === "function") { + if (!allowStream) throw new TypeError("stream inputs must be created by the inputs callback"); if (value.locked) throw new TypeError(name + " stream is locked"); return { stream: value, request: null }; } @@ -24,9 +27,11 @@ function merge(options) { if (!options || typeof options !== "object") throw new TypeError("merge options are required"); - const video = input(options.video, "video"); - const audio = input(options.audio, "audio"); - if (video.stream && video.stream === audio.stream) throw new TypeError("video and audio must be distinct streams"); + const factory = options.inputs; + if (factory !== undefined && typeof factory !== "function") throw new TypeError("inputs must be a callback"); + if (factory && (options.video !== undefined || options.audio !== undefined)) throw new TypeError("inputs cannot be combined with video/audio"); + let video = factory ? null : input(options.video, "video", false); + let audio = factory ? null : input(options.audio, "audio", false); const args = options.args === undefined ? [] : options.args; if (!Array.isArray(args) || args.some(v => typeof v !== "string")) throw new TypeError("args must be a string array"); const format = options.format === undefined ? "mp4" : options.format; @@ -34,11 +39,20 @@ const signal = options.signal; let id, controller, stopped = false; const readers = []; + const lifetime = new AbortController(); + let ownedInputs; + function discardInputs(inputs, reason) { + for (const source of new Set([inputs && inputs.video, inputs && inputs.audio])) { + if (source && typeof source.cancel === "function" && !source.locked) Promise.resolve(source.cancel(reason)).catch(() => {}); + } + } function stop(reason) { if (stopped) return; stopped = true; active.delete(shutdown); + lifetime.abort(); + discardInputs(ownedInputs, reason); if (id) cancel(id); if (signal) signal.removeEventListener("abort", abort); for (const reader of readers) { @@ -90,11 +104,21 @@ if (stopped) return; try { if (!id) { - // Acquire both readers before starting native work. + id = create({ format, args }); + await acquire(id); + if (stopped) return; + if (factory) { + const inputs = await factory({ signal: lifetime.signal }); + if (stopped) { discardInputs(inputs, new Error("FFmpeg operation cancelled")); return; } + ownedInputs = inputs; + video = input(inputs && inputs.video, "video", true); + audio = input(inputs && inputs.audio, "audio", true); + } + if (video.stream && video.stream === audio.stream) throw new TypeError("video and audio must be distinct streams"); for (const source of [video, audio]) { if (source.stream) readers.push(source.stream.getReader()); } - id = create({ video: video.request, audio: audio.request, format, args }); + start(id, { video: video.request, audio: audio.request, format, args }); let readerIndex = 0; for (const [index, source] of [video, audio].entries()) { if (source.stream) void pump(readers[readerIndex++], index); @@ -111,5 +135,5 @@ cancel(reason) { stop(reason); } }); } - globalThis.__gopeed_ffmpeg = Object.freeze({ merge }); + globalThis.__gopeed_ffmpeg = Object.freeze({ merge, supportsInputFactory: true }); })(); diff --git a/pkg/download/engine/inject/ffmpeg/module.go b/pkg/download/engine/inject/ffmpeg/module.go index 42dec9cb..cc73e990 100644 --- a/pkg/download/engine/inject/ffmpeg/module.go +++ b/pkg/download/engine/inject/ffmpeg/module.go @@ -37,11 +37,17 @@ type job struct { cancel context.CancelFunc output *io.PipeReader streams [2]*media.StreamInput + mu sync.Mutex + ready chan struct{} + start chan options + started bool } func (j *job) stop() { j.cancel() j.output.CloseWithError(context.Canceled) + j.mu.Lock() + defer j.mu.Unlock() for _, s := range j.streams { if s != nil { s.Close() @@ -99,6 +105,100 @@ func Enable(vm *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error { if _, err := media.Arguments(opts.Format, opts.Args); err != nil { panic(vm.NewGoError(err)) } + ctx, cancel := context.WithCancel(context.Background()) + r, w := io.Pipe() + j := &job{ctx: ctx, cancel: cancel, output: r, ready: make(chan struct{}), start: make(chan options, 1)} + id := fmt.Sprintf("ffmpeg-%d", next.Add(1)) + mu.Lock() + jobs[id] = j + mu.Unlock() + go func() { + var inputs [2]media.Input + var runErr error + defer func() { + if v := recover(); v != nil { + runErr = fmt.Errorf("ffmpeg panic: %v", v) + } + cancel() + for _, in := range inputs { + if in != nil { + in.Close() + } + } + 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: + case <-ctx.Done(): + runErr = ctx.Err() + return + } + for i, source := range []*media.HTTPSource{opts.Video, opts.Audio} { + if source == nil { + inputs[i] = j.streams[i] + } else { + inputs[i], runErr = media.OpenHTTP(ctx, client, *source) + if runErr != nil { + return + } + } + } + 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) + if j == nil { + panic(vm.NewGoError(io.ErrClosedPipe)) + } + var opts options + if err := vm.ExportTo(call.Argument(1), &opts); err != nil { + panic(vm.NewGoError(err)) + } + if _, err := media.Arguments(opts.Format, opts.Args); err != nil { + panic(vm.NewGoError(err)) + } for _, source := range []*media.HTTPSource{opts.Video, opts.Audio} { if source == nil || cfg.DefaultUserAgent == nil { continue @@ -117,48 +217,20 @@ func Enable(vm *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error { source.Headers["User-Agent"] = *cfg.DefaultUserAgent } } - ctx, cancel := context.WithCancel(context.Background()) - r, w := io.Pipe() - j := &job{ctx: ctx, cancel: cancel, output: r} + + j.mu.Lock() + defer j.mu.Unlock() + if j.started || j.ctx.Err() != nil { + panic(vm.NewGoError(io.ErrClosedPipe)) + } + j.started = true for i, source := range []*media.HTTPSource{opts.Video, opts.Audio} { if source == nil { - j.streams[i] = media.NewStreamInput(ctx) + j.streams[i] = media.NewStreamInput(j.ctx) } } - id := fmt.Sprintf("ffmpeg-%d", next.Add(1)) - mu.Lock() - jobs[id] = j - mu.Unlock() - go func() { - var inputs [2]media.Input - var runErr error - defer func() { - if v := recover(); v != nil { - runErr = fmt.Errorf("ffmpeg panic: %v", v) - } - cancel() - for _, in := range inputs { - if in != nil { - in.Close() - } - } - // Closing only after Run returns prevents partial output from - // being mistaken for a successful download on FFmpeg failure. - w.CloseWithError(runErr) - }() - for i, source := range []*media.HTTPSource{opts.Video, opts.Audio} { - if source == nil { - inputs[i] = j.streams[i] - } else { - inputs[i], runErr = media.OpenHTTP(ctx, client, *source) - if runErr != nil { - return - } - } - } - runErr = media.Run(ctx, inputs[0], inputs[1], w, opts.Format, opts.Args) - }() - return vm.ToValue(id) + j.start <- opts + return goja.Undefined() }); err != nil { return err }