stream: avoid leaking consumers on signal failure

Validate Broadcast.push() and Share.pull() signals before registering
raw consumers. Return a rejecting iterable for pre-aborted signals
without adding a cursor.
This prevents failed subscriptions from leaving unreachable cursors
that inflate consumerCount and can permanently impose backpressure.
Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: codex:gpt-5.6-sol
PR-URL: #65299Fixes: #65298
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Jason Zhang <xzha4350@gmail.com>

Author: trivikr

lib/internal/streams/iter/broadcast.js

@@ -116,17 +116,32 @@ class BroadcastImpl {
 
 push(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
 // When transforms are present, delegate to pull() which creates its
 // own internal AbortController that follows the external signal.
 // When no transforms, return rawConsumer directly (controller elided
 // per PULL-02 optimization -- no transforms means no signal recipient).
-if(transforms.length>0||options?.signal){
+if(transforms.length>0||signal){
 constpullArgs=[...transforms];
-if(options?.signal){
+if(signal){
 ArrayPrototypePush(pullArgs,
-{__proto__: null,signal: options.signal});
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...pullArgs);
 }

lib/internal/streams/iter/share.js

@@ -93,11 +93,29 @@ class ShareImpl {
 
 pull(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
-if(transforms.length>0||options?.signal){
-if(options){
-returnpullWithTransforms(rawConsumer, ...transforms,options);
+if(transforms.length>0||signal){
+if(signal){
+returnpullWithTransforms(
+rawConsumer,
+ ...transforms,
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...transforms);
 }

test/parallel/test-stream-iter-broadcast-basic.js

@@ -203,6 +203,17 @@ async function testPushAbortSignalRejectsPendingNext() {
 awaitrejected;
 }
 
+asyncfunctiontestPushPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+const{broadcast: bc}=broadcast();
+constiter=bc.push({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(bc.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // =============================================================================
 // Writer fail detaches consumers
 // =============================================================================
@@ -331,6 +342,7 @@ Promise.all([
 testCancelWithFalsyReason(),
 testPendingNextSettlesAfterReturn(),
 testPushAbortSignalRejectsPendingNext(),
+testPushPreAbortedSignalDoesNotAddConsumer(),
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),

test/parallel/test-stream-iter-share-async.js

@@ -226,6 +226,17 @@ async function testSharePullAbortSignalRejectsPendingNext() {
 shared.cancel();
 }
 
+asyncfunctiontestSharePullPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+constshared=share(from('data'));
+constiter=shared.pull({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(shared.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(shared.consumerCount,0);
+}
+
 asyncfunctiontestShareAlreadyAborted(){
 constshared=share(from('data'),{signal: AbortSignal.abort()});
 constconsumer=shared.pull();
@@ -372,6 +383,7 @@ Promise.all([
 testShareAbortSignal(),
 testShareAbortSignalWhileSourcePullPending(),
 testSharePullAbortSignalRejectsPendingNext(),
+testSharePullPreAbortedSignalDoesNotAddConsumer(),
 testShareAlreadyAborted(),
 testShareSourceError(),
 testShareLateJoiningConsumer(),

test/parallel/test-stream-iter-validation.js

@@ -156,6 +156,15 @@ assert.throws(() => broadcast({ budget: 16383 }), { code: 'ERR_OUT_OF_RANGE' });
 assert.throws(()=>broadcast({signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>broadcast({backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Broadcast consumer options.signal must be AbortSignal and validation must
+// not leave a consumer registered.
+{
+const{broadcast: bc}=broadcast();
+assert.throws(()=>bc.push({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // BroadcastWriter options.signal must be AbortSignal
 {
 const{ writer }=broadcast();
@@ -212,6 +221,15 @@ assert.throws(() => share(from('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }),
 assert.throws(()=>share(from('a'),{signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>share(from('a'),{backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Share consumer options.signal must be AbortSignal and validation must not
+// leave a consumer registered.
+{
+constshared=share(from('a'));
+assert.throws(()=>shared.pull({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(shared.consumerCount,0);
+}
+
 // share() values < 16384 are rejected
 assert.throws(()=>share(from('a'),{budget: 0}),{code: 'ERR_OUT_OF_RANGE'});
 assert.throws(()=>share(from('a'),{budget: -1}),{code: 'ERR_OUT_OF_RANGE'});

lib/internal/streams/iter/broadcast.js

@@ -116,17 +116,32 @@ class BroadcastImpl {
 
 push(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
 // When transforms are present, delegate to pull() which creates its
 // own internal AbortController that follows the external signal.
 // When no transforms, return rawConsumer directly (controller elided
 // per PULL-02 optimization -- no transforms means no signal recipient).
-if(transforms.length>0||options?.signal){
+if(transforms.length>0||signal){
 constpullArgs=[...transforms];
-if(options?.signal){
+if(signal){
 ArrayPrototypePush(pullArgs,
-{__proto__: null,signal: options.signal});
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...pullArgs);
 }

lib/internal/streams/iter/share.js

@@ -93,11 +93,29 @@ class ShareImpl {
 
 pull(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
-if(transforms.length>0||options?.signal){
-if(options){
-returnpullWithTransforms(rawConsumer, ...transforms,options);
+if(transforms.length>0||signal){
+if(signal){
+returnpullWithTransforms(
+rawConsumer,
+ ...transforms,
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...transforms);
 }

test/parallel/test-stream-iter-broadcast-basic.js

@@ -203,6 +203,17 @@ async function testPushAbortSignalRejectsPendingNext() {
 awaitrejected;
 }
 
+asyncfunctiontestPushPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+const{broadcast: bc}=broadcast();
+constiter=bc.push({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(bc.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // =============================================================================
 // Writer fail detaches consumers
 // =============================================================================
@@ -331,6 +342,7 @@ Promise.all([
 testCancelWithFalsyReason(),
 testPendingNextSettlesAfterReturn(),
 testPushAbortSignalRejectsPendingNext(),
+testPushPreAbortedSignalDoesNotAddConsumer(),
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),

test/parallel/test-stream-iter-share-async.js

@@ -226,6 +226,17 @@ async function testSharePullAbortSignalRejectsPendingNext() {
 shared.cancel();
 }
 
+asyncfunctiontestSharePullPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+constshared=share(from('data'));
+constiter=shared.pull({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(shared.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(shared.consumerCount,0);
+}
+
 asyncfunctiontestShareAlreadyAborted(){
 constshared=share(from('data'),{signal: AbortSignal.abort()});
 constconsumer=shared.pull();
@@ -372,6 +383,7 @@ Promise.all([
 testShareAbortSignal(),
 testShareAbortSignalWhileSourcePullPending(),
 testSharePullAbortSignalRejectsPendingNext(),
+testSharePullPreAbortedSignalDoesNotAddConsumer(),
 testShareAlreadyAborted(),
 testShareSourceError(),
 testShareLateJoiningConsumer(),

test/parallel/test-stream-iter-validation.js

@@ -156,6 +156,15 @@ assert.throws(() => broadcast({ budget: 16383 }), { code: 'ERR_OUT_OF_RANGE' });
 assert.throws(()=>broadcast({signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>broadcast({backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Broadcast consumer options.signal must be AbortSignal and validation must
+// not leave a consumer registered.
+{
+const{broadcast: bc}=broadcast();
+assert.throws(()=>bc.push({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // BroadcastWriter options.signal must be AbortSignal
 {
 const{ writer }=broadcast();
@@ -212,6 +221,15 @@ assert.throws(() => share(from('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }),
 assert.throws(()=>share(from('a'),{signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>share(from('a'),{backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Share consumer options.signal must be AbortSignal and validation must not
+// leave a consumer registered.
+{
+constshared=share(from('a'));
+assert.throws(()=>shared.pull({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(shared.consumerCount,0);
+}
+
 // share() values < 16384 are rejected
 assert.throws(()=>share(from('a'),{budget: 0}),{code: 'ERR_OUT_OF_RANGE'});
 assert.throws(()=>share(from('a'),{budget: -1}),{code: 'ERR_OUT_OF_RANGE'});

lib/internal/streams/iter/broadcast.js

@@ -116,17 +116,32 @@ class BroadcastImpl {
 
 push(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
 // When transforms are present, delegate to pull() which creates its
 // own internal AbortController that follows the external signal.
 // When no transforms, return rawConsumer directly (controller elided
 // per PULL-02 optimization -- no transforms means no signal recipient).
-if(transforms.length>0||options?.signal){
+if(transforms.length>0||signal){
 constpullArgs=[...transforms];
-if(options?.signal){
+if(signal){
 ArrayPrototypePush(pullArgs,
-{__proto__: null,signal: options.signal});
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...pullArgs);
 }

lib/internal/streams/iter/share.js

@@ -93,11 +93,29 @@ class ShareImpl {
 
 pull(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
-if(transforms.length>0||options?.signal){
-if(options){
-returnpullWithTransforms(rawConsumer, ...transforms,options);
+if(transforms.length>0||signal){
+if(signal){
+returnpullWithTransforms(
+rawConsumer,
+ ...transforms,
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...transforms);
 }

test/parallel/test-stream-iter-broadcast-basic.js

@@ -203,6 +203,17 @@ async function testPushAbortSignalRejectsPendingNext() {
 awaitrejected;
 }
 
+asyncfunctiontestPushPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+const{broadcast: bc}=broadcast();
+constiter=bc.push({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(bc.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // =============================================================================
 // Writer fail detaches consumers
 // =============================================================================
@@ -331,6 +342,7 @@ Promise.all([
 testCancelWithFalsyReason(),
 testPendingNextSettlesAfterReturn(),
 testPushAbortSignalRejectsPendingNext(),
+testPushPreAbortedSignalDoesNotAddConsumer(),
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),

test/parallel/test-stream-iter-share-async.js

@@ -226,6 +226,17 @@ async function testSharePullAbortSignalRejectsPendingNext() {
 shared.cancel();
 }
 
+asyncfunctiontestSharePullPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+constshared=share(from('data'));
+constiter=shared.pull({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(shared.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(shared.consumerCount,0);
+}
+
 asyncfunctiontestShareAlreadyAborted(){
 constshared=share(from('data'),{signal: AbortSignal.abort()});
 constconsumer=shared.pull();
@@ -372,6 +383,7 @@ Promise.all([
 testShareAbortSignal(),
 testShareAbortSignalWhileSourcePullPending(),
 testSharePullAbortSignalRejectsPendingNext(),
+testSharePullPreAbortedSignalDoesNotAddConsumer(),
 testShareAlreadyAborted(),
 testShareSourceError(),
 testShareLateJoiningConsumer(),

test/parallel/test-stream-iter-validation.js

@@ -156,6 +156,15 @@ assert.throws(() => broadcast({ budget: 16383 }), { code: 'ERR_OUT_OF_RANGE' });
 assert.throws(()=>broadcast({signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>broadcast({backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Broadcast consumer options.signal must be AbortSignal and validation must
+// not leave a consumer registered.
+{
+const{broadcast: bc}=broadcast();
+assert.throws(()=>bc.push({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // BroadcastWriter options.signal must be AbortSignal
 {
 const{ writer }=broadcast();
@@ -212,6 +221,15 @@ assert.throws(() => share(from('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }),
 assert.throws(()=>share(from('a'),{signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>share(from('a'),{backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Share consumer options.signal must be AbortSignal and validation must not
+// leave a consumer registered.
+{
+constshared=share(from('a'));
+assert.throws(()=>shared.pull({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(shared.consumerCount,0);
+}
+
 // share() values < 16384 are rejected
 assert.throws(()=>share(from('a'),{budget: 0}),{code: 'ERR_OUT_OF_RANGE'});
 assert.throws(()=>share(from('a'),{budget: -1}),{code: 'ERR_OUT_OF_RANGE'});

lib/internal/streams/iter/broadcast.js

@@ -116,17 +116,32 @@ class BroadcastImpl {
 
 push(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
 // When transforms are present, delegate to pull() which creates its
 // own internal AbortController that follows the external signal.
 // When no transforms, return rawConsumer directly (controller elided
 // per PULL-02 optimization -- no transforms means no signal recipient).
-if(transforms.length>0||options?.signal){
+if(transforms.length>0||signal){
 constpullArgs=[...transforms];
-if(options?.signal){
+if(signal){
 ArrayPrototypePush(pullArgs,
-{__proto__: null,signal: options.signal});
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...pullArgs);
 }

lib/internal/streams/iter/share.js

@@ -93,11 +93,29 @@ class ShareImpl {
 
 pull(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
-if(transforms.length>0||options?.signal){
-if(options){
-returnpullWithTransforms(rawConsumer, ...transforms,options);
+if(transforms.length>0||signal){
+if(signal){
+returnpullWithTransforms(
+rawConsumer,
+ ...transforms,
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...transforms);
 }

test/parallel/test-stream-iter-broadcast-basic.js

@@ -203,6 +203,17 @@ async function testPushAbortSignalRejectsPendingNext() {
 awaitrejected;
 }
 
+asyncfunctiontestPushPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+const{broadcast: bc}=broadcast();
+constiter=bc.push({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(bc.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // =============================================================================
 // Writer fail detaches consumers
 // =============================================================================
@@ -331,6 +342,7 @@ Promise.all([
 testCancelWithFalsyReason(),
 testPendingNextSettlesAfterReturn(),
 testPushAbortSignalRejectsPendingNext(),
+testPushPreAbortedSignalDoesNotAddConsumer(),
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),

test/parallel/test-stream-iter-share-async.js

@@ -226,6 +226,17 @@ async function testSharePullAbortSignalRejectsPendingNext() {
 shared.cancel();
 }
 
+asyncfunctiontestSharePullPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+constshared=share(from('data'));
+constiter=shared.pull({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(shared.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(shared.consumerCount,0);
+}
+
 asyncfunctiontestShareAlreadyAborted(){
 constshared=share(from('data'),{signal: AbortSignal.abort()});
 constconsumer=shared.pull();
@@ -372,6 +383,7 @@ Promise.all([
 testShareAbortSignal(),
 testShareAbortSignalWhileSourcePullPending(),
 testSharePullAbortSignalRejectsPendingNext(),
+testSharePullPreAbortedSignalDoesNotAddConsumer(),
 testShareAlreadyAborted(),
 testShareSourceError(),
 testShareLateJoiningConsumer(),

test/parallel/test-stream-iter-validation.js

@@ -156,6 +156,15 @@ assert.throws(() => broadcast({ budget: 16383 }), { code: 'ERR_OUT_OF_RANGE' });
 assert.throws(()=>broadcast({signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>broadcast({backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Broadcast consumer options.signal must be AbortSignal and validation must
+// not leave a consumer registered.
+{
+const{broadcast: bc}=broadcast();
+assert.throws(()=>bc.push({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // BroadcastWriter options.signal must be AbortSignal
 {
 const{ writer }=broadcast();
@@ -212,6 +221,15 @@ assert.throws(() => share(from('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }),
 assert.throws(()=>share(from('a'),{signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>share(from('a'),{backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Share consumer options.signal must be AbortSignal and validation must not
+// leave a consumer registered.
+{
+constshared=share(from('a'));
+assert.throws(()=>shared.pull({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(shared.consumerCount,0);
+}
+
 // share() values < 16384 are rejected
 assert.throws(()=>share(from('a'),{budget: 0}),{code: 'ERR_OUT_OF_RANGE'});
 assert.throws(()=>share(from('a'),{budget: -1}),{code: 'ERR_OUT_OF_RANGE'});

lib/internal/streams/iter/broadcast.js

@@ -116,17 +116,32 @@ class BroadcastImpl {
 
 push(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
 // When transforms are present, delegate to pull() which creates its
 // own internal AbortController that follows the external signal.
 // When no transforms, return rawConsumer directly (controller elided
 // per PULL-02 optimization -- no transforms means no signal recipient).
-if(transforms.length>0||options?.signal){
+if(transforms.length>0||signal){
 constpullArgs=[...transforms];
-if(options?.signal){
+if(signal){
 ArrayPrototypePush(pullArgs,
-{__proto__: null,signal: options.signal});
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...pullArgs);
 }

lib/internal/streams/iter/share.js

@@ -93,11 +93,29 @@ class ShareImpl {
 
 pull(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
-if(transforms.length>0||options?.signal){
-if(options){
-returnpullWithTransforms(rawConsumer, ...transforms,options);
+if(transforms.length>0||signal){
+if(signal){
+returnpullWithTransforms(
+rawConsumer,
+ ...transforms,
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...transforms);
 }

test/parallel/test-stream-iter-broadcast-basic.js

@@ -203,6 +203,17 @@ async function testPushAbortSignalRejectsPendingNext() {
 awaitrejected;
 }
 
+asyncfunctiontestPushPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+const{broadcast: bc}=broadcast();
+constiter=bc.push({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(bc.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // =============================================================================
 // Writer fail detaches consumers
 // =============================================================================
@@ -331,6 +342,7 @@ Promise.all([
 testCancelWithFalsyReason(),
 testPendingNextSettlesAfterReturn(),
 testPushAbortSignalRejectsPendingNext(),
+testPushPreAbortedSignalDoesNotAddConsumer(),
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),

test/parallel/test-stream-iter-share-async.js

@@ -226,6 +226,17 @@ async function testSharePullAbortSignalRejectsPendingNext() {
 shared.cancel();
 }
 
+asyncfunctiontestSharePullPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+constshared=share(from('data'));
+constiter=shared.pull({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(shared.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(shared.consumerCount,0);
+}
+
 asyncfunctiontestShareAlreadyAborted(){
 constshared=share(from('data'),{signal: AbortSignal.abort()});
 constconsumer=shared.pull();
@@ -372,6 +383,7 @@ Promise.all([
 testShareAbortSignal(),
 testShareAbortSignalWhileSourcePullPending(),
 testSharePullAbortSignalRejectsPendingNext(),
+testSharePullPreAbortedSignalDoesNotAddConsumer(),
 testShareAlreadyAborted(),
 testShareSourceError(),
 testShareLateJoiningConsumer(),

test/parallel/test-stream-iter-validation.js

@@ -156,6 +156,15 @@ assert.throws(() => broadcast({ budget: 16383 }), { code: 'ERR_OUT_OF_RANGE' });
 assert.throws(()=>broadcast({signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>broadcast({backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Broadcast consumer options.signal must be AbortSignal and validation must
+// not leave a consumer registered.
+{
+const{broadcast: bc}=broadcast();
+assert.throws(()=>bc.push({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // BroadcastWriter options.signal must be AbortSignal
 {
 const{ writer }=broadcast();
@@ -212,6 +221,15 @@ assert.throws(() => share(from('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }),
 assert.throws(()=>share(from('a'),{signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>share(from('a'),{backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Share consumer options.signal must be AbortSignal and validation must not
+// leave a consumer registered.
+{
+constshared=share(from('a'));
+assert.throws(()=>shared.pull({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(shared.consumerCount,0);
+}
+
 // share() values < 16384 are rejected
 assert.throws(()=>share(from('a'),{budget: 0}),{code: 'ERR_OUT_OF_RANGE'});
 assert.throws(()=>share(from('a'),{budget: -1}),{code: 'ERR_OUT_OF_RANGE'});

lib/internal/streams/iter/broadcast.js

@@ -116,17 +116,32 @@ class BroadcastImpl {
 
 push(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
 // When transforms are present, delegate to pull() which creates its
 // own internal AbortController that follows the external signal.
 // When no transforms, return rawConsumer directly (controller elided
 // per PULL-02 optimization -- no transforms means no signal recipient).
-if(transforms.length>0||options?.signal){
+if(transforms.length>0||signal){
 constpullArgs=[...transforms];
-if(options?.signal){
+if(signal){
 ArrayPrototypePush(pullArgs,
-{__proto__: null,signal: options.signal});
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...pullArgs);
 }

lib/internal/streams/iter/share.js

@@ -93,11 +93,29 @@ class ShareImpl {
 
 pull(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
-if(transforms.length>0||options?.signal){
-if(options){
-returnpullWithTransforms(rawConsumer, ...transforms,options);
+if(transforms.length>0||signal){
+if(signal){
+returnpullWithTransforms(
+rawConsumer,
+ ...transforms,
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...transforms);
 }

test/parallel/test-stream-iter-broadcast-basic.js

@@ -203,6 +203,17 @@ async function testPushAbortSignalRejectsPendingNext() {
 awaitrejected;
 }
 
+asyncfunctiontestPushPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+const{broadcast: bc}=broadcast();
+constiter=bc.push({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(bc.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // =============================================================================
 // Writer fail detaches consumers
 // =============================================================================
@@ -331,6 +342,7 @@ Promise.all([
 testCancelWithFalsyReason(),
 testPendingNextSettlesAfterReturn(),
 testPushAbortSignalRejectsPendingNext(),
+testPushPreAbortedSignalDoesNotAddConsumer(),
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),

test/parallel/test-stream-iter-share-async.js

@@ -226,6 +226,17 @@ async function testSharePullAbortSignalRejectsPendingNext() {
 shared.cancel();
 }
 
+asyncfunctiontestSharePullPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+constshared=share(from('data'));
+constiter=shared.pull({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(shared.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(shared.consumerCount,0);
+}
+
 asyncfunctiontestShareAlreadyAborted(){
 constshared=share(from('data'),{signal: AbortSignal.abort()});
 constconsumer=shared.pull();
@@ -372,6 +383,7 @@ Promise.all([
 testShareAbortSignal(),
 testShareAbortSignalWhileSourcePullPending(),
 testSharePullAbortSignalRejectsPendingNext(),
+testSharePullPreAbortedSignalDoesNotAddConsumer(),
 testShareAlreadyAborted(),
 testShareSourceError(),
 testShareLateJoiningConsumer(),

test/parallel/test-stream-iter-validation.js

@@ -156,6 +156,15 @@ assert.throws(() => broadcast({ budget: 16383 }), { code: 'ERR_OUT_OF_RANGE' });
 assert.throws(()=>broadcast({signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>broadcast({backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Broadcast consumer options.signal must be AbortSignal and validation must
+// not leave a consumer registered.
+{
+const{broadcast: bc}=broadcast();
+assert.throws(()=>bc.push({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // BroadcastWriter options.signal must be AbortSignal
 {
 const{ writer }=broadcast();
@@ -212,6 +221,15 @@ assert.throws(() => share(from('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }),
 assert.throws(()=>share(from('a'),{signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>share(from('a'),{backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Share consumer options.signal must be AbortSignal and validation must not
+// leave a consumer registered.
+{
+constshared=share(from('a'));
+assert.throws(()=>shared.pull({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(shared.consumerCount,0);
+}
+
 // share() values < 16384 are rejected
 assert.throws(()=>share(from('a'),{budget: 0}),{code: 'ERR_OUT_OF_RANGE'});
 assert.throws(()=>share(from('a'),{budget: -1}),{code: 'ERR_OUT_OF_RANGE'});

lib/internal/streams/iter/broadcast.js

@@ -116,17 +116,32 @@ class BroadcastImpl {
 
 push(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
 // When transforms are present, delegate to pull() which creates its
 // own internal AbortController that follows the external signal.
 // When no transforms, return rawConsumer directly (controller elided
 // per PULL-02 optimization -- no transforms means no signal recipient).
-if(transforms.length>0||options?.signal){
+if(transforms.length>0||signal){
 constpullArgs=[...transforms];
-if(options?.signal){
+if(signal){
 ArrayPrototypePush(pullArgs,
-{__proto__: null,signal: options.signal});
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...pullArgs);
 }

lib/internal/streams/iter/share.js

@@ -93,11 +93,29 @@ class ShareImpl {
 
 pull(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
-if(transforms.length>0||options?.signal){
-if(options){
-returnpullWithTransforms(rawConsumer, ...transforms,options);
+if(transforms.length>0||signal){
+if(signal){
+returnpullWithTransforms(
+rawConsumer,
+ ...transforms,
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...transforms);
 }

test/parallel/test-stream-iter-broadcast-basic.js

@@ -203,6 +203,17 @@ async function testPushAbortSignalRejectsPendingNext() {
 awaitrejected;
 }
 
+asyncfunctiontestPushPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+const{broadcast: bc}=broadcast();
+constiter=bc.push({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(bc.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // =============================================================================
 // Writer fail detaches consumers
 // =============================================================================
@@ -331,6 +342,7 @@ Promise.all([
 testCancelWithFalsyReason(),
 testPendingNextSettlesAfterReturn(),
 testPushAbortSignalRejectsPendingNext(),
+testPushPreAbortedSignalDoesNotAddConsumer(),
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),

test/parallel/test-stream-iter-share-async.js

@@ -226,6 +226,17 @@ async function testSharePullAbortSignalRejectsPendingNext() {
 shared.cancel();
 }
 
+asyncfunctiontestSharePullPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+constshared=share(from('data'));
+constiter=shared.pull({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(shared.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(shared.consumerCount,0);
+}
+
 asyncfunctiontestShareAlreadyAborted(){
 constshared=share(from('data'),{signal: AbortSignal.abort()});
 constconsumer=shared.pull();
@@ -372,6 +383,7 @@ Promise.all([
 testShareAbortSignal(),
 testShareAbortSignalWhileSourcePullPending(),
 testSharePullAbortSignalRejectsPendingNext(),
+testSharePullPreAbortedSignalDoesNotAddConsumer(),
 testShareAlreadyAborted(),
 testShareSourceError(),
 testShareLateJoiningConsumer(),

test/parallel/test-stream-iter-validation.js

@@ -156,6 +156,15 @@ assert.throws(() => broadcast({ budget: 16383 }), { code: 'ERR_OUT_OF_RANGE' });
 assert.throws(()=>broadcast({signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>broadcast({backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Broadcast consumer options.signal must be AbortSignal and validation must
+// not leave a consumer registered.
+{
+const{broadcast: bc}=broadcast();
+assert.throws(()=>bc.push({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // BroadcastWriter options.signal must be AbortSignal
 {
 const{ writer }=broadcast();
@@ -212,6 +221,15 @@ assert.throws(() => share(from('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }),
 assert.throws(()=>share(from('a'),{signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>share(from('a'),{backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Share consumer options.signal must be AbortSignal and validation must not
+// leave a consumer registered.
+{
+constshared=share(from('a'));
+assert.throws(()=>shared.pull({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(shared.consumerCount,0);
+}
+
 // share() values < 16384 are rejected
 assert.throws(()=>share(from('a'),{budget: 0}),{code: 'ERR_OUT_OF_RANGE'});
 assert.throws(()=>share(from('a'),{budget: -1}),{code: 'ERR_OUT_OF_RANGE'});

lib/internal/streams/iter/broadcast.js

@@ -116,17 +116,32 @@ class BroadcastImpl {
 
 push(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
 // When transforms are present, delegate to pull() which creates its
 // own internal AbortController that follows the external signal.
 // When no transforms, return rawConsumer directly (controller elided
 // per PULL-02 optimization -- no transforms means no signal recipient).
-if(transforms.length>0||options?.signal){
+if(transforms.length>0||signal){
 constpullArgs=[...transforms];
-if(options?.signal){
+if(signal){
 ArrayPrototypePush(pullArgs,
-{__proto__: null,signal: options.signal});
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...pullArgs);
 }

lib/internal/streams/iter/share.js

@@ -93,11 +93,29 @@ class ShareImpl {
 
 pull(...args){
 const{ transforms, options }=parsePullArgs(args);
+constsignal=options?.signal;
+validateAbortSignal(signal,'options.signal');
+
+// Avoid registering a consumer that the pre-aborted pipeline will never
+// read or detach.
+if(signal?.aborted){
+return{
+__proto__: null,
+// eslint-disable-next-line require-yield
+async*[SymbolAsyncIterator](){
+throwsignal.reason;
+},
+};
+}
+
 constrawConsumer=this.#createRawConsumer();
 
-if(transforms.length>0||options?.signal){
-if(options){
-returnpullWithTransforms(rawConsumer, ...transforms,options);
+if(transforms.length>0||signal){
+if(signal){
+returnpullWithTransforms(
+rawConsumer,
+ ...transforms,
+{__proto__: null, signal });
 }
 returnpullWithTransforms(rawConsumer, ...transforms);
 }

test/parallel/test-stream-iter-broadcast-basic.js

@@ -203,6 +203,17 @@ async function testPushAbortSignalRejectsPendingNext() {
 awaitrejected;
 }
 
+asyncfunctiontestPushPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+const{broadcast: bc}=broadcast();
+constiter=bc.push({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(bc.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // =============================================================================
 // Writer fail detaches consumers
 // =============================================================================
@@ -331,6 +342,7 @@ Promise.all([
 testCancelWithFalsyReason(),
 testPendingNextSettlesAfterReturn(),
 testPushAbortSignalRejectsPendingNext(),
+testPushPreAbortedSignalDoesNotAddConsumer(),
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),

test/parallel/test-stream-iter-share-async.js

@@ -226,6 +226,17 @@ async function testSharePullAbortSignalRejectsPendingNext() {
 shared.cancel();
 }
 
+asyncfunctiontestSharePullPreAbortedSignalDoesNotAddConsumer(){
+constreason=newError('already aborted');
+constsignal=AbortSignal.abort(reason);
+constshared=share(from('data'));
+constiter=shared.pull({ signal })[Symbol.asyncIterator]();
+
+assert.strictEqual(shared.consumerCount,0);
+awaitassert.rejects(iter.next(),(error)=>error===reason);
+assert.strictEqual(shared.consumerCount,0);
+}
+
 asyncfunctiontestShareAlreadyAborted(){
 constshared=share(from('data'),{signal: AbortSignal.abort()});
 constconsumer=shared.pull();
@@ -372,6 +383,7 @@ Promise.all([
 testShareAbortSignal(),
 testShareAbortSignalWhileSourcePullPending(),
 testSharePullAbortSignalRejectsPendingNext(),
+testSharePullPreAbortedSignalDoesNotAddConsumer(),
 testShareAlreadyAborted(),
 testShareSourceError(),
 testShareLateJoiningConsumer(),

test/parallel/test-stream-iter-validation.js

@@ -156,6 +156,15 @@ assert.throws(() => broadcast({ budget: 16383 }), { code: 'ERR_OUT_OF_RANGE' });
 assert.throws(()=>broadcast({signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>broadcast({backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Broadcast consumer options.signal must be AbortSignal and validation must
+// not leave a consumer registered.
+{
+const{broadcast: bc}=broadcast();
+assert.throws(()=>bc.push({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 // BroadcastWriter options.signal must be AbortSignal
 {
 const{ writer }=broadcast();
@@ -212,6 +221,15 @@ assert.throws(() => share(from('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }),
 assert.throws(()=>share(from('a'),{signal: {}}),{code: 'ERR_INVALID_ARG_TYPE'});
 assert.throws(()=>share(from('a'),{backpressure: 'bad'}),{code: 'ERR_INVALID_ARG_VALUE'});
 
+// Share consumer options.signal must be AbortSignal and validation must not
+// leave a consumer registered.
+{
+constshared=share(from('a'));
+assert.throws(()=>shared.pull({signal: {}}),
+{code: 'ERR_INVALID_ARG_TYPE'});
+assert.strictEqual(shared.consumerCount,0);
+}
+
 // share() values < 16384 are rejected
 assert.throws(()=>share(from('a'),{budget: 0}),{code: 'ERR_OUT_OF_RANGE'});
 assert.throws(()=>share(from('a'),{budget: -1}),{code: 'ERR_OUT_OF_RANGE'});