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 2f3e1ad6..63e86c21 100644 --- a/pkg/download/engine/ffmpeg/input_test.go +++ b/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") + } +} diff --git a/pkg/download/engine/ffmpeg/stream.go b/pkg/download/engine/ffmpeg/stream.go index 14aa656d..7db44f4f 100644 --- a/pkg/download/engine/ffmpeg/stream.go +++ b/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 } 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(); }