bug: using structuredClone with ReadableStream prevents process from exiting

Version

v18.10.0

Platform

Microsoft Windows NT 10.0.19043.0 x64

Subsystem

No response

What steps will reproduce the bug?

const rs = new ReadableStream({
  start (controller) {
    controller.enqueue(new Uint8Array([65]))
    controller.close()
  }
})

const cloned = structuredClone(rs, { transfer: [rs] })

After this the process will indefinitely stay open.

How often does it reproduce? Is there a required condition?

always

What is the expected behavior?

the process should exit

What do you see instead?

the process stays open indefinitely

Additional information

Exporting readableStreamTee would also solve my use case.

node/lib/internal/webstreams/readablestream.js

Line 1443 in 214354f

function readableStreamTee(stream, cloneForBranch2) {

A workaround is to tee the cloned stream.

@jasnell

@jasnell @addaleax

Ref: nodejs/undici#1700

It looks like 2 ports aren't unref'd

[
  MessagePort [EventTarget] {
    active: true,
    refed: true,
    [Symbol(kEvents)]: SafeMap(4) [Map] {        
      'newListener' => [Object],
      'removeListener' => [Object],
      'message' => [Object],
      'messageerror' => [Object]
    },
    [Symbol(events.maxEventTargetListeners)]: 10,
    [Symbol(events.maxEventTargetListenersWarned)]: false,
    [Symbol(kNewListener)]: [Function (anonymous)],
    [Symbol(kRemoveListener)]: [Function (anonymous)],
    [Symbol(nodejs.internal.kCurrentlyReceivingPorts)]: undefined,
    [Symbol(khandlers)]: SafeMap(2) [Map] {
      'message' => [Function],
      'messageerror' => [Function]
    }
  },
  MessagePort [EventTarget] {
    active: true,
    refed: true,
    [Symbol(kEvents)]: SafeMap(4) [Map] {
      'newListener' => [Object],
      'removeListener' => [Object],
      'message' => [Object],
      'messageerror' => [Object]
    },
    [Symbol(events.maxEventTargetListeners)]: 10,
    [Symbol(events.maxEventTargetListenersWarned)]: false,
    [Symbol(kNewListener)]: [Function (anonymous)],
    [Symbol(kRemoveListener)]: [Function (anonymous)],
    [Symbol(nodejs.internal.kCurrentlyReceivingPorts)]: undefined,
    [Symbol(khandlers)]: SafeMap(2) [Map] {
      'message' => [Function],
      'messageerror' => [Function]
    }
  }
]

which from my best guess are these:

node/lib/internal/webstreams/readablestream.js

Lines 579 to 580 in 2672219

    
           this[kState].transfer.port1 = port1; 
        
           this[kState].transfer.port2 = port2;

@joyeecheung

@joyeecheung

@ronag

@ronag
I have been able to solve this problem.

const { kTransfer } = require("internal/worker/js_transferable");

const { readableStreamPipeTo } = require("internal/webstreams/readablestream");

const { setPromiseHandled, kState } = require("internal/webstreams/util");

const {
  CrossRealmTransformWritableSink,
} = require("internal/webstreams/transfer");

function newCrossRealmWritableSink(readable, port) {
  const source = new CrossRealmTransformWritableSink(port);
  const writable = new WritableStream(source);
  const promise = readableStreamPipeTo(readable, writable, false, false, false);
  setPromiseHandled(promise);
  return {
    writable,
    source,
    promise,
  };
}

const registry = new FinalizationRegistry(({ source }) => {
  source.close();
});

ReadableStream.prototype[kTransfer] = function () {
  if (this.locked) {
    this[kState].transfer.port1?.close();
    this[kState].transfer.port1 = undefined;
    this[kState].transfer.port2 = undefined;
    throw new DOMException(
      "Cannot transfer a locked ReadableStream",
      "DataCloneError"
    );
  }

  const { port1, port2 } = this[kState].transfer;
  this[kState].transfer.port2 = undefined;

  const { writable, promise, source } = newCrossRealmWritableSink(this, port1);

  this[kState].transfer.writable = writable;
  this[kState].transfer.promise = promise;

  registry.register(port2, { source });

  return {
    data: { port: port2 },
    deserializeInfo:
      "internal/webstreams/readablestream:TransferredReadableStream",
  };
};

const stack = [
  new Uint8Array([65]),
  new Uint8Array([65]),
  new Uint8Array([65]),
];

const rs = new ReadableStream({
  pull(controller) {
    const data = stack.shift();
    if (data) {
      controller.enqueue(data);
    } else {
      controller.close();
    }
  },
});

const cloned = structuredClone(rs, { transfer: [rs] });

@ronag

@ronag
This problem occurs not only with ReadableStream but also with WritableStream.
I am stuck on this point.
Could you please work on this instead of me?

Sorry. Webstreams are beyond my expertise and frankly my interest...

I apologize for the inconvenience that I have caused you.

No worries. No inconvenience at all. 🤗

Workaround nodejs/node#44985

stream: fix cloned webstreams not being unref'd

Reopening after #51491

Revert: Workaround nodejs/node#44985

refactor: Use fetch for image runner again

Navigation Menu

Repository navigation

bug: using structuredClone with ReadableStream prevents process from exiting #44985

Version

Platform

Subsystem

What steps will reproduce the bug?

How often does it reproduce? Is there a required condition?

What is the expected behavior?

What do you see instead?

Additional information

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	this[kState].transfer.port1 = port1;
	this[kState].transfer.port2 = port2;

bug: using structuredClone with ReadableStream prevents process from exiting #44985

Description

Version

Platform

Subsystem

What steps will reproduce the bug?

How often does it reproduce? Is there a required condition?

What is the expected behavior?

What do you see instead?

Additional information

Activity

KhafraDev commented on Oct 13, 2022

ronag commented on Oct 14, 2022

ronag commented on Oct 14, 2022

KhafraDev commented on Oct 18, 2022

ronag commented on Dec 10, 2023

tsctx commented on Dec 10, 2023

tsctx commented on Dec 10, 2023

ronag commented on Dec 10, 2023

tsctx commented on Dec 10, 2023

ronag commented on Dec 10, 2023

mcollina commented on Jan 19, 2024

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions