The cloned writable problem

[ricea]

Principle: cloning a stream should be a no-op (except that queue size increases by one chunk).

Cloning formulas

"Cloning" is an operation which locks the original stream and constructs a new stream that behaves just like it. It is a useful building block for stream manipulations and it is also used in the construction of transferable streams.

Cloning a readable

const rs2 = rs1.pipeThrough(new TransformStream());

Cloning a writable

const internalTS = new TransformStream();
internalTS.readable.pipeTo(ws1);
ws2 = internalTS.writable;

Problem

The cloned readable stream rs2 preserves the properties of the original stream rs1.

However, the cloned writable stream ws2 loses information.

If you do readable.pipeTo(ws1) then the promise returned by pipeTo() will not resolve until underlyingSink.close() has resolved. However, if you do readable.pipeTo(ws2), pipeTo() will resolve as soon as readable is closed.

This problem was identified while working on the tests for transferable streams.

Cause

An underlying source is not notified when downstream has finished consuming the data. Once the ReadableStream is closed, it gets no further notification about the state of the pipe. pipeTo() waits for all writes to complete by design, but this is independent of the readable stream's main state machine.

Proposed fix

ReadableStream's underlyingSource will gain an additional method, finally(), which is called once the stream is completely quiescent, ie. after it is CLOSED or ERRORED and no operations are pending. See #636 for background and other use cases for the finally() method. There will be a new abstract operation, ReadableStreamMakeFinallyWaitFor(promise) which will delay the call to finally() until the passed-in promise is settled. This new operation could in principle be exposed on the ReadableStreamDefaultReader object, but isn't because of its limited utility. Instead, we keep it as an internal abstract operation for now, only used by the pipeTo implementation.

The finally() method will be passed a TBD argument indicating whether the ReadableStream closed without error. If the promise passed to ReadableStreamMakeFinallyWaitFor is rejected, then this argument will indicate an error occurred.

CreateReadableStream() will gain an extra operation argument corresponding to the finally() method. This will be used by the TransformStream implementation to control the timing of changing the state of the writable side to CLOSED or ERRORED, which in turn will change the timing of when a pipe to the writable side completes.

The new effect is that a pipe to ws2 will close no sooner than close() or abort() on ws1's underylingSink complete.

[domenic]

+1 to all this. I'd love to hear from community members what they think, about both the problem space and proposed solution.

Regarding

This new operation could in principle be exposed on the ReadableStreamDefaultReader object, but isn't because of its limited utility. Instead, we keep it as an internal abstract operation for now, only used by the pipeTo implementation.

it would also be interesting to hear if folks can think of a use case for exposing this. I am a bit uneasy with making pipeTo depend on functionality that is not also exposed as public APIs, but it's hard to figure out what else would use this, so I agree with the OP that it's better to not expose it for now. Maybe some of the stuff in #329 or #324 is related, although this issue focuses on the close signal and not the individual chunk signals.