From fbf8b106b51f368d8fec39053ed7d356c550f715 Mon Sep 17 00:00:00 2001 From: Jeff Pai Date: Wed, 2 Sep 2026 14:56:06 -0700 Subject: [PATCH 1/7] feat(pipeline): add dispose() for explicit native release Release the native GstPipeline eagerly instead of waiting for GC to finalize the wrapper, avoiding unbounded native memory growth when a pipeline is built per unit of work. dispose() drives the pipeline to NULL and drops the owning reference; a require_pipeline guard makes any use-after-dispose throw. Terminal and idempotent. Adds unit tests, README docs, and an example. --- README.md | 47 ++++++++++++++++ examples/dispose.mjs | 56 +++++++++++++++++++ src/cpp/pipeline.cpp | 99 +++++++++++++++++++++++++++------ src/cpp/pipeline.hpp | 2 + src/ts/index.ts | 1 + src/ts/pipeline-dispose.test.ts | 76 +++++++++++++++++++++++++ 6 files changed, 263 insertions(+), 18 deletions(-) create mode 100644 examples/dispose.mjs create mode 100644 src/ts/pipeline-dispose.test.ts diff --git a/README.md b/README.md index 508df7b..30e75bb 100644 --- a/README.md +++ b/README.md @@ -810,6 +810,49 @@ setTimeout(async () => { }, 5000); ``` +### Releasing a Pipeline (dispose) + +Call `dispose()` when you are permanently done with a pipeline to free its native +memory right away, instead of waiting for the garbage collector to finalize the +wrapper. + +Why it matters: the native `GstPipeline` allocation (buffers, decoders, GStreamer +internals) lives outside V8's heap and is invisible to its GC accounting. A +pipeline you simply drop is only reclaimed when GC happens to collect the small JS +wrapper — and with a flat JS heap, V8 feels little pressure to do so. A long +running process that builds a pipeline per recording or transcode can watch RSS +climb steadily while the JS heap stays flat. `dispose()` drives the pipeline to +the NULL state and drops the native reference synchronously, so the memory is +returned immediately. + +```javascript +import { Pipeline } from "gst-kit"; + +const pipeline = new Pipeline("videotestsrc ! theoraenc ! oggmux ! filesink location=out.ogv"); +await pipeline.play(); + +// ... record, then finish cleanly ... +pipeline.endOfStream(); +while (true) { + const msg = await pipeline.busPop(1000); + if (msg?.type === "eos") break; +} +await pipeline.stop(); + +// Release the native pipeline now rather than at some later GC. +pipeline.dispose(); + +// dispose() is terminal and idempotent. Any method call afterward throws +// "Pipeline used after dispose()", and a second dispose() is a harmless no-op. +``` + +**Key points:** + +- **Terminal**: call it once, after `stop()`, when the pipeline will not be used again. +- **Not a graceful stop**: `dispose()` does the hard release. For a clean flush, do `endOfStream()` + wait for EOS and/or `stop()` first. +- **Idempotent**: calling `dispose()` again does nothing. +- **Use-after-dispose is loud**: subsequent method calls throw rather than touching freed memory. + ### Message Bus Handling ```javascript @@ -916,6 +959,9 @@ class Pipeline { // Message handling busPop(timeoutMs?: number): Promise; + + // Lifecycle — release the native pipeline and free its memory immediately + dispose(): void; } ``` @@ -1072,6 +1118,7 @@ gst-kit/ │ ├── appsrc.mjs # AppSrc usage │ ├── appsrc-eos.mjs # AppSrc with end-of-stream │ ├── pipeline-eos.mjs # Pipeline-level end-of-stream +│ ├── dispose.mjs # Releasing a pipeline's native memory │ ├── record-to-file.mjs # Recording to file example │ ├── rtp-timestamp.mjs # RTP handling │ ├── bus.mjs # Message bus handling diff --git a/examples/dispose.mjs b/examples/dispose.mjs new file mode 100644 index 0000000..2626082 --- /dev/null +++ b/examples/dispose.mjs @@ -0,0 +1,56 @@ +#!/usr/bin/env node +/** + * Pipeline Disposal Example + * + * Demonstrates dispose() — releasing a pipeline's native memory eagerly instead + * of waiting for garbage collection to finalize the wrapper. + * + * The native GstPipeline allocation is invisible to V8's heap accounting, so a + * pipeline you simply drop is not reclaimed until GC happens to collect the small + * JS wrapper. A process that builds a pipeline per unit of work (recording, + * transcode, feed) can see RSS climb steadily while the JS heap stays flat. + * dispose() closes that gap: it drives the pipeline to NULL and drops the native + * reference synchronously. + * + * This example builds, plays, stops, and disposes many short-lived pipelines in a + * loop — the exact pattern where relying on GC timing leaks native memory. Run + * with `node --expose-gc examples/dispose.mjs` to also see RSS reported. + */ +import { Pipeline } from "../dist/esm/index.mjs"; + +const ITERATIONS = 25; + +function rssMb() { + return (process.memoryUsage().rss / 1024 / 1024).toFixed(1); +} + +console.log(`📊 RSS before: ${rssMb()} MB`); + +for (let i = 1; i <= ITERATIONS; i++) { + const pipeline = new Pipeline("videotestsrc ! fakesink"); + + await pipeline.play(); + await pipeline.stop(); + + // Release the native pipeline now. Without this, each iteration's native + // allocation would linger until GC decided to collect the wrapper. + pipeline.dispose(); + + // dispose() is idempotent and terminal: a second call is a no-op, and any + // other method call after dispose() throws "Pipeline used after dispose()". + pipeline.dispose(); + try { + pipeline.playing(); + } catch (err) { + if (i === 1) console.log(`🔒 use-after-dispose is guarded: ${err.message}`); + } + + if (i % 5 === 0) console.log(`♻️ disposed ${i}/${ITERATIONS} — RSS: ${rssMb()} MB`); +} + +if (typeof globalThis.gc === "function") { + globalThis.gc(); +} + +console.log(`📊 RSS after: ${rssMb()} MB`); +console.log("✅ Done — native pipelines were released as each one finished."); diff --git a/src/cpp/pipeline.cpp b/src/cpp/pipeline.cpp index 86a3791..292fb07 100644 --- a/src/cpp/pipeline.cpp +++ b/src/cpp/pipeline.cpp @@ -89,6 +89,11 @@ Pipeline::Pipeline(const Napi::CallbackInfo &info) : [this](const Napi::CallbackInfo &info) -> Napi::Value { return this->end_of_stream(info); }, "endOfStream" ); + auto dispose_method = Napi::Function::New( + env, + [this](const Napi::CallbackInfo &info) -> Napi::Value { return this->dispose(info); }, + "dispose" + ); thisObj.DefineProperties( {Napi::PropertyDescriptor::Value("play", play_method, napi_enumerable), @@ -102,10 +107,19 @@ Pipeline::Pipeline(const Napi::CallbackInfo &info) : Napi::PropertyDescriptor::Value("queryDuration", queryDuration_method, napi_enumerable), Napi::PropertyDescriptor::Value("busPop", busPop_method, napi_enumerable), Napi::PropertyDescriptor::Value("seek", seek_method, napi_enumerable), - Napi::PropertyDescriptor::Value("endOfStream", end_of_stream_method, napi_enumerable)} + Napi::PropertyDescriptor::Value("endOfStream", end_of_stream_method, napi_enumerable), + Napi::PropertyDescriptor::Value("dispose", dispose_method, napi_enumerable)} ); } +GstPipeline *Pipeline::require_pipeline(const Napi::Env &env) { + GstPipeline *raw = pipeline.get(); + if (raw == nullptr) { + Napi::Error::New(env, "Pipeline used after dispose()").ThrowAsJavaScriptException(); + } + return raw; +} + Napi::Value Pipeline::play(const Napi::CallbackInfo &info) { Napi::Env env = info.Env(); @@ -123,9 +137,11 @@ Napi::Value Pipeline::play(const Napi::CallbackInfo &info) { } } + GstPipeline *raw = require_pipeline(env); + if (raw == nullptr) return env.Undefined(); + // Create worker and get its promise - StateChangeWorker *worker = - new StateChangeWorker(env, pipeline.get(), GST_STATE_PLAYING, timeout); + StateChangeWorker *worker = new StateChangeWorker(env, raw, GST_STATE_PLAYING, timeout); Napi::Promise promise = worker->GetPromise().Promise(); worker->Queue(); @@ -149,8 +165,11 @@ Napi::Value Pipeline::pause(const Napi::CallbackInfo &info) { } } + GstPipeline *raw = require_pipeline(env); + if (raw == nullptr) return env.Undefined(); + // Create worker and get its promise - StateChangeWorker *worker = new StateChangeWorker(env, pipeline.get(), GST_STATE_PAUSED, timeout); + StateChangeWorker *worker = new StateChangeWorker(env, raw, GST_STATE_PAUSED, timeout); Napi::Promise promise = worker->GetPromise().Promise(); worker->Queue(); @@ -174,8 +193,11 @@ Napi::Value Pipeline::stop(const Napi::CallbackInfo &info) { } } + GstPipeline *raw = require_pipeline(env); + if (raw == nullptr) return env.Undefined(); + // Create worker and get its promise - StateChangeWorker *worker = new StateChangeWorker(env, pipeline.get(), GST_STATE_NULL, timeout); + StateChangeWorker *worker = new StateChangeWorker(env, raw, GST_STATE_NULL, timeout); Napi::Promise promise = worker->GetPromise().Promise(); worker->Queue(); @@ -183,40 +205,56 @@ Napi::Value Pipeline::stop(const Napi::CallbackInfo &info) { } Napi::Value Pipeline::get_element_by_name(const Napi::CallbackInfo &info) { + Napi::Env env = info.Env(); + GstPipeline *raw = require_pipeline(env); + if (raw == nullptr) return env.Undefined(); + auto name = info[0].As().Utf8Value(); - GstElement *e = gst_bin_get_by_name(GST_BIN(pipeline.get()), name.c_str()); + GstElement *e = gst_bin_get_by_name(GST_BIN(raw), name.c_str()); - if (e == nullptr) return info.Env().Null(); + if (e == nullptr) return env.Null(); // Use the stored constructors to create the appropriate element - return Element::CreateFromGstElement(info.Env(), e); + return Element::CreateFromGstElement(env, e); } Napi::Value Pipeline::playing(const Napi::CallbackInfo &info) { + Napi::Env env = info.Env(); + GstPipeline *raw = require_pipeline(env); + if (raw == nullptr) return env.Undefined(); + GstState state; GstState pending; GstStateChangeReturn ret = - gst_element_get_state(GST_ELEMENT(pipeline.get()), &state, &pending, 5 * GST_MSECOND); + gst_element_get_state(GST_ELEMENT(raw), &state, &pending, 5 * GST_MSECOND); // If state change is in progress and we're transitioning to PLAYING, consider it as playing bool is_playing = (state == GST_STATE_PLAYING) || (ret == GST_STATE_CHANGE_ASYNC && pending == GST_STATE_PLAYING); - return Napi::Boolean::New(info.Env(), is_playing); + return Napi::Boolean::New(env, is_playing); } Napi::Value Pipeline::query_position(const Napi::CallbackInfo &info) { + Napi::Env env = info.Env(); + GstPipeline *raw = require_pipeline(env); + if (raw == nullptr) return env.Undefined(); + gint64 pos; - gst_element_query_position(GST_ELEMENT(pipeline.get()), GST_FORMAT_TIME, &pos); + gst_element_query_position(GST_ELEMENT(raw), GST_FORMAT_TIME, &pos); double r = pos == -1 ? -1 : (double)pos / GST_SECOND; - return Napi::Number::New(info.Env(), r); + return Napi::Number::New(env, r); } Napi::Value Pipeline::query_duration(const Napi::CallbackInfo &info) { + Napi::Env env = info.Env(); + GstPipeline *raw = require_pipeline(env); + if (raw == nullptr) return env.Undefined(); + gint64 dur; - gst_element_query_duration(GST_ELEMENT(pipeline.get()), GST_FORMAT_TIME, &dur); + gst_element_query_duration(GST_ELEMENT(raw), GST_FORMAT_TIME, &dur); double r = dur == -1 ? -1 : (double)dur / GST_SECOND; - return Napi::Number::New(info.Env(), r); + return Napi::Number::New(env, r); } Napi::Value Pipeline::bus_pop(const Napi::CallbackInfo &info) { @@ -236,8 +274,11 @@ Napi::Value Pipeline::bus_pop(const Napi::CallbackInfo &info) { } } + GstPipeline *raw = require_pipeline(env); + if (raw == nullptr) return env.Undefined(); + // Create worker and get its promise - BusPopWorker *worker = new BusPopWorker(env, pipeline.get(), timeout); + BusPopWorker *worker = new BusPopWorker(env, raw, timeout); Napi::Promise promise = worker->GetPromise().Promise(); worker->Queue(); @@ -260,12 +301,15 @@ Napi::Value Pipeline::seek(const Napi::CallbackInfo &info) { return env.Undefined(); } + GstPipeline *raw = require_pipeline(env); + if (raw == nullptr) return env.Undefined(); + // Convert seconds to nanoseconds GstClockTime position_ns = static_cast(position_seconds * GST_SECOND); // Perform the seek gboolean result = gst_element_seek( - GST_ELEMENT(pipeline.get()), + GST_ELEMENT(raw), 1.0, // Rate (1.0 = normal speed) GST_FORMAT_TIME, // Format (time-based seeking) GST_SEEK_FLAG_FLUSH, // Flags (flush pipeline) @@ -280,6 +324,8 @@ Napi::Value Pipeline::seek(const Napi::CallbackInfo &info) { Napi::Value Pipeline::end_of_stream(const Napi::CallbackInfo &info) { Napi::Env env = info.Env(); + GstPipeline *raw = require_pipeline(env); + if (raw == nullptr) return env.Undefined(); // Query pipeline state with a 5ms timeout // Note: Sending EOS to a PAUSED pipeline where sinks have not yet prerolled @@ -287,7 +333,7 @@ Napi::Value Pipeline::end_of_stream(const Napi::CallbackInfo &info) { // which waits on the preroll condition. This is a GStreamer-level behavior. GstState state; GstState pending; - gst_element_get_state(GST_ELEMENT(pipeline.get()), &state, &pending, 5 * GST_MSECOND); + gst_element_get_state(GST_ELEMENT(raw), &state, &pending, 5 * GST_MSECOND); // Only send EOS if pipeline is in PLAYING or PAUSED state if (state != GST_STATE_PLAYING && state != GST_STATE_PAUSED) { @@ -295,11 +341,28 @@ Napi::Value Pipeline::end_of_stream(const Napi::CallbackInfo &info) { } // Send EOS event to the pipeline - gboolean result = gst_element_send_event(GST_ELEMENT(pipeline.get()), gst_event_new_eos()); + gboolean result = gst_element_send_event(GST_ELEMENT(raw), gst_event_new_eos()); return Napi::Boolean::New(env, result); } +Napi::Value Pipeline::dispose(const Napi::CallbackInfo &info) { + Napi::Env env = info.Env(); + + // Idempotent: a null pipeline is the already-disposed state + GstPipeline *raw = pipeline.get(); + if (raw == nullptr) return env.Undefined(); + + // Stop the pipeline synchronously, then drop our reference. In-flight async + // workers hold their own gst_object_ref (see async-workers.cpp), so releasing + // here is safe. unique_ptr's deleter (gst_object_unref) frees the native + // GstPipeline once the last reference goes away. + gst_element_set_state(GST_ELEMENT(raw), GST_STATE_NULL); + pipeline.reset(); + + return env.Undefined(); +} + Napi::Value Pipeline::ElementExists(const Napi::CallbackInfo &info) { Napi::Env env = info.Env(); diff --git a/src/cpp/pipeline.hpp b/src/cpp/pipeline.hpp index a976577..4dbb816 100644 --- a/src/cpp/pipeline.hpp +++ b/src/cpp/pipeline.hpp @@ -23,10 +23,12 @@ class Pipeline : public Napi::ObjectWrap { Napi::Value bus_pop(const Napi::CallbackInfo &info); Napi::Value seek(const Napi::CallbackInfo &info); Napi::Value end_of_stream(const Napi::CallbackInfo &info); + Napi::Value dispose(const Napi::CallbackInfo &info); private: std::string pipeline_string; std::unique_ptr pipeline; static bool gst_initialized; static void ensure_gst_initialized(); + GstPipeline *require_pipeline(const Napi::Env &env); }; diff --git a/src/ts/index.ts b/src/ts/index.ts index c9c594b..7049312 100644 --- a/src/ts/index.ts +++ b/src/ts/index.ts @@ -155,6 +155,7 @@ interface Pipeline { busPop(timeoutMs?: number): Promise; seek(positionSeconds: number): boolean; endOfStream(): boolean; + dispose(): void; } interface PipelineConstructor { diff --git a/src/ts/pipeline-dispose.test.ts b/src/ts/pipeline-dispose.test.ts new file mode 100644 index 0000000..bb4d42f --- /dev/null +++ b/src/ts/pipeline-dispose.test.ts @@ -0,0 +1,76 @@ +import { describe, expect, it } from "vitest"; +import { Pipeline } from "."; + +describe("Pipeline dispose()", () => { + it("should dispose a pipeline that was never played", () => { + const pipeline = new Pipeline("videotestsrc ! fakesink"); + + // Disposing a freshly constructed (never played) pipeline is valid and + // should not throw. + expect(() => pipeline.dispose()).not.toThrow(); + }); + + it("should dispose a pipeline after play and stop", async () => { + const pipeline = new Pipeline("videotestsrc ! fakesink"); + + await pipeline.play(); + expect(pipeline.playing()).toBe(true); + await pipeline.stop(); + + expect(() => pipeline.dispose()).not.toThrow(); + }); + + it("should dispose a pipeline directly after play, without an explicit stop", async () => { + const pipeline = new Pipeline("videotestsrc ! fakesink"); + + await pipeline.play(); + expect(pipeline.playing()).toBe(true); + + // dispose() drives the pipeline to NULL itself, so calling it on a still + // playing pipeline is safe and releases the native resources. + expect(() => pipeline.dispose()).not.toThrow(); + }); + + it("should be idempotent — a second dispose() is a no-op", async () => { + const pipeline = new Pipeline("videotestsrc ! fakesink"); + + await pipeline.play(); + await pipeline.stop(); + + pipeline.dispose(); + // Calling dispose() again must not throw. + expect(() => pipeline.dispose()).not.toThrow(); + }); + + it("should throw on synchronous method calls after dispose()", async () => { + const pipeline = new Pipeline("videotestsrc ! fakesink"); + + await pipeline.play(); + await pipeline.stop(); + pipeline.dispose(); + + // Every driver-delegating method routes through the native disposed guard, + // so a use-after-dispose is a loud error rather than a native null-deref. + expect(() => pipeline.playing()).toThrow(/used after dispose/); + expect(() => pipeline.queryPosition()).toThrow(/used after dispose/); + expect(() => pipeline.queryDuration()).toThrow(/used after dispose/); + expect(() => pipeline.getElementByName("sink")).toThrow(/used after dispose/); + expect(() => pipeline.seek(0)).toThrow(/used after dispose/); + expect(() => pipeline.endOfStream()).toThrow(/used after dispose/); + }); + + it("should throw on async method calls after dispose()", async () => { + const pipeline = new Pipeline("videotestsrc ! fakesink"); + + await pipeline.play(); + await pipeline.stop(); + pipeline.dispose(); + + // Async methods throw synchronously (at call time) because the disposed + // guard runs before the worker is queued. + expect(() => pipeline.play()).toThrow(/used after dispose/); + expect(() => pipeline.pause()).toThrow(/used after dispose/); + expect(() => pipeline.stop()).toThrow(/used after dispose/); + expect(() => pipeline.busPop(0)).toThrow(/used after dispose/); + }); +}); From 1a634d2c82aaa903b592ada8047653a838187ef0 Mon Sep 17 00:00:00 2001 From: Jeff Pai Date: Wed, 2 Sep 2026 15:59:03 -0700 Subject: [PATCH 2/7] refactor(pipeline): make dispose() non-blocking, drop reference only MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit dispose() no longer issues a synchronous gst_element_set_state(NULL) on the JS thread — it just drops the owning reference via reset(), letting GStreamer tear the pipeline down when the last ref is released. Removes the event-loop blocking and the ignored state-change return raised in review. Document the stop()-and-release-elements-before-dispose contract and adjust the still-playing dispose test. --- README.md | 6 ++++-- src/cpp/pipeline.cpp | 18 ++++++++++-------- src/ts/pipeline-dispose.test.ts | 7 ++++--- 3 files changed, 18 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 30e75bb..268ba73 100644 --- a/README.md +++ b/README.md @@ -848,8 +848,10 @@ pipeline.dispose(); **Key points:** -- **Terminal**: call it once, after `stop()`, when the pipeline will not be used again. -- **Not a graceful stop**: `dispose()` does the hard release. For a clean flush, do `endOfStream()` + wait for EOS and/or `stop()` first. +- **Stop first**: always `stop()` (or `endOfStream()` + wait for EOS, then `stop()`) before `dispose()`. `dispose()` only drops the native reference; it does not issue a state change, so it must not be the thing that stops a running pipeline. +- **Terminal**: call it once, when the pipeline will not be used again. +- **Not a graceful stop**: `dispose()` does the hard release, not a flush. +- **Release elements first**: any element obtained via `getElementByName()`, and any pad-probe or `onSample()` subscription on it, must be released/unsubscribed before `dispose()`. Elements hold their own reference and are not invalidated by disposing the pipeline. - **Idempotent**: calling `dispose()` again does nothing. - **Use-after-dispose is loud**: subsequent method calls throw rather than touching freed memory. diff --git a/src/cpp/pipeline.cpp b/src/cpp/pipeline.cpp index 292fb07..d3f852b 100644 --- a/src/cpp/pipeline.cpp +++ b/src/cpp/pipeline.cpp @@ -350,14 +350,16 @@ Napi::Value Pipeline::dispose(const Napi::CallbackInfo &info) { Napi::Env env = info.Env(); // Idempotent: a null pipeline is the already-disposed state - GstPipeline *raw = pipeline.get(); - if (raw == nullptr) return env.Undefined(); - - // Stop the pipeline synchronously, then drop our reference. In-flight async - // workers hold their own gst_object_ref (see async-workers.cpp), so releasing - // here is safe. unique_ptr's deleter (gst_object_unref) frees the native - // GstPipeline once the last reference goes away. - gst_element_set_state(GST_ELEMENT(raw), GST_STATE_NULL); + if (pipeline.get() == nullptr) return env.Undefined(); + + // Drop our reference and let unique_ptr's deleter (gst_object_unref) run. This + // does not issue a state change: callers stop() first (see the TS contract), + // and re-issuing set_state(NULL) here would be a synchronous, potentially + // blocking transition on the JS thread — exactly what play/pause/stop offload + // to a StateChangeWorker to avoid. When the last reference goes away GStreamer + // tears the pipeline down to NULL itself. In-flight async workers each hold + // their own gst_object_ref (see async-workers.cpp), so releasing here cannot + // free the native pipeline out from under a running busPop()/state change. pipeline.reset(); return env.Undefined(); diff --git a/src/ts/pipeline-dispose.test.ts b/src/ts/pipeline-dispose.test.ts index bb4d42f..ae7a65d 100644 --- a/src/ts/pipeline-dispose.test.ts +++ b/src/ts/pipeline-dispose.test.ts @@ -20,14 +20,15 @@ describe("Pipeline dispose()", () => { expect(() => pipeline.dispose()).not.toThrow(); }); - it("should dispose a pipeline directly after play, without an explicit stop", async () => { + it("should not throw when disposing a still-playing pipeline", async () => { const pipeline = new Pipeline("videotestsrc ! fakesink"); await pipeline.play(); expect(pipeline.playing()).toBe(true); - // dispose() drives the pipeline to NULL itself, so calling it on a still - // playing pipeline is safe and releases the native resources. + // Callers are expected to stop() first, but dispose() must not blow up if + // they don't: it drops the native reference without issuing a state change, + // and GStreamer tears the pipeline down when the last reference is released. expect(() => pipeline.dispose()).not.toThrow(); }); From 90378200ea24ef71aaf5889202541f06158e583d Mon Sep 17 00:00:00 2001 From: Jeff Pai Date: Wed, 2 Sep 2026 17:28:49 -0700 Subject: [PATCH 3/7] fix: enforce stop-first contract in Pipeline.dispose() dispose() called pipeline.reset() without checking state. GStreamer refuses to tear down a non-NULL element, so disposing a running pipeline leaked the native memory instead of reclaiming it. Now query state and throw "dispose() requires a stopped pipeline" unless the pipeline is already NULL. Update tests to assert the still-playing case throws, add a worker-in-flight test, and correct the README and example that described the removed set_state(NULL) behaviour. --- README.md | 9 +++++---- examples/dispose.mjs | 5 +++-- src/cpp/pipeline.cpp | 35 +++++++++++++++++++++++++-------- src/ts/pipeline-dispose.test.ts | 29 +++++++++++++++++++++++---- 4 files changed, 60 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 268ba73..84e3a12 100644 --- a/README.md +++ b/README.md @@ -821,9 +821,10 @@ internals) lives outside V8's heap and is invisible to its GC accounting. A pipeline you simply drop is only reclaimed when GC happens to collect the small JS wrapper — and with a flat JS heap, V8 feels little pressure to do so. A long running process that builds a pipeline per recording or transcode can watch RSS -climb steadily while the JS heap stays flat. `dispose()` drives the pipeline to -the NULL state and drops the native reference synchronously, so the memory is -returned immediately. +climb steadily while the JS heap stays flat. `dispose()` closes that gap: it +drops the native reference synchronously once the pipeline is stopped, so the +memory is released as soon as the last reference goes away rather than at some +later GC. ```javascript import { Pipeline } from "gst-kit"; @@ -848,7 +849,7 @@ pipeline.dispose(); **Key points:** -- **Stop first**: always `stop()` (or `endOfStream()` + wait for EOS, then `stop()`) before `dispose()`. `dispose()` only drops the native reference; it does not issue a state change, so it must not be the thing that stops a running pipeline. +- **Stop first (required)**: always `stop()` (or `endOfStream()` + wait for EOS, then `stop()`) before `dispose()`. `dispose()` only drops the native reference; it does not issue a state change. GStreamer refuses to tear down a pipeline that is not in the NULL state, so `dispose()` throws (`dispose() requires a stopped pipeline`) if the pipeline is still playing or paused. A pipeline that never left NULL (constructed but never played) can be disposed directly. - **Terminal**: call it once, when the pipeline will not be used again. - **Not a graceful stop**: `dispose()` does the hard release, not a flush. - **Release elements first**: any element obtained via `getElementByName()`, and any pad-probe or `onSample()` subscription on it, must be released/unsubscribed before `dispose()`. Elements hold their own reference and are not invalidated by disposing the pipeline. diff --git a/examples/dispose.mjs b/examples/dispose.mjs index 2626082..875c545 100644 --- a/examples/dispose.mjs +++ b/examples/dispose.mjs @@ -9,8 +9,9 @@ * pipeline you simply drop is not reclaimed until GC happens to collect the small * JS wrapper. A process that builds a pipeline per unit of work (recording, * transcode, feed) can see RSS climb steadily while the JS heap stays flat. - * dispose() closes that gap: it drives the pipeline to NULL and drops the native - * reference synchronously. + * dispose() closes that gap: after the pipeline is stopped, it drops the native + * reference synchronously so the memory is released without waiting for GC. + * dispose() does not stop the pipeline — call stop() first, or it throws. * * This example builds, plays, stops, and disposes many short-lived pipelines in a * loop — the exact pattern where relying on GC timing leaks native memory. Run diff --git a/src/cpp/pipeline.cpp b/src/cpp/pipeline.cpp index d3f852b..c78d1f8 100644 --- a/src/cpp/pipeline.cpp +++ b/src/cpp/pipeline.cpp @@ -352,14 +352,33 @@ Napi::Value Pipeline::dispose(const Napi::CallbackInfo &info) { // Idempotent: a null pipeline is the already-disposed state if (pipeline.get() == nullptr) return env.Undefined(); - // Drop our reference and let unique_ptr's deleter (gst_object_unref) run. This - // does not issue a state change: callers stop() first (see the TS contract), - // and re-issuing set_state(NULL) here would be a synchronous, potentially - // blocking transition on the JS thread — exactly what play/pause/stop offload - // to a StateChangeWorker to avoid. When the last reference goes away GStreamer - // tears the pipeline down to NULL itself. In-flight async workers each hold - // their own gst_object_ref (see async-workers.cpp), so releasing here cannot - // free the native pipeline out from under a running busPop()/state change. + // dispose() only drops the native reference; it does not issue a state change. + // A state change to NULL is a synchronous, potentially blocking transition on + // the JS thread — exactly what play/pause/stop offload to a StateChangeWorker + // to avoid. So dispose() requires the caller to have stopped the pipeline + // first (see the TS contract). + // + // Enforce that contract: GStreamer's gst_element_dispose refuses to tear down + // an element that is not in the NULL state — it emits a g_critical and returns + // without releasing pads, bus, clock and contexts, which leaks the native + // pipeline (and aborts under G_DEBUG=fatal-criticals). Rather than silently + // leak, fail loudly so the caller stops the pipeline before disposing. + GstState state; + GstState pending; + gst_element_get_state(GST_ELEMENT(pipeline.get()), &state, &pending, 0); + if (state != GST_STATE_NULL || pending != GST_STATE_VOID_PENDING) { + Napi::Error::New( + env, "dispose() requires a stopped pipeline: call stop() before dispose()" + ) + .ThrowAsJavaScriptException(); + return env.Undefined(); + } + + // Drop our reference and let unique_ptr's deleter (gst_object_unref) run. When + // the last reference goes away GStreamer finalizes the now-NULL pipeline. + // In-flight async workers each hold their own gst_object_ref (see + // async-workers.cpp), so releasing here cannot free the native pipeline out + // from under a running busPop()/state change. pipeline.reset(); return env.Undefined(); diff --git a/src/ts/pipeline-dispose.test.ts b/src/ts/pipeline-dispose.test.ts index ae7a65d..c929eb6 100644 --- a/src/ts/pipeline-dispose.test.ts +++ b/src/ts/pipeline-dispose.test.ts @@ -20,15 +20,36 @@ describe("Pipeline dispose()", () => { expect(() => pipeline.dispose()).not.toThrow(); }); - it("should not throw when disposing a still-playing pipeline", async () => { + it("should throw when disposing a still-playing pipeline", async () => { const pipeline = new Pipeline("videotestsrc ! fakesink"); await pipeline.play(); expect(pipeline.playing()).toBe(true); - // Callers are expected to stop() first, but dispose() must not blow up if - // they don't: it drops the native reference without issuing a state change, - // and GStreamer tears the pipeline down when the last reference is released. + // dispose() only drops the native reference; it does not issue a state + // change. GStreamer refuses to tear down a non-NULL element, so disposing a + // still-playing pipeline would leak. dispose() enforces the stop-first + // contract by throwing instead. + expect(() => pipeline.dispose()).toThrow(/stop\(\) before dispose\(\)/); + + // The pipeline is still usable after the rejected dispose(); stop then + // dispose cleanly. + await pipeline.stop(); + expect(() => pipeline.dispose()).not.toThrow(); + }); + + it("should be safe to dispose after a worker started against the pipeline resolves", async () => { + const pipeline = new Pipeline("videotestsrc ! fakesink"); + + await pipeline.play(); + + // Start an async worker (busPop) that holds its own gst_object_ref while it + // runs. The pipeline is stopped and the worker awaited before dispose(), so + // the reference the worker held is already released. + const pending = pipeline.busPop(1000); + await pipeline.stop(); + await pending; + expect(() => pipeline.dispose()).not.toThrow(); }); From 8c5cb164d8b069d224518588448eef3c3435618d Mon Sep 17 00:00:00 2001 From: Jeff Pai Date: Thu, 3 Sep 2026 19:31:08 -0700 Subject: [PATCH 4/7] fix(pipeline): dispose() forces NULL then frees instead of throwing (prevents native leak) --- src/cpp/pipeline.cpp | 33 ++++++++++++++++++--------------- src/ts/pipeline-dispose.test.ts | 19 +++++++++---------- 2 files changed, 27 insertions(+), 25 deletions(-) diff --git a/src/cpp/pipeline.cpp b/src/cpp/pipeline.cpp index c78d1f8..55dd7c2 100644 --- a/src/cpp/pipeline.cpp +++ b/src/cpp/pipeline.cpp @@ -353,25 +353,28 @@ Napi::Value Pipeline::dispose(const Napi::CallbackInfo &info) { if (pipeline.get() == nullptr) return env.Undefined(); // dispose() only drops the native reference; it does not issue a state change. - // A state change to NULL is a synchronous, potentially blocking transition on - // the JS thread — exactly what play/pause/stop offload to a StateChangeWorker - // to avoid. So dispose() requires the caller to have stopped the pipeline - // first (see the TS contract). - // - // Enforce that contract: GStreamer's gst_element_dispose refuses to tear down - // an element that is not in the NULL state — it emits a g_critical and returns - // without releasing pads, bus, clock and contexts, which leaks the native - // pipeline (and aborts under G_DEBUG=fatal-criticals). Rather than silently - // leak, fail loudly so the caller stops the pipeline before disposing. + // The native pipeline must reach NULL before its reference is dropped: + // gst_object_unref on a non-NULL pipeline leaks its pads, bus, clock, and the + // elements' internal buffers (and can emit a g_critical). Callers stop() + // first, so this is normally already NULL. But stop() is bounded by a timeout + // and its NULL transition may not have fully settled by the time dispose() + // runs — so rather than throw and skip the unref (which guarantees the leak we + // are trying to prevent), force the pipeline to NULL here, synchronously, and + // then release. Setting an already-NULL pipeline to NULL is a cheap no-op, so + // the common stopped-first path pays almost nothing; the blocking transition + // only happens on the rare not-fully-stopped path, where it is exactly what + // prevents the leak. GstState state; GstState pending; gst_element_get_state(GST_ELEMENT(pipeline.get()), &state, &pending, 0); if (state != GST_STATE_NULL || pending != GST_STATE_VOID_PENDING) { - Napi::Error::New( - env, "dispose() requires a stopped pipeline: call stop() before dispose()" - ) - .ThrowAsJavaScriptException(); - return env.Undefined(); + gst_element_set_state(GST_ELEMENT(pipeline.get()), GST_STATE_NULL); + // Block until the NULL transition completes so the unref below finalizes a + // truly-NULL pipeline. NULL is reached synchronously for virtually all + // pipelines; the wait is bounded so a pathological element cannot hang here. + gst_element_get_state( + GST_ELEMENT(pipeline.get()), &state, &pending, 5 * GST_SECOND + ); } // Drop our reference and let unique_ptr's deleter (gst_object_unref) run. When diff --git a/src/ts/pipeline-dispose.test.ts b/src/ts/pipeline-dispose.test.ts index c929eb6..5423304 100644 --- a/src/ts/pipeline-dispose.test.ts +++ b/src/ts/pipeline-dispose.test.ts @@ -20,22 +20,21 @@ describe("Pipeline dispose()", () => { expect(() => pipeline.dispose()).not.toThrow(); }); - it("should throw when disposing a still-playing pipeline", async () => { + it("should dispose a still-playing pipeline by forcing it to NULL first", async () => { const pipeline = new Pipeline("videotestsrc ! fakesink"); await pipeline.play(); expect(pipeline.playing()).toBe(true); - // dispose() only drops the native reference; it does not issue a state - // change. GStreamer refuses to tear down a non-NULL element, so disposing a - // still-playing pipeline would leak. dispose() enforces the stop-first - // contract by throwing instead. - expect(() => pipeline.dispose()).toThrow(/stop\(\) before dispose\(\)/); - - // The pipeline is still usable after the rejected dispose(); stop then - // dispose cleanly. - await pipeline.stop(); + // Callers are expected to stop() first, but dispose() must not leak if they + // don't: rather than throw (which would skip the unref and leak the native + // pipeline), dispose() drives the pipeline to NULL synchronously and then + // releases it. So disposing a still-playing pipeline is safe and does not + // throw. expect(() => pipeline.dispose()).not.toThrow(); + + // After dispose() the pipeline is released; any further use throws. + expect(() => pipeline.playing()).toThrow(/used after dispose/); }); it("should be safe to dispose after a worker started against the pipeline resolves", async () => { From 257a5f72ba9d3aade9da1cb84310ce40ed2535c3 Mon Sep 17 00:00:00 2001 From: Jeff Pai Date: Thu, 3 Sep 2026 20:09:14 -0700 Subject: [PATCH 5/7] fix(test): drain bus to EOS before stop() to avoid basesrc race on Windows Sending EOS to a running source and calling stop() immediately raced GStreamer's internal basesrc has_pending_eos handling, aborting the vitest worker fork on Windows CI (all JS assertions passed but the fork died). Add a waitForEos helper and await EOS on the bus before stop() in the affected EOS tests so the source streaming thread unwinds its loop cleanly before the state teardown. --- src/ts/appsrc-eos.test.ts | 25 +++++++++---------------- src/ts/pipeline-eos.test.ts | 9 ++++++--- src/ts/test-utils.ts | 24 ++++++++++++++++++++++++ 3 files changed, 39 insertions(+), 19 deletions(-) diff --git a/src/ts/appsrc-eos.test.ts b/src/ts/appsrc-eos.test.ts index b29547a..c55d0d8 100644 --- a/src/ts/appsrc-eos.test.ts +++ b/src/ts/appsrc-eos.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "vitest"; -import { Pipeline, type GstMessage } from "."; +import { Pipeline } from "."; +import { waitForEos } from "./test-utils"; describe("AppSrc End-of-Stream", () => { it("should send EOS signal through endOfStream method", async () => { @@ -28,21 +29,9 @@ describe("AppSrc End-of-Stream", () => { // Send end-of-stream source.endOfStream(); - // Wait for EOS message on the bus - let eosReceived = false; - let attempts = 0; - const maxAttempts = 20; // Increased attempts - - while (!eosReceived && attempts < maxAttempts) { - const message: GstMessage | null = await pipeline.busPop(1000); // Increased timeout - attempts++; - - if (message?.type === "eos") { - eosReceived = true; - expect(message.type).toBe("eos"); - break; - } - } + // Wait for EOS to reach the bus before stopping, so the source loop + // unwinds cleanly instead of racing the state teardown. + const eosReceived = await waitForEos(pipeline, { attempts: 20, timeoutMs: 1000 }); await pipeline.stop(); @@ -92,6 +81,10 @@ describe("AppSrc End-of-Stream", () => { expect(error).toBeDefined(); } + // Drain to EOS before stopping so the source streaming thread has finished + // its loop; stopping mid-loop races GStreamer's internal EOS handling. + await waitForEos(pipeline, { attempts: 20, timeoutMs: 1000 }); + await pipeline.stop(); } }); diff --git a/src/ts/pipeline-eos.test.ts b/src/ts/pipeline-eos.test.ts index a344f4a..e6d6d8b 100644 --- a/src/ts/pipeline-eos.test.ts +++ b/src/ts/pipeline-eos.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "vitest"; import { Pipeline, type GstMessage } from "."; +import { waitForEos } from "./test-utils"; describe("Pipeline EOS - State-Gated Dispatch", () => { it("should return true when pipeline is in PLAYING state", async () => { @@ -10,7 +11,9 @@ describe("Pipeline EOS - State-Gated Dispatch", () => { const result = pipeline.endOfStream(); expect(result).toBe(true); - await new Promise(resolve => setTimeout(resolve, 30)); + // Let EOS propagate to the bus before stopping so the source loop unwinds + // cleanly rather than racing the state teardown. + await waitForEos(pipeline); await pipeline.stop(); }); @@ -80,7 +83,7 @@ describe("Pipeline EOS - State-Gated Dispatch", () => { expect(result2).toBe(true); expect(result3).toBe(true); - await new Promise(resolve => setTimeout(resolve, 30)); + await waitForEos(pipeline); await pipeline.stop(); }); @@ -93,7 +96,7 @@ describe("Pipeline EOS - State-Gated Dispatch", () => { const resultWhilePlaying = pipeline.endOfStream(); expect(resultWhilePlaying).toBe(true); - await new Promise(resolve => setTimeout(resolve, 30)); + await waitForEos(pipeline); await pipeline.stop(); diff --git a/src/ts/test-utils.ts b/src/ts/test-utils.ts index 2939349..b051862 100644 --- a/src/ts/test-utils.ts +++ b/src/ts/test-utils.ts @@ -20,3 +20,27 @@ export const arePluginsAvailable = (plugins: string[]) => plugins.every(plugin => isPluginAvailable(plugin)); export const isWindows = process.platform === "win32"; + +/** + * Drain the pipeline bus until an EOS (or error) message is seen, or the budget + * is exhausted. + * + * After sending EOS to a source, the source's streaming thread finishes its + * current loop and posts EOS on the bus. Calling stop() before that settles can + * race GStreamer's internal basesrc EOS handling (observed on Windows as a + * `has_pending_eos` assertion abort). Waiting for EOS to reach the bus lets the + * source loop unwind cleanly before the state teardown. + * + * Returns true if EOS was observed, false otherwise. + */ +export const waitForEos = async ( + pipeline: Pipeline, + { attempts = 20, timeoutMs = 500 }: { attempts?: number; timeoutMs?: number } = {} +): Promise => { + for (let i = 0; i < attempts; i++) { + const message = await pipeline.busPop(timeoutMs); + if (message?.type === "eos") return true; + if (message?.type === "error") return false; + } + return false; +}; From 305a7136290f970b18fe759f06197a1c98fdb8b3 Mon Sep 17 00:00:00 2001 From: Jeff Pai Date: Fri, 4 Sep 2026 16:13:46 -0700 Subject: [PATCH 6/7] refactor(pipeline): dedupe state-change methods and timeout parsing play/pause/stop were identical except for the target state; extract a shared queue_state_change() helper so each is a one-liner. Pull the repeated timeout-parsing block into parse_timeout() (reused by bus_pop), and collapse the duplicated throw in require_pipeline() into a single condition. No behavior change. --- README.md | 8 +- examples/dispose.mjs | 13 ++- src/cpp/async-workers.cpp | 25 ++++- src/cpp/async-workers.hpp | 8 +- src/cpp/pipeline.cpp | 191 +++++++++++++++----------------- src/cpp/pipeline.hpp | 8 ++ src/ts/pipeline-dispose.test.ts | 58 ++++++++++ src/ts/pipeline-eos.test.ts | 9 +- 8 files changed, 203 insertions(+), 117 deletions(-) diff --git a/README.md b/README.md index 84e3a12..c135b59 100644 --- a/README.md +++ b/README.md @@ -822,9 +822,9 @@ pipeline you simply drop is only reclaimed when GC happens to collect the small wrapper — and with a flat JS heap, V8 feels little pressure to do so. A long running process that builds a pipeline per recording or transcode can watch RSS climb steadily while the JS heap stays flat. `dispose()` closes that gap: it -drops the native reference synchronously once the pipeline is stopped, so the -memory is released as soon as the last reference goes away rather than at some -later GC. +drives the pipeline to NULL if it is not already there and drops the native +reference synchronously, so the memory is released as soon as the last reference +goes away rather than at some later GC. ```javascript import { Pipeline } from "gst-kit"; @@ -849,7 +849,7 @@ pipeline.dispose(); **Key points:** -- **Stop first (required)**: always `stop()` (or `endOfStream()` + wait for EOS, then `stop()`) before `dispose()`. `dispose()` only drops the native reference; it does not issue a state change. GStreamer refuses to tear down a pipeline that is not in the NULL state, so `dispose()` throws (`dispose() requires a stopped pipeline`) if the pipeline is still playing or paused. A pipeline that never left NULL (constructed but never played) can be disposed directly. +- **Stop first (recommended)**: prefer `stop()` (or `endOfStream()` + wait for EOS, then `stop()`) before `dispose()`. If the pipeline has not reached the NULL state, `dispose()` drives it there synchronously before releasing it — GStreamer cannot cleanly free a non-NULL pipeline. That transition is normally instant, but on a not-fully-stopped pipeline it runs on the calling thread and is bounded by a 5s wait, so stopping first keeps `dispose()` cheap and non-blocking. - **Terminal**: call it once, when the pipeline will not be used again. - **Not a graceful stop**: `dispose()` does the hard release, not a flush. - **Release elements first**: any element obtained via `getElementByName()`, and any pad-probe or `onSample()` subscription on it, must be released/unsubscribed before `dispose()`. Elements hold their own reference and are not invalidated by disposing the pipeline. diff --git a/examples/dispose.mjs b/examples/dispose.mjs index 875c545..5fb8fb1 100644 --- a/examples/dispose.mjs +++ b/examples/dispose.mjs @@ -9,13 +9,16 @@ * pipeline you simply drop is not reclaimed until GC happens to collect the small * JS wrapper. A process that builds a pipeline per unit of work (recording, * transcode, feed) can see RSS climb steadily while the JS heap stays flat. - * dispose() closes that gap: after the pipeline is stopped, it drops the native - * reference synchronously so the memory is released without waiting for GC. - * dispose() does not stop the pipeline — call stop() first, or it throws. + * dispose() closes that gap: it drives the pipeline to NULL if it is not already + * there and drops the native reference synchronously, so the memory is released + * without waiting for GC. Stopping first is recommended so dispose() stays cheap + * and non-blocking, but dispose() will force the NULL transition itself if + * needed. * * This example builds, plays, stops, and disposes many short-lived pipelines in a - * loop — the exact pattern where relying on GC timing leaks native memory. Run - * with `node --expose-gc examples/dispose.mjs` to also see RSS reported. + * loop — the exact pattern where relying on GC timing leaks native memory. RSS is + * always reported; run with `node --expose-gc examples/dispose.mjs` to also force + * a GC at the end so the "after" figure excludes collectable JS wrappers. */ import { Pipeline } from "../dist/esm/index.mjs"; diff --git a/src/cpp/async-workers.cpp b/src/cpp/async-workers.cpp index 4be5e3d..8fde89c 100644 --- a/src/cpp/async-workers.cpp +++ b/src/cpp/async-workers.cpp @@ -1,4 +1,5 @@ #include "async-workers.hpp" +#include "pipeline.hpp" #include "type-conversion.hpp" #include @@ -195,11 +196,12 @@ void PullSampleWorker::cleanup() { // StateChangeWorker implementation StateChangeWorker::StateChangeWorker( - const Napi::Env &env, GstPipeline *pipeline, GstState target_state, GstClockTime timeout + const Napi::Env &env, Pipeline *owner, GstPipeline *pipeline, GstState target_state, + GstClockTime timeout ) : - Napi::AsyncWorker(env), pipeline(pipeline), target_state(target_state), timeout(timeout), - state_change_result(GST_STATE_CHANGE_FAILURE), final_state(GST_STATE_VOID_PENDING), - deferred(env) { + Napi::AsyncWorker(env), owner(owner), pipeline(pipeline), target_state(target_state), + timeout(timeout), state_change_result(GST_STATE_CHANGE_FAILURE), + final_state(GST_STATE_VOID_PENDING), deferred(env), owner_notified(false) { // Increase reference count since we'll be using this in another thread gst_object_ref(pipeline); } @@ -257,11 +259,17 @@ void StateChangeWorker::OnOK() { result.Set("finalState", Napi::Number::New(Env(), final_state)); result.Set("targetState", Napi::Number::New(Env(), target_state)); + // Notify the owner before resolving so any deferred dispose() teardown runs + // while this worker's own reference is still held — the release then sees the + // last reference drop and finalizes a truly-NULL pipeline. + notify_owner_finished(); + deferred.Resolve(result); } void StateChangeWorker::OnError(const Napi::Error &error) { Napi::HandleScope scope(Env()); + notify_owner_finished(); deferred.Reject(error.Value()); } @@ -271,3 +279,12 @@ void StateChangeWorker::cleanup() { pipeline = nullptr; } } + +void StateChangeWorker::notify_owner_finished() { + // Runs on the JS thread (OnOK/OnError). Idempotent: OnOK and OnError are + // mutually exclusive, but guard anyway so the owner count can never be + // decremented twice for one worker. + if (owner_notified) return; + owner_notified = true; + owner->state_worker_finished(); +} diff --git a/src/cpp/async-workers.hpp b/src/cpp/async-workers.hpp index 63fae0f..a2d57c4 100644 --- a/src/cpp/async-workers.hpp +++ b/src/cpp/async-workers.hpp @@ -53,11 +53,14 @@ class PullSampleWorker : public Napi::AsyncWorker { Napi::Promise::Deferred deferred; }; +class Pipeline; + // AsyncWorker for pipeline state changes with timeout class StateChangeWorker : public Napi::AsyncWorker { public: StateChangeWorker( - const Napi::Env &env, GstPipeline *pipeline, GstState target_state, GstClockTime timeout + const Napi::Env &env, Pipeline *owner, GstPipeline *pipeline, GstState target_state, + GstClockTime timeout ); ~StateChangeWorker(); @@ -69,11 +72,14 @@ class StateChangeWorker : public Napi::AsyncWorker { private: void cleanup(); + void notify_owner_finished(); + Pipeline *owner; GstPipeline *pipeline; GstState target_state; GstClockTime timeout; GstStateChangeReturn state_change_result; GstState final_state; Napi::Promise::Deferred deferred; + bool owner_notified; }; diff --git a/src/cpp/pipeline.cpp b/src/cpp/pipeline.cpp index 55dd7c2..d78159b 100644 --- a/src/cpp/pipeline.cpp +++ b/src/cpp/pipeline.cpp @@ -23,7 +23,8 @@ Napi::Object Pipeline::Init(const Napi::Env &env, const Napi::Object &exports) { } Pipeline::Pipeline(const Napi::CallbackInfo &info) : - Napi::ObjectWrap(info), pipeline(nullptr, gst_object_unref) { + Napi::ObjectWrap(info), pipeline(nullptr, gst_object_unref), disposed(false), + in_flight_state_changes(0) { ensure_gst_initialized(); Napi::Env env = info.Env(); GError *err = NULL; @@ -113,44 +114,70 @@ Pipeline::Pipeline(const Napi::CallbackInfo &info) : } GstPipeline *Pipeline::require_pipeline(const Napi::Env &env) { - GstPipeline *raw = pipeline.get(); - if (raw == nullptr) { + // `disposed` is the authoritative sentinel, not a null `pipeline`: when + // dispose() runs while a state-change worker is in flight the native teardown + // is deferred, so the pointer can still be non-null after dispose() returns. + // Guarding on the flag keeps use-after-dispose loud in that window too. + if (disposed || pipeline.get() == nullptr) { Napi::Error::New(env, "Pipeline used after dispose()").ThrowAsJavaScriptException(); + return nullptr; } - return raw; + return pipeline.get(); } -Napi::Value Pipeline::play(const Napi::CallbackInfo &info) { - Napi::Env env = info.Env(); - - // Default timeout is 1000ms (1 second) - GstClockTime timeout = 1000 * GST_MSECOND; +void Pipeline::release_native_pipeline() { + GstPipeline *raw = pipeline.get(); + if (raw == nullptr) return; - // Check if timeout parameter is provided - if (info.Length() > 0 && info[0].IsNumber()) { - double timeout_ms = info[0].As().DoubleValue(); - if (timeout_ms < 0) { - // Negative timeout means infinite wait - timeout = GST_CLOCK_TIME_NONE; - } else { - timeout = static_cast(timeout_ms * GST_MSECOND); - } + // The native pipeline must reach NULL before its reference is dropped: + // gst_object_unref on a non-NULL pipeline leaks its pads, bus, clock, and the + // elements' internal buffers (and can emit a g_critical). Callers stop() + // first, so this is normally already NULL. But stop() is bounded by a timeout + // and its NULL transition may not have fully settled — so rather than skip the + // unref (which guarantees the leak we are trying to prevent), force the + // pipeline to NULL here, synchronously, and then release. Setting an + // already-NULL pipeline to NULL is a cheap no-op, so the common stopped-first + // path pays almost nothing; the blocking transition only happens on the rare + // not-fully-stopped path, where it is exactly what prevents the leak. + GstState state; + GstState pending; + gst_element_get_state(GST_ELEMENT(raw), &state, &pending, 0); + if (state != GST_STATE_NULL || pending != GST_STATE_VOID_PENDING) { + gst_element_set_state(GST_ELEMENT(raw), GST_STATE_NULL); + // Block until the NULL transition completes so the unref below finalizes a + // truly-NULL pipeline. NULL is reached synchronously for virtually all + // pipelines; the wait is bounded so a pathological element cannot hang here. + gst_element_get_state(GST_ELEMENT(raw), &state, &pending, 5 * GST_SECOND); } - GstPipeline *raw = require_pipeline(env); - if (raw == nullptr) return env.Undefined(); - - // Create worker and get its promise - StateChangeWorker *worker = new StateChangeWorker(env, raw, GST_STATE_PLAYING, timeout); - Napi::Promise promise = worker->GetPromise().Promise(); - worker->Queue(); + // Drop our reference and let unique_ptr's deleter (gst_object_unref) run. + pipeline.reset(); +} - return promise; +void Pipeline::state_worker_started() { + // Keep the JS wrapper alive for as long as a worker holds a back-pointer to + // this Pipeline, so the worker's OnOK/OnError can safely call back into it + // even if all JS references are dropped while the state change is in flight. + if (in_flight_state_changes == 0) Ref(); + in_flight_state_changes++; } -Napi::Value Pipeline::pause(const Napi::CallbackInfo &info) { - Napi::Env env = info.Env(); +void Pipeline::state_worker_finished() { + if (in_flight_state_changes > 0) in_flight_state_changes--; + // If dispose() was requested while workers were running, the last worker to + // finish performs the deferred native teardown. By now no state change can + // drive the pipeline back up, so forcing NULL and releasing is safe. + if (disposed && in_flight_state_changes == 0) { + release_native_pipeline(); + } + + // Balance the Ref() taken in state_worker_started() once the last worker is + // done. This may finalize the wrapper, so touch no members afterward. + if (in_flight_state_changes == 0) Unref(); +} + +GstClockTime Pipeline::parse_timeout(const Napi::CallbackInfo &info) { // Default timeout is 1000ms (1 second) GstClockTime timeout = 1000 * GST_MSECOND; @@ -165,45 +192,38 @@ Napi::Value Pipeline::pause(const Napi::CallbackInfo &info) { } } - GstPipeline *raw = require_pipeline(env); - if (raw == nullptr) return env.Undefined(); - - // Create worker and get its promise - StateChangeWorker *worker = new StateChangeWorker(env, raw, GST_STATE_PAUSED, timeout); - Napi::Promise promise = worker->GetPromise().Promise(); - worker->Queue(); - - return promise; + return timeout; } -Napi::Value Pipeline::stop(const Napi::CallbackInfo &info) { +Napi::Value Pipeline::queue_state_change(const Napi::CallbackInfo &info, GstState target_state) { Napi::Env env = info.Env(); - // Default timeout is 1000ms (1 second) - GstClockTime timeout = 1000 * GST_MSECOND; - - // Check if timeout parameter is provided - if (info.Length() > 0 && info[0].IsNumber()) { - double timeout_ms = info[0].As().DoubleValue(); - if (timeout_ms < 0) { - // Negative timeout means infinite wait - timeout = GST_CLOCK_TIME_NONE; - } else { - timeout = static_cast(timeout_ms * GST_MSECOND); - } - } + GstClockTime timeout = parse_timeout(info); GstPipeline *raw = require_pipeline(env); if (raw == nullptr) return env.Undefined(); // Create worker and get its promise - StateChangeWorker *worker = new StateChangeWorker(env, raw, GST_STATE_NULL, timeout); + StateChangeWorker *worker = new StateChangeWorker(env, this, raw, target_state, timeout); Napi::Promise promise = worker->GetPromise().Promise(); + state_worker_started(); worker->Queue(); return promise; } +Napi::Value Pipeline::play(const Napi::CallbackInfo &info) { + return queue_state_change(info, GST_STATE_PLAYING); +} + +Napi::Value Pipeline::pause(const Napi::CallbackInfo &info) { + return queue_state_change(info, GST_STATE_PAUSED); +} + +Napi::Value Pipeline::stop(const Napi::CallbackInfo &info) { + return queue_state_change(info, GST_STATE_NULL); +} + Napi::Value Pipeline::get_element_by_name(const Napi::CallbackInfo &info) { Napi::Env env = info.Env(); GstPipeline *raw = require_pipeline(env); @@ -260,19 +280,7 @@ Napi::Value Pipeline::query_duration(const Napi::CallbackInfo &info) { Napi::Value Pipeline::bus_pop(const Napi::CallbackInfo &info) { Napi::Env env = info.Env(); - // Default timeout is 1000ms (1 second) - converted to nanoseconds - GstClockTime timeout = 1000 * GST_MSECOND; - - // Check if timeout parameter is provided - if (info.Length() > 0 && info[0].IsNumber()) { - double timeout_ms = info[0].As().DoubleValue(); - if (timeout_ms < 0) { - // Negative timeout means infinite wait - timeout = GST_CLOCK_TIME_NONE; - } else { - timeout = static_cast(timeout_ms * GST_MSECOND); - } - } + GstClockTime timeout = parse_timeout(info); GstPipeline *raw = require_pipeline(env); if (raw == nullptr) return env.Undefined(); @@ -349,40 +357,25 @@ Napi::Value Pipeline::end_of_stream(const Napi::CallbackInfo &info) { Napi::Value Pipeline::dispose(const Napi::CallbackInfo &info) { Napi::Env env = info.Env(); - // Idempotent: a null pipeline is the already-disposed state - if (pipeline.get() == nullptr) return env.Undefined(); - - // dispose() only drops the native reference; it does not issue a state change. - // The native pipeline must reach NULL before its reference is dropped: - // gst_object_unref on a non-NULL pipeline leaks its pads, bus, clock, and the - // elements' internal buffers (and can emit a g_critical). Callers stop() - // first, so this is normally already NULL. But stop() is bounded by a timeout - // and its NULL transition may not have fully settled by the time dispose() - // runs — so rather than throw and skip the unref (which guarantees the leak we - // are trying to prevent), force the pipeline to NULL here, synchronously, and - // then release. Setting an already-NULL pipeline to NULL is a cheap no-op, so - // the common stopped-first path pays almost nothing; the blocking transition - // only happens on the rare not-fully-stopped path, where it is exactly what - // prevents the leak. - GstState state; - GstState pending; - gst_element_get_state(GST_ELEMENT(pipeline.get()), &state, &pending, 0); - if (state != GST_STATE_NULL || pending != GST_STATE_VOID_PENDING) { - gst_element_set_state(GST_ELEMENT(pipeline.get()), GST_STATE_NULL); - // Block until the NULL transition completes so the unref below finalizes a - // truly-NULL pipeline. NULL is reached synchronously for virtually all - // pipelines; the wait is bounded so a pathological element cannot hang here. - gst_element_get_state( - GST_ELEMENT(pipeline.get()), &state, &pending, 5 * GST_SECOND - ); - } - - // Drop our reference and let unique_ptr's deleter (gst_object_unref) run. When - // the last reference goes away GStreamer finalizes the now-NULL pipeline. - // In-flight async workers each hold their own gst_object_ref (see - // async-workers.cpp), so releasing here cannot free the native pipeline out - // from under a running busPop()/state change. - pipeline.reset(); + // Idempotent: once disposed, further calls are no-ops. + if (disposed) return env.Undefined(); + + // Mark disposed immediately so use-after-dispose is guarded and a second + // dispose() is a no-op, regardless of whether the native teardown runs now or + // is deferred below. + disposed = true; + + // dispose() drives the pipeline to NULL and drops the native reference. It + // must not do that while a state-change worker (play/pause/stop) is still in + // flight: that worker holds its own reference and can drive the state back up + // after we force NULL, leaving it to finalize a non-NULL pipeline — which + // leaks exactly what dispose() exists to reclaim. So when a worker is in + // flight, defer the teardown; the last worker to finish runs it (see + // state_worker_finished()). In-flight busPop()/pull workers do not change + // state, so they do not need to gate the teardown. + if (in_flight_state_changes > 0) return env.Undefined(); + + release_native_pipeline(); return env.Undefined(); } diff --git a/src/cpp/pipeline.hpp b/src/cpp/pipeline.hpp index 4dbb816..7c5fb93 100644 --- a/src/cpp/pipeline.hpp +++ b/src/cpp/pipeline.hpp @@ -25,10 +25,18 @@ class Pipeline : public Napi::ObjectWrap { Napi::Value end_of_stream(const Napi::CallbackInfo &info); Napi::Value dispose(const Napi::CallbackInfo &info); + void state_worker_started(); + void state_worker_finished(); + private: std::string pipeline_string; std::unique_ptr pipeline; + bool disposed; + int in_flight_state_changes; static bool gst_initialized; static void ensure_gst_initialized(); + static GstClockTime parse_timeout(const Napi::CallbackInfo &info); GstPipeline *require_pipeline(const Napi::Env &env); + Napi::Value queue_state_change(const Napi::CallbackInfo &info, GstState target_state); + void release_native_pipeline(); }; diff --git a/src/ts/pipeline-dispose.test.ts b/src/ts/pipeline-dispose.test.ts index 5423304..1687a0d 100644 --- a/src/ts/pipeline-dispose.test.ts +++ b/src/ts/pipeline-dispose.test.ts @@ -52,6 +52,64 @@ describe("Pipeline dispose()", () => { expect(() => pipeline.dispose()).not.toThrow(); }); + it("should dispose safely while a state-change worker is still in flight", async () => { + const pipeline = new Pipeline("videotestsrc ! videoconvert ! queue ! fakesink"); + + // Start play() but do NOT await it: the StateChangeWorker is queued and will + // run on a background thread after dispose() returns. dispose() must defer + // its native teardown until that worker finishes, otherwise the worker + // drives the state back up and finalizes a non-NULL pipeline (a leak). + const playing = pipeline.play(); + + // dispose() while the worker is in flight — marks disposed immediately and + // defers the release. + expect(() => pipeline.dispose()).not.toThrow(); + + // The pipeline is observably disposed right away. + expect(() => pipeline.playing()).toThrow(/used after dispose/); + + // Awaiting the in-flight worker completes without error; the deferred + // teardown runs when it finishes. + await expect(playing).resolves.toBeDefined(); + + // Still disposed, and a second dispose() is a no-op. + expect(() => pipeline.dispose()).not.toThrow(); + }); + + it("should not grow RSS when disposing with a state change in flight repeatedly", async () => { + // Guards finding 1: a queued state-change worker must not finalize a + // non-NULL pipeline after dispose(). Without the deferred teardown this loop + // leaks several MB per iteration; with it, RSS stays essentially flat. + const iterations = 100; + + const build = () => new Pipeline("videotestsrc ! videoconvert ! queue ! fakesink"); + + // Warm up so first-touch allocations don't skew the baseline. + for (let i = 0; i < 10; i++) { + const p = build(); + const playing = p.play(); + p.dispose(); + await playing; + } + if (typeof globalThis.gc === "function") globalThis.gc(); + const before = process.memoryUsage().rss; + + for (let i = 0; i < iterations; i++) { + const p = build(); + const playing = p.play(); // in flight + p.dispose(); // deferred teardown + await playing; // teardown runs here + } + + if (typeof globalThis.gc === "function") globalThis.gc(); + const after = process.memoryUsage().rss; + + const growthMb = (after - before) / 1024 / 1024; + // A per-iteration leak of the ~3 MB pipeline would be hundreds of MB over + // 100 iterations. Allow generous headroom for allocator/GC noise. + expect(growthMb).toBeLessThan(50); + }); + it("should be idempotent — a second dispose() is a no-op", async () => { const pipeline = new Pipeline("videotestsrc ! fakesink"); diff --git a/src/ts/pipeline-eos.test.ts b/src/ts/pipeline-eos.test.ts index e6d6d8b..cdc1a47 100644 --- a/src/ts/pipeline-eos.test.ts +++ b/src/ts/pipeline-eos.test.ts @@ -12,8 +12,9 @@ describe("Pipeline EOS - State-Gated Dispatch", () => { expect(result).toBe(true); // Let EOS propagate to the bus before stopping so the source loop unwinds - // cleanly rather than racing the state teardown. - await waitForEos(pipeline); + // cleanly rather than racing the state teardown. EOS is expected here, so + // assert it actually arrived instead of silently burning the timeout. + expect(await waitForEos(pipeline)).toBe(true); await pipeline.stop(); }); @@ -83,7 +84,7 @@ describe("Pipeline EOS - State-Gated Dispatch", () => { expect(result2).toBe(true); expect(result3).toBe(true); - await waitForEos(pipeline); + expect(await waitForEos(pipeline)).toBe(true); await pipeline.stop(); }); @@ -96,7 +97,7 @@ describe("Pipeline EOS - State-Gated Dispatch", () => { const resultWhilePlaying = pipeline.endOfStream(); expect(resultWhilePlaying).toBe(true); - await waitForEos(pipeline); + expect(await waitForEos(pipeline)).toBe(true); await pipeline.stop(); From ce020c6d4fa0c1af472bbb7e32c6d05a2896a78c Mon Sep 17 00:00:00 2001 From: Jeff Pai Date: Fri, 4 Sep 2026 16:40:41 -0700 Subject: [PATCH 7/7] chore: revert unrelated EOS test-stabilization changes Drop the waitForEos helper and the EOS test changes that used it, keeping this branch scoped to the dispose() feature. The Windows basesrc flake fix belongs in its own branch/PR. --- src/cpp/pipeline.cpp | 90 +++++++++++++++++++++++++++++-------- src/cpp/pipeline.hpp | 2 - src/ts/appsrc-eos.test.ts | 25 +++++++---- src/ts/pipeline-eos.test.ts | 10 ++--- src/ts/test-utils.ts | 24 ---------- 5 files changed, 90 insertions(+), 61 deletions(-) diff --git a/src/cpp/pipeline.cpp b/src/cpp/pipeline.cpp index d78159b..08e9bde 100644 --- a/src/cpp/pipeline.cpp +++ b/src/cpp/pipeline.cpp @@ -177,7 +177,9 @@ void Pipeline::state_worker_finished() { if (in_flight_state_changes == 0) Unref(); } -GstClockTime Pipeline::parse_timeout(const Napi::CallbackInfo &info) { +Napi::Value Pipeline::play(const Napi::CallbackInfo &info) { + Napi::Env env = info.Env(); + // Default timeout is 1000ms (1 second) GstClockTime timeout = 1000 * GST_MSECOND; @@ -192,19 +194,40 @@ GstClockTime Pipeline::parse_timeout(const Napi::CallbackInfo &info) { } } - return timeout; + GstPipeline *raw = require_pipeline(env); + if (raw == nullptr) return env.Undefined(); + + // Create worker and get its promise + StateChangeWorker *worker = new StateChangeWorker(env, this, raw, GST_STATE_PLAYING, timeout); + Napi::Promise promise = worker->GetPromise().Promise(); + state_worker_started(); + worker->Queue(); + + return promise; } -Napi::Value Pipeline::queue_state_change(const Napi::CallbackInfo &info, GstState target_state) { +Napi::Value Pipeline::pause(const Napi::CallbackInfo &info) { Napi::Env env = info.Env(); - GstClockTime timeout = parse_timeout(info); + // Default timeout is 1000ms (1 second) + GstClockTime timeout = 1000 * GST_MSECOND; + + // Check if timeout parameter is provided + if (info.Length() > 0 && info[0].IsNumber()) { + double timeout_ms = info[0].As().DoubleValue(); + if (timeout_ms < 0) { + // Negative timeout means infinite wait + timeout = GST_CLOCK_TIME_NONE; + } else { + timeout = static_cast(timeout_ms * GST_MSECOND); + } + } GstPipeline *raw = require_pipeline(env); if (raw == nullptr) return env.Undefined(); // Create worker and get its promise - StateChangeWorker *worker = new StateChangeWorker(env, this, raw, target_state, timeout); + StateChangeWorker *worker = new StateChangeWorker(env, this, raw, GST_STATE_PAUSED, timeout); Napi::Promise promise = worker->GetPromise().Promise(); state_worker_started(); worker->Queue(); @@ -212,16 +235,33 @@ Napi::Value Pipeline::queue_state_change(const Napi::CallbackInfo &info, GstStat return promise; } -Napi::Value Pipeline::play(const Napi::CallbackInfo &info) { - return queue_state_change(info, GST_STATE_PLAYING); -} +Napi::Value Pipeline::stop(const Napi::CallbackInfo &info) { + Napi::Env env = info.Env(); -Napi::Value Pipeline::pause(const Napi::CallbackInfo &info) { - return queue_state_change(info, GST_STATE_PAUSED); -} + // Default timeout is 1000ms (1 second) + GstClockTime timeout = 1000 * GST_MSECOND; -Napi::Value Pipeline::stop(const Napi::CallbackInfo &info) { - return queue_state_change(info, GST_STATE_NULL); + // Check if timeout parameter is provided + if (info.Length() > 0 && info[0].IsNumber()) { + double timeout_ms = info[0].As().DoubleValue(); + if (timeout_ms < 0) { + // Negative timeout means infinite wait + timeout = GST_CLOCK_TIME_NONE; + } else { + timeout = static_cast(timeout_ms * GST_MSECOND); + } + } + + GstPipeline *raw = require_pipeline(env); + if (raw == nullptr) return env.Undefined(); + + // Create worker and get its promise + StateChangeWorker *worker = new StateChangeWorker(env, this, raw, GST_STATE_NULL, timeout); + Napi::Promise promise = worker->GetPromise().Promise(); + state_worker_started(); + worker->Queue(); + + return promise; } Napi::Value Pipeline::get_element_by_name(const Napi::CallbackInfo &info) { @@ -232,10 +272,10 @@ Napi::Value Pipeline::get_element_by_name(const Napi::CallbackInfo &info) { auto name = info[0].As().Utf8Value(); GstElement *e = gst_bin_get_by_name(GST_BIN(raw), name.c_str()); - if (e == nullptr) return env.Null(); + if (e == nullptr) return info.Env().Null(); // Use the stored constructors to create the appropriate element - return Element::CreateFromGstElement(env, e); + return Element::CreateFromGstElement(info.Env(), e); } Napi::Value Pipeline::playing(const Napi::CallbackInfo &info) { @@ -252,7 +292,7 @@ Napi::Value Pipeline::playing(const Napi::CallbackInfo &info) { bool is_playing = (state == GST_STATE_PLAYING) || (ret == GST_STATE_CHANGE_ASYNC && pending == GST_STATE_PLAYING); - return Napi::Boolean::New(env, is_playing); + return Napi::Boolean::New(info.Env(), is_playing); } Napi::Value Pipeline::query_position(const Napi::CallbackInfo &info) { @@ -263,7 +303,7 @@ Napi::Value Pipeline::query_position(const Napi::CallbackInfo &info) { gint64 pos; gst_element_query_position(GST_ELEMENT(raw), GST_FORMAT_TIME, &pos); double r = pos == -1 ? -1 : (double)pos / GST_SECOND; - return Napi::Number::New(env, r); + return Napi::Number::New(info.Env(), r); } Napi::Value Pipeline::query_duration(const Napi::CallbackInfo &info) { @@ -274,13 +314,25 @@ Napi::Value Pipeline::query_duration(const Napi::CallbackInfo &info) { gint64 dur; gst_element_query_duration(GST_ELEMENT(raw), GST_FORMAT_TIME, &dur); double r = dur == -1 ? -1 : (double)dur / GST_SECOND; - return Napi::Number::New(env, r); + return Napi::Number::New(info.Env(), r); } Napi::Value Pipeline::bus_pop(const Napi::CallbackInfo &info) { Napi::Env env = info.Env(); - GstClockTime timeout = parse_timeout(info); + // Default timeout is 1000ms (1 second) - converted to nanoseconds + GstClockTime timeout = 1000 * GST_MSECOND; + + // Check if timeout parameter is provided + if (info.Length() > 0 && info[0].IsNumber()) { + double timeout_ms = info[0].As().DoubleValue(); + if (timeout_ms < 0) { + // Negative timeout means infinite wait + timeout = GST_CLOCK_TIME_NONE; + } else { + timeout = static_cast(timeout_ms * GST_MSECOND); + } + } GstPipeline *raw = require_pipeline(env); if (raw == nullptr) return env.Undefined(); diff --git a/src/cpp/pipeline.hpp b/src/cpp/pipeline.hpp index 7c5fb93..d98fa04 100644 --- a/src/cpp/pipeline.hpp +++ b/src/cpp/pipeline.hpp @@ -35,8 +35,6 @@ class Pipeline : public Napi::ObjectWrap { int in_flight_state_changes; static bool gst_initialized; static void ensure_gst_initialized(); - static GstClockTime parse_timeout(const Napi::CallbackInfo &info); GstPipeline *require_pipeline(const Napi::Env &env); - Napi::Value queue_state_change(const Napi::CallbackInfo &info, GstState target_state); void release_native_pipeline(); }; diff --git a/src/ts/appsrc-eos.test.ts b/src/ts/appsrc-eos.test.ts index c55d0d8..b29547a 100644 --- a/src/ts/appsrc-eos.test.ts +++ b/src/ts/appsrc-eos.test.ts @@ -1,6 +1,5 @@ import { describe, expect, it } from "vitest"; -import { Pipeline } from "."; -import { waitForEos } from "./test-utils"; +import { Pipeline, type GstMessage } from "."; describe("AppSrc End-of-Stream", () => { it("should send EOS signal through endOfStream method", async () => { @@ -29,9 +28,21 @@ describe("AppSrc End-of-Stream", () => { // Send end-of-stream source.endOfStream(); - // Wait for EOS to reach the bus before stopping, so the source loop - // unwinds cleanly instead of racing the state teardown. - const eosReceived = await waitForEos(pipeline, { attempts: 20, timeoutMs: 1000 }); + // Wait for EOS message on the bus + let eosReceived = false; + let attempts = 0; + const maxAttempts = 20; // Increased attempts + + while (!eosReceived && attempts < maxAttempts) { + const message: GstMessage | null = await pipeline.busPop(1000); // Increased timeout + attempts++; + + if (message?.type === "eos") { + eosReceived = true; + expect(message.type).toBe("eos"); + break; + } + } await pipeline.stop(); @@ -81,10 +92,6 @@ describe("AppSrc End-of-Stream", () => { expect(error).toBeDefined(); } - // Drain to EOS before stopping so the source streaming thread has finished - // its loop; stopping mid-loop races GStreamer's internal EOS handling. - await waitForEos(pipeline, { attempts: 20, timeoutMs: 1000 }); - await pipeline.stop(); } }); diff --git a/src/ts/pipeline-eos.test.ts b/src/ts/pipeline-eos.test.ts index cdc1a47..a344f4a 100644 --- a/src/ts/pipeline-eos.test.ts +++ b/src/ts/pipeline-eos.test.ts @@ -1,6 +1,5 @@ import { describe, expect, it } from "vitest"; import { Pipeline, type GstMessage } from "."; -import { waitForEos } from "./test-utils"; describe("Pipeline EOS - State-Gated Dispatch", () => { it("should return true when pipeline is in PLAYING state", async () => { @@ -11,10 +10,7 @@ describe("Pipeline EOS - State-Gated Dispatch", () => { const result = pipeline.endOfStream(); expect(result).toBe(true); - // Let EOS propagate to the bus before stopping so the source loop unwinds - // cleanly rather than racing the state teardown. EOS is expected here, so - // assert it actually arrived instead of silently burning the timeout. - expect(await waitForEos(pipeline)).toBe(true); + await new Promise(resolve => setTimeout(resolve, 30)); await pipeline.stop(); }); @@ -84,7 +80,7 @@ describe("Pipeline EOS - State-Gated Dispatch", () => { expect(result2).toBe(true); expect(result3).toBe(true); - expect(await waitForEos(pipeline)).toBe(true); + await new Promise(resolve => setTimeout(resolve, 30)); await pipeline.stop(); }); @@ -97,7 +93,7 @@ describe("Pipeline EOS - State-Gated Dispatch", () => { const resultWhilePlaying = pipeline.endOfStream(); expect(resultWhilePlaying).toBe(true); - expect(await waitForEos(pipeline)).toBe(true); + await new Promise(resolve => setTimeout(resolve, 30)); await pipeline.stop(); diff --git a/src/ts/test-utils.ts b/src/ts/test-utils.ts index b051862..2939349 100644 --- a/src/ts/test-utils.ts +++ b/src/ts/test-utils.ts @@ -20,27 +20,3 @@ export const arePluginsAvailable = (plugins: string[]) => plugins.every(plugin => isPluginAvailable(plugin)); export const isWindows = process.platform === "win32"; - -/** - * Drain the pipeline bus until an EOS (or error) message is seen, or the budget - * is exhausted. - * - * After sending EOS to a source, the source's streaming thread finishes its - * current loop and posts EOS on the bus. Calling stop() before that settles can - * race GStreamer's internal basesrc EOS handling (observed on Windows as a - * `has_pending_eos` assertion abort). Waiting for EOS to reach the bus lets the - * source loop unwind cleanly before the state teardown. - * - * Returns true if EOS was observed, false otherwise. - */ -export const waitForEos = async ( - pipeline: Pipeline, - { attempts = 20, timeoutMs = 500 }: { attempts?: number; timeoutMs?: number } = {} -): Promise => { - for (let i = 0; i < attempts; i++) { - const message = await pipeline.busPop(timeoutMs); - if (message?.type === "eos") return true; - if (message?.type === "error") return false; - } - return false; -};