Skip to content

Addresses multiple spec bugs - #53

Open
jasnell wants to merge 22 commits into
mainfrom
spec-bugs
Open

jasnell wants to merge 22 commits into
mainfrom
spec-bugs

Conversation

@jasnell

@jasnell jasnell commented Oct 6, 2026

Copy link
Copy Markdown
Collaborator

See each individual commit for details

jasnell added 21 commits October 6, 2026 14:33
SyncWriter.endSync() returned unsigned long long and threw a TypeError
when the writer was closed or errored, while Writer.endSync() returns
long long, returns -1 when it cannot end synchronously, and never
throws. Since one object may implement both interfaces, align
SyncWriter.endSync() with Writer.endSync().

Fixes #33
canWrite, write(), writev(), writeSync() and writevSync() only handled
the open, closed and errored states. While closing, canWrite is now
null, write()/writev() reject with a TypeError and writeSync()/
writevSync() return false.

endSync() now returns -1 while closing instead of the byte total, since
the writer is not closed until the drain finishes, and end() returns the
pending promise from the first call instead of resolving early.

Fixes #34
Writes can still be in the pending writes queue when end() is called
(for example under "unbounded"), but the closing branch of fail() did
not settle them. Reject them with the fail reason, as the open branch
does.

Fixes #35
end() takes a WriteOptions argument but the algorithm never used its
signal. An already-aborted signal now rejects the call without changing
the writer's state, and a signal that aborts while the drain is pending
rejects only the promise returned by that call, as signal does for
write(). The writer stays closing and the drain continues.

To support per-call signals, the promise that settles when the drain
finishes is now named the writer's close promise; end() returns it
directly when no signal is given.

Fixes #36
The 16384-byte floor applies only to the implementation-defined default,
but the note in the options section read as if it applied to any value.
Reword it, state that an explicit budget must be at least 1 (0 throws a
RangeError), and add [EnforceRange] to every budget member so that, for
example, -1 is rejected instead of being converted to 2^64-1.

Fixes #37
Symbol.asyncIterator is handled by step 7, not step 6.

Fixes #38
Only the trailing run of stateless transforms received the null flush
signal; a run that preceded a stateful transform was never flushed,
contrary to the prose and to the tap() contract. Factor the run logic
into an "apply a stateless run" operation used by both the async and
sync composers. It flushes each function once, in order, after all of
its input, including output that earlier functions emitted while
flushing.

Also make "normalize transform output" cover promises and async
iterables (async pipelines only), which the prose already allowed.

Fixes #39
The parse variadic transform arguments algorithm accepted an options
dictionary anywhere among the arguments, so pull(src, { signal }, fn)
was valid even though the note says options always come last. Only
check the final argument; any other non-transform argument is a
TypeError.

Fixes #40
text() and textSync() checked encoding only after collecting every
byte, so an unsupported encoding with an endless source never settled
and with a finite source read all of the data for nothing. Validate
the encoding first.

Fixes #41
The broadcast backpressure prose says late joiners begin reading from
the oldest entry still in the buffer, but Broadcast.push() created the
cursor at the current write position, skipping buffered data.
Share.pull() had the same mismatch. Create the cursor at the oldest
buffered entry in both, and state the share behavior explicitly.

Fixes #42
The note claimed that a rejected pull leaves the iterator usable and
that for await...of closes it with return(). Neither holds in practice:
when next() rejects or throws, for await...of and for...of exit without
calling return(), and generator-based transforms finish once they
throw. A consumer left registered would pin the buffer forever and make
every other consumer hit the budget too.

Detach the consumer when its pull is rejected so later next() calls
return { done: true }, for both share() and shareSync(), and rewrite
the note.

Fixes #43
The spec said the upstream pull result is discarded but not what the
consumer that triggered the pull gets back. Specify that one upstream
batch is pulled and discarded, and that the pull then waits for budget
(or for the source to end, error, or be cancelled) before proceeding
normally. The consumer loses only the discarded batch, does not see a
false end of stream, and cannot drain or spin on the source. Describe
the policy that way.

Fixes #44
A synchronous next() cannot wait for budget, and the slowest consumer
cannot advance while another consumer's next() is running, so
"unbounded" and "drop-newest" have no workable meaning for
shareSync(). Give ShareSyncOptions.backpressure its own
SyncShareBackpressurePolicy enum with only the two policies that work,
so other values are rejected with a TypeError during conversion.

Fixes #45
cancel() only described what happens to existing consumers. Specify
that it also closes the broadcast's writer: pending writes reject with
an AbortError DOMException, a pending close promise resolves, drain
waiters are notified, and further writes behave as on any closed
writer. Also specify that push() after cancel() returns a consumer that
is already done, or already errored with the cancel reason, and that
reads no leftover buffered data.

Fixes #46
broadcastProtocol, shareProtocol and shareSyncProtocol were defined but
no algorithm checked for them. share() and shareSync() now check their
source for shareProtocol/shareSyncProtocol first and return the
method's result.

broadcast() takes no source, so add Stream.broadcastFrom(input,
options), which returns the result of broadcastProtocol when input has
it, and otherwise creates a broadcast channel, pumps input into its
writer, and returns the Broadcast.

Fixes #47
pipeToSync() ignored the result of endSync(), so when the writer could
not close synchronously it stayed open and pipeToSync() still returned
the byte total. Now it throws a TypeError without failing the writer,
since all the data was accepted and the caller may still close it
asynchronously. Add PipeToSyncOptions.failOnIncompleteClose (default
false) to fail the writer with that error instead; preventFail takes
precedence. Also state that a writer without endSync() is not closed.

Fixes #48
For async iterables the implementation itself created that already
yield normalized Uint8Array[] batches (such as push() readables and
pull() pipelines), wrapping them again only adds a layer to every read.
Since pull(), pipeTo() and the consumers call from() on their source,
that cost applied to every pipe from a push stream. Allow from() to
return such an input itself.

Fixes #49
…dinary

The spec did not say what the iterator results of returned iterables
look like, or what object a transform receives as its
TransformCallbackOptions. Ordinary objects inherit from
Object.prototype, so properties added there (such as then) would show
through. Add a note that these objects need not be ordinary objects and
that code should only rely on done/value and signal respectively.

Fixes #50
When the signal aborts while an upstream next() is pending, pull() must
both call the upstream iterator's return() and reject the pending pull,
but the order was unspecified even though it is observable to the
source and to the consumer. Require return() to be called first, as
Node.js does; the rejection does not wait for return() to settle.

Fixes #51
pull() only described closing the upstream iterator when the signal
aborts while an upstream next() is pending. If it aborts while no pull
is pending, later pulls reject, but nothing said whether or when the
upstream iterator's return() is called. Require it to be called no
later than the next pull, which then rejects with the abort reason
(as Node.js does), and allow implementations to call it as soon as the
signal aborts.

Fixes #52
BroadcastOptions has a signal member, but the broadcast() algorithm
never used it. Aborting the signal now cancels the broadcast with the
abort reason, as Broadcast.cancel(reason) does: the writer is closed,
pending writes reject, consumers' pulls reject with the reason, and
later consumers are already errored. An already-aborted signal returns
an already-cancelled broadcast. The abort steps are removed once the
writer is closed or errored and no consumers remain. This matches
Node.js.

Also simplify broadcastFrom(), whose separate signal check is now
covered by the broadcast being cancelled.
@cloudflare-workers-and-pages

cloudflare-workers-and-pages Bot commented Oct 6, 2026 •

Copy link
Copy Markdown

Deploying new-streams with  Cloudflare Pages  Cloudflare Pages

Latest commit: 2a883fc
Status: ✅  Deploy successful!
Preview URL: https://395856d4.new-streams.pages.dev
Branch Preview URL: https://spec-bugs.new-streams.pages.dev

View logs

Aborting BroadcastOptions.signal went through cancel(), which leaves the
writer closed, so an abort looked like a normal finish to the producer.
Fail the writer with the abort reason first, as push() does when its
signal aborts, and then cancel the broadcast so consumers and later
push() calls see the abort reason.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant