Merge latest PR #1512 into 2.0.0-beta.2

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,7 +7,6 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
-	"os"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -120,7 +119,8 @@ 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.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)
@@ -130,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")
 	}
 }
 
@@ -159,14 +158,7 @@ func TestStreamBackpressureAndCancel(t *testing.T) {
 	defer cancel()
 	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 }()
@@ -175,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 {
@@ -197,3 +189,34 @@ func TestStreamBackpressureAndCancel(t *testing.T) {
 		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")
+	}
+}

pkg/download/engine/ffmpeg/stream.go

@@ -4,21 +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
 	mu            sync.Mutex
-	file          *os.File
+	buffer        []byte
 	read, written int64
 	ended, closed bool
 	err           error
@@ -36,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 {
@@ -69,23 +56,25 @@ func (s *StreamInput) Write(p []byte) (int, error) {
 	if len(p) == 0 {
 		return 0, nil
 	}
-	if s.file == nil {
-		f, err := os.CreateTemp("", "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{}{}:
@@ -130,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()
@@ -166,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/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();
         }