fix(stream): bound FFmpeg memory buffers and propagate Blob backpressure

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,

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

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")
 	}
 }

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)

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 }

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))

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
 	}

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")
+}

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();
         }

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,