diff --git a/pkg/download/engine/engine.go b/pkg/download/engine/engine.go index 96420225..147f1f98 100644 --- a/pkg/download/engine/engine.go +++ b/pkg/download/engine/engine.go @@ -160,8 +160,6 @@ func (e *Engine) addCleanup(cleanup func()) { } type Config struct { - // TempDir supplies an app-writable directory for FFmpeg stream buffers. - TempDir string // HTTPUserAgent is a snapshot of the downloader HTTP default. Nil leaves standalone engines unchanged. HTTPUserAgent *string ProxyConfig *base.DownloaderProxyConfig @@ -209,7 +207,6 @@ func NewEngine(cfg *Config) *Engine { return } if err := ffmpegapi.Enable(runtime, loop, &ffmpegapi.Config{ - TempDir: cfg.TempDir, 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 001ca7dc..7e97540c 100644 --- a/pkg/download/engine/ffmpeg/README.md +++ b/pkg/download/engine/ffmpeg/README.md @@ -92,14 +92,15 @@ 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 temporary disk ring -buffers (up to **64 MiB per input**, independent of total media size). This -allows shared upstream demultiplexers to make progress when FFmpeg reads one -track ahead of another. It does not enable seeking. Full buffers apply async -backpressure; a full buffer stalled for 30 seconds fails explicitly. Extensions -must also bound their own upstream queues. Disk buffers are removed on normal -completion, failure, cancellation and engine shutdown; abrupt process death -can leave temporary files in the OS temporary directory. +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. 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/input_test.go b/pkg/download/engine/ffmpeg/input_test.go index dc0aa8f8..63e86c21 100644 --- a/pkg/download/engine/ffmpeg/input_test.go +++ b/pkg/download/engine/ffmpeg/input_test.go @@ -7,8 +7,6 @@ import ( "io" "net/http" "net/http/httptest" - "os" - "path/filepath" "strings" "sync/atomic" "testing" @@ -118,10 +116,11 @@ func TestHTTPSequentialAndValidation(t *testing.T) { } func TestStreamRing(t *testing.T) { - s := NewStreamInput(context.Background(), "") + s := NewStreamInput(context.Background()) defer s.Close() // Start near the boundary to exercise wrap-around without a huge fixture. - s.read = StreamBufferLimit - 3 + 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) @@ -131,12 +130,11 @@ func TestStreamRing(t *testing.T) { if err != nil || string(b) != "abcdefgh" { t.Fatalf("%q %v", b, err) } - name := s.file.Name() if err = s.Close(); err != nil { t.Fatal(err) } - if _, err = os.Stat(name); err == nil { - t.Fatal("temporary file leaked") + if s.buffer != nil { + t.Fatal("buffer retained after close") } } @@ -158,16 +156,9 @@ func TestOutputArguments(t *testing.T) { func TestStreamBackpressureAndCancel(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - s := NewStreamInput(ctx, "") + s := NewStreamInput(ctx) defer s.Close() - f, err := os.CreateTemp(t.TempDir(), "ring") - if err != nil { - t.Fatal(err) - } - s.file = f - if err = f.Truncate(StreamBufferLimit); err != nil { - t.Fatal(err) - } + s.buffer = make([]byte, StreamBufferLimit) s.written = StreamBufferLimit finished := make(chan error, 1) go func() { _, err := s.Write([]byte{42}); finished <- err }() @@ -176,7 +167,7 @@ func TestStreamBackpressureAndCancel(t *testing.T) { t.Fatalf("did not apply backpressure: %v", err) case <-time.After(10 * time.Millisecond): } - if _, err = s.Read(make([]byte, 1)); err != nil { + if _, err := s.Read(make([]byte, 1)); err != nil { t.Fatal(err) } select { @@ -199,37 +190,33 @@ func TestStreamBackpressureAndCancel(t *testing.T) { } } -// A mobile app can write to its storage directory even when the OS default -// temporary directory is unavailable. The buffer must also be removed on close. -func TestStreamInputAppTempDir(t *testing.T) { - root := t.TempDir() - unavailable := filepath.Join(root, "not-a-directory") - if err := os.WriteFile(unavailable, []byte("blocked"), 0600); err != nil { +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) } - t.Setenv("TMPDIR", unavailable) - t.Setenv("TMP", unavailable) - t.Setenv("TEMP", unavailable) - dir := filepath.Join(root, "app", "temp", "ffmpeg") - s := NewStreamInput(context.Background(), dir) - t.Cleanup(func() { s.Close() }) - data := []byte("media stream") - if _, err := s.Write(data); err != nil { - t.Fatal(err) - } - if filepath.Dir(s.file.Name()) != dir { - t.Fatalf("unexpected buffer: %s", s.file.Name()) - } + 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) - if err != nil || !bytes.Equal(got, data) { - t.Fatalf("read %q, error %v", got, err) + 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 err := s.Close(); err != nil { - t.Fatal(err) + if len(s.buffer) > StreamBufferLimit { + t.Fatal("buffer exceeded limit") } - entries, err := os.ReadDir(dir) - if err != nil || len(entries) != 0 { - t.Fatalf("buffer cleanup: %v, %v", entries, err) + s.Close() + if s.buffer != nil { + t.Fatal("buffer retained on close") } } diff --git a/pkg/download/engine/ffmpeg/runtime_test.go b/pkg/download/engine/ffmpeg/runtime_test.go index d7ea11fe..ee57c180 100644 --- a/pkg/download/engine/ffmpeg/runtime_test.go +++ b/pkg/download/engine/ffmpeg/runtime_test.go @@ -128,7 +128,7 @@ func TestWebMToMP4(t *testing.T) { if err != nil { t.Fatal(err) } - inputs[i] = NewStreamInput(ctx, "") + inputs[i] = NewStreamInput(ctx) defer inputs[i].Close() if _, err = inputs[i].Write(data); err != nil { t.Fatal(err) @@ -205,7 +205,7 @@ func TestWASMMerge(t *testing.T) { }) } // A malformed input must produce an error, not successful empty output. - s := NewStreamInput(context.Background(), "") + s := NewStreamInput(context.Background()) defer s.Close() s.Write([]byte("invalid")) s.End(nil) diff --git a/pkg/download/engine/ffmpeg/stream.go b/pkg/download/engine/ffmpeg/stream.go index a2c1208c..7db44f4f 100644 --- a/pkg/download/engine/ffmpeg/stream.go +++ b/pkg/download/engine/ffmpeg/stream.go @@ -4,22 +4,19 @@ import ( "context" "errors" "io" - "os" "sync" - "time" ) -// A bounded disk ring lets both JS inputs drain independently, including when +// 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 applies backpressure. A prolonged stall fails instead of -// deadlocking a shared upstream or growing forever. -const StreamBufferLimit = 64 * 1024 * 1024 +// 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 - tempDir string mu sync.Mutex - file *os.File + buffer []byte read, written int64 ended, closed bool err error @@ -27,8 +24,8 @@ type StreamInput struct { space chan struct{} } -func NewStreamInput(ctx context.Context, tempDir string) *StreamInput { - return &StreamInput{ctx: ctx, tempDir: tempDir, wake: make(chan struct{}, 1), space: make(chan struct{}, 1)} +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) { @@ -37,24 +34,13 @@ func (s *StreamInput) Write(p []byte) (int, error) { if len(p) > StreamBufferLimit { return 0, errors.New("ffmpeg: input chunk exceeds buffer capacity") } - var timer *time.Timer - defer func() { - if timer != nil { - timer.Stop() - } - }() for int64(len(p)) > StreamBufferLimit-(s.written-s.read) && !s.closed && !s.ended { - if timer == nil { - timer = time.NewTimer(30 * time.Second) - } s.mu.Unlock() var err error select { case <-s.space: case <-s.ctx.Done(): err = s.ctx.Err() - case <-timer.C: - err = errors.New("ffmpeg: input buffer stalled for 30 seconds; upstream tracks may be too far apart") } s.mu.Lock() if err != nil { @@ -70,29 +56,25 @@ func (s *StreamInput) Write(p []byte) (int, error) { if len(p) == 0 { return 0, nil } - if s.file == nil { - // Android apps cannot write to Go's default /data/local/tmp. - if s.tempDir != "" { - if err := os.MkdirAll(s.tempDir, 0700); err != nil { - return 0, err - } - } - f, err := os.CreateTemp(s.tempDir, "gopeed-ffmpeg-*") - if err != nil { - return 0, err - } - s.file = f - } - off := s.written % StreamBufferLimit - first := min(len(p), StreamBufferLimit-int(off)) - if _, err := s.file.WriteAt(p[:first], off); err != nil { - return 0, err - } - if first < len(p) { - if _, err := s.file.WriteAt(p[first:], 0); err != nil { - return 0, err + // 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{}{}: @@ -137,16 +119,16 @@ func (s *StreamInput) Read(p []byte) (int, error) { return 0, err } if s.read < s.written { - off := s.read % StreamBufferLimit - n := min(int64(len(p)), s.written-s.read, StreamBufferLimit-off) - count, err := s.file.ReadAt(p[:n], off) + 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, err + return count, nil } ended := s.ended s.mu.Unlock() @@ -173,12 +155,7 @@ func (s *StreamInput) Close() error { case s.wake <- struct{}{}: default: } - if s.file != nil { - name := s.file.Name() - s.file.Close() - s.file = nil - return os.Remove(name) - } + s.buffer = nil return nil } func (*StreamInput) Seek(int64, int) (int64, error) { return 0, ErrSeek } diff --git a/pkg/download/engine/inject/ffmpeg/module.go b/pkg/download/engine/inject/ffmpeg/module.go index 2488d2d1..42dec9cb 100644 --- a/pkg/download/engine/inject/ffmpeg/module.go +++ b/pkg/download/engine/inject/ffmpeg/module.go @@ -22,7 +22,6 @@ import ( var script string type Config struct { - TempDir string DefaultUserAgent *string ProxyHandler func(*http.Request) (*url.URL, error) RegisterCleanup func(func()) @@ -123,7 +122,7 @@ func Enable(vm *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error { j := &job{ctx: ctx, cancel: cancel, output: r} for i, source := range []*media.HTTPSource{opts.Video, opts.Audio} { if source == nil { - j.streams[i] = media.NewStreamInput(ctx, cfg.TempDir) + j.streams[i] = media.NewStreamInput(ctx) } } id := fmt.Sprintf("ffmpeg-%d", next.Add(1)) diff --git a/pkg/download/engine/inject/stream/module.go b/pkg/download/engine/inject/stream/module.go index 5aaa214f..89db11fc 100644 --- a/pkg/download/engine/inject/stream/module.go +++ b/pkg/download/engine/inject/stream/module.go @@ -86,18 +86,23 @@ func Enable(runtime *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error } if err := runtime.Set("__gopeed_blob_pipe_chunk", func(call goja.FunctionCall) goja.Value { id := call.Argument(0).String() - ok := blobPipes.Push(id, exportBytes(call.Argument(1))) - return runtime.ToValue(ok) + chunk := append([]byte(nil), exportBytes(call.Argument(1))...) + promise, resolve, _ := runtime.NewPromise() + go func() { + ok := blobPipes.Push(id, chunk) + loop.RunOnLoop(func(vm *goja.Runtime) { resolve(ok) }) + }() + return runtime.ToValue(promise) }); err != nil { return err } if err := runtime.Set("__gopeed_blob_pipe_close", func(id string) { - blobPipes.Close(id, nil) + go blobPipes.Close(id, nil) }); err != nil { return err } if err := runtime.Set("__gopeed_blob_pipe_error", func(id string, message string) { - blobPipes.Close(id, errors.New(message)) + go blobPipes.Close(id, errors.New(message)) }); err != nil { return err } diff --git a/pkg/download/engine/inject/stream/module_integration_test.go b/pkg/download/engine/inject/stream/module_integration_test.go index b2c11cec..e9e07411 100644 --- a/pkg/download/engine/inject/stream/module_integration_test.go +++ b/pkg/download/engine/inject/stream/module_integration_test.go @@ -269,3 +269,57 @@ func newStreamTestEngine(t *testing.T, created chan<- createdObjectURL) (*engine }) return e, e.Close } + +// A stalled disk/network consumer must pause the producer without blocking +// the JS event loop, so cancellation and other tracks can still run. +func TestBlobBackpressureLeavesEventLoopResponsive(t *testing.T) { + created := make(chan createdObjectURL, 1) + e, cleanup := newStreamTestEngine(t, created) + defer cleanup() + _, err := e.RunString(` + globalThis.produced = 0; + globalThis.sourceCancelled = false; + __gopeed_blob_create_object_url(() => new ReadableStream({ + pull(c) { produced++; c.enqueue(new Uint8Array(64*1024)); }, + cancel() { sourceCancelled = true; } + })); + `) + if err != nil { + t.Fatal(err) + } + object := <-created + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + reader, err := object.open(ctx, stream.ObjectURLOpenRequest{End: -1}) + if err != nil { + t.Fatal(err) + } + defer reader.Close() + time.Sleep(100 * time.Millisecond) + value, err := e.RunString("produced") + if err != nil { + t.Fatal(err) + } + before := value.(int64) + if before > 10 { + t.Fatalf("unbounded production: %d", before) + } + time.Sleep(50 * time.Millisecond) + after, err := e.RunString("produced") + if err != nil || after != value { + t.Fatalf("producer did not pause: %v -> %v (%v)", value, after, err) + } + if ctx.Err() != nil { + t.Fatal("event loop only resumed after cancellation") + } + if _, err = reader.Read(make([]byte, 64*1024)); err != nil { + t.Fatal(err) + } + time.Sleep(30 * time.Millisecond) + after, err = e.RunString("produced") + if err != nil || after.(int64) <= before { + t.Fatalf("producer did not resume: %v (%v)", after, err) + } + reader.Close() + waitForJSValue(t, e, "String(sourceCancelled)", "true") +} diff --git a/pkg/download/engine/inject/stream/stream.js b/pkg/download/engine/inject/stream/stream.js index c6203559..829a9344 100644 --- a/pkg/download/engine/inject/stream/stream.js +++ b/pkg/download/engine/inject/stream/stream.js @@ -764,11 +764,14 @@ return; } const chunk = value instanceof Uint8Array ? value : new Uint8Array(value); - const buffer = chunk.buffer.slice(chunk.byteOffset, chunk.byteOffset + chunk.byteLength); - if (!globalThis.__gopeed_blob_pipe_chunk(pipeId, buffer)) { - state.cancelRequested = true; - state.cancelReason = "blob pipe closed"; - return; + for (let offset = 0; offset < chunk.byteLength; offset += 256 * 1024) { + const part = chunk.subarray(offset, offset + 256 * 1024); + const buffer = part.buffer.slice(part.byteOffset, part.byteOffset + part.byteLength); + if (!await globalThis.__gopeed_blob_pipe_chunk(pipeId, buffer)) { + state.cancelRequested = true; + state.cancelReason = "blob pipe closed"; + return; + } } await yieldBlobPipeTask(); } diff --git a/pkg/download/extension_engine_session.go b/pkg/download/extension_engine_session.go index 4fa6819b..4f248c88 100644 --- a/pkg/download/extension_engine_session.go +++ b/pkg/download/extension_engine_session.go @@ -3,7 +3,6 @@ package download import ( "context" "io" - "path/filepath" "strings" "sync" @@ -127,12 +126,7 @@ func (d *Downloader) newExtensionEngine() (*engine.Engine, *engineSession) { }{UserAgent: httpProtocol.DefaultUserAgent} d.getProtocolConfig("http", &httpConfig) userAgent := strings.TrimSpace(httpConfig.UserAgent) - tempDir := "" - if d.cfg.StorageDir != "" { - tempDir = filepath.Join(d.cfg.StorageDir, "temp", "ffmpeg") - } e := engine.NewEngine(&engine.Config{ - TempDir: tempDir, HTTPUserAgent: &userAgent, ProxyConfig: d.cfg.Proxy, StreamConfig: engineCfg,