Repository navigation
fix(sdk): a throwing subscriber silently stops delivery to the others #473
Description
Activity
Scope is wider than the three fan-out loops — three more unguarded paths
Found while documenting the throwing-handler contract on #470. This issue as filed named only the
_subscribersloops atcontroller.ts:28,:45,:58. There are three more places a consumer callback runs unprotected, and the first is the one most likely to fire.1.
subscribe()'s initialstatuscall (clients/ts/src/stream/controller.ts:130)subscribe(subscriber: StreamSubscriber<T>): () => void { this._subscribers.add(subscriber); subscriber.status?.(this._status); // synchronous, unguarded return () => { … }; }
This runs before the transport is involved at all, so the transport's guards can't help. A throw propagates straight back out of
.subscribe(), and leaves the caller in a genuinely bad state: the subscriber is already registered (so it keeps receiving), but the unsubscribe function was never returned, so it can never be removed. It also always fires — everysubscribe()delivers the current state immediately — which makes it the first thing a throwingstatushandler does, not an edge case. The docs' own example handler isupdateIndicator(state), exactly the shape that throws when a DOM node is missing.2. A concurrent
for awaitis starved, not just later subscriberscontroller.ts:28-36runs the_subscribersloop before resolving_waiters/ pushing to_buffer. So a throw in any subscriber means the async-iterator consumer never sees that event either — the iterator isn't a separate delivery path, it's downstream of the loop. The issue text said "later subscribers"; it should say "later subscribers and anyfor awaitconsumer".3.
liveQuery()'s backfill flush swallows and discards (clients/ts/src/stream/live-query.ts:56, 84-91)_runBackfillwraps theinitial()call and the whole buffered-event flush in onetry, whosecatchsets_buffer = []. So a throw frominitial(), or fromnext()partway through the flush, is absorbed and takes the remaining buffered events with it — silently, with noconsole.error, noerrorcallback, and no way for the consumer to know events were dropped. That is worse than the fan-out case: not just delayed or skipped delivery, but data loss on the backfill seam the feature exists to provide.What this changes about the fix
Guarding only the three loops would leave the most-likely path (1) and the most-damaging path (3) unfixed, while making the docs' "isolated and logged" claim look complete. Worth deciding together:
- Path 1 needs a decision the loops don't: if the initial
statusthrows, doessubscribe()still return the unsubscribe function (recommended — the caller needs it precisely because their handler is broken), or roll back the registration? - Path 3 needs to distinguish "the fetch failed" from "your callback threw" — they currently share one
catch, and only the first justifies clearing the buffer.
#470 documents all three as known carve-outs in
sdk/reference.md§Error Handling rather than fixing them, since they live incontroller.ts/live-query.tsrather than the transport. Whatever lands here should shorten that section.— Claude Opus 5, via Claude Code
- Path 1 needs a decision the loops don't: if the initial
One more consequence of path 1 (
subscribe()'s unguarded initialstatuscall), found reviewing #470 and worth having here before anyone picks this up — it changes what a fix has to do.Through
.liveQuery(), the same line strands an unclosable connection.LiveQuery's constructor subscribes withstatus: (s) => subscriber.status?.(s)(live-query.ts:41), so a caller'sstatushandler is on the far end ofcontroller.ts:130exactly as it is for a bare.subscribe(). The difference is what the throw unwinds through:this._unsubStream = stream.subscribe({ … }); // live-query.ts:32
The assignment happens after
subscribe()returns. A throw from that first synchronousstatuscall therefore unwinds before_unsubStreamis ever set, out through theLiveQueryconstructor, out throughclient.liveQuery(). The caller gets no object at all — while the controller has already registered the subscriber (controller.ts:129, before thestatuscall) and the SSE connection is open and re-dialing. There is no handle to.close()it with, and_unsubStream?.()at:99is unreachable because there is no instance.So it is strictly worse than the bare-
subscribe()case: that one at least leaves the caller holding a live stream they can close, just without the unsubscribe function for the one subscriber.Why this bears on the open design question. "Should
subscribe()still return the unsubscribe function when the handler threw?" isn't sufficient on its own —LiveQuerydiscards the return value on the way out of the constructor, so fixing onlysubscribe()'s return path leaves the.liveQuery()case exactly as broken. Whatever the fix is, it has to hold for a caller who never receives the object that owns the handle. Guarding the initialstatuscall atcontroller.ts:130the way the transport guards its own callbacks would cover both at once, which is an argument for fixing it there rather than at the return.Pre-existing on
mainand outside #470's diff, so not fixed there. Verified againstlive-query.ts:19,32,41,99andcontroller.ts:129-131.Three more behaviours found while re-reviewing the #470 docs, all executed rather than read. None are new regressions — they are all pre-existing on
main— but they change the scope of a fix, so recording them here rather than letting them be rediscovered.1. For a concurrent
for await, the event is dropped, not merely delayed.controller.ts:27-38runs the subscriber fan-out at:28-30before both_waiters.shift()at:32and_buffer.push()at:36. A throw at:29exits the wholeonEventarrow function, so the event never reaches a waiting iterator and never lands in the buffer for a laternext(). It is gone. "Starved" was the wrong word for this — the iterator isn't waiting on something late, it is missing something that will never arrive.2. A throw on the terminal
closedstatus leaves afor awaithanging forever.Same shape, worse consequence.
controller.ts:40-55sets_statusat:44, fans out at:45-47, and only then, at:48-54, sets_done = trueand resolves the outstanding waiters. A throwingstatushandler aborts the loop at:46, so_donestaysfalseand every waiter stays pending — against a stream the transport has already torn down. This is reachable from every transport-initiated terminal close: a4xx,SSE_REDIRECT,SSE_BAD_CONTENT_TYPE.It is recoverable, but only by something calling
close():controller.ts:150-163sets_doneand resolves waiters outside theif (this._status !== "closed")guard, so on a second pass the fan-out is skipped, nothing throws, and the iterator terminates. A caller who is sitting infor awaitand not watchingstatushas no reason to make that call.3.
.connected()can reject with its timeout against a stream that is alreadylive.connected()appends its internal watcher to_subscribersatcontroller.ts:123. In the ordering the docs themselves demonstrate —subscribe(...)and thenawait stream.connected()— the user's subscriber is earlier in the set, so a throw from itsstatushandler aborts the fan-out before the watcher is reached. The watcher never observeslive, the timer fires, and the promise rejects withStream did not connect within Nmswhilecontroller.statusreadslive.That makes the rejection mean neither of the two things
.connected()documents it as meaning.
Bearing on the fix. (1) and (3) are both consequences of the fan-out loops being unguarded, so guarding each
sub.next(...)/sub.status?.(...)/sub.error?.(...)the way the transport guards its own callbacks resolves them together. (2) needs one thing more: the_done/waiter bookkeeping at:48-54has to be reachable even when a subscriber throws — moving it before the fan-out, or into afinally, since correctness of the iterator's termination shouldn't depend on subscriber behaviour.Also worth noting for whoever picks this up:
live-query.ts:58-61drops the buffer on a backfill errorResultwithout going through thecatchat:87-91—_buffering = falseandreturn, with_bufferstill populated and never flushed or cleared. Same user-visible outcome as the throwing-handler case (buffered events silently lost), different path, so a fix aimed only at thecatchwould miss it.Verified against
controller.ts:27-55,123,150-163andlive-query.ts:56-91ata67c0648. #470 documents all of this insdk/reference.mdandsdk/streaming.mdbut does not change the behaviour.- added 17 commits that reference this issue
on Aug 18, 2026
Area: sdk — streaming. Surfaced reviewing #470; the crash it caused is fixed there, this is the residual.
StreamControllerfans out to subscribers with a bare loop (clients/ts/src/stream/controller.ts:28,:45,:58):One subscriber throwing aborts the loop, so every subscriber registered after it silently misses that event, status, or error. Iteration order is insertion order, so which consumers are affected depends on the order they happened to subscribe in.
Why now
#470 hit the severe form of this. The transport called
onStatus/onErrorunguarded, so a throw propagated out of the fan-out, unwound the reconnect loop, and ended as a process-fatal unhandled rejection. That is fixed in the transport (sse.tsnow isolates all three callbacks, matching what_dispatchalready did), which contains the blast radius to "the stream survives".What it does not fix is the fan-out itself: with two subscribers on one stream, a throw in the first still means the second never hears about that frame. Silent, order-dependent, and invisible to the throwing consumer.
The full surface
Five behaviours, found across three review rounds on #470 (details and execution traces in the comments below). #470 documents all of these in
sdk/reference.mdandsdk/streaming.mdbut changes none of them, so the docs currently point here for every one.controller.ts:28,:45,:58— the three fan-out loops:27-38for awaitloses the event entirely — the fan-out precedes both_waiters.shift()and_buffer.push(), so it is neither delivered nor queued:40-55closedstatus the aborted loop skips_done = trueand the waiter resolution at:48-54, so afor awaitnever exits. Only a laterclose()heals it (:158-162sit outside the status guard);unsub()cannot, because auto-close at:134requires_waiters.length === 0and the hung iterator is itself a waitercontroller.ts:130—subscribe()'s initial synchronousstatuscall.subscribe(): subscriber registered, no unsubscribe returned. Out of.liveQuery()(which forwardsstatusatlive-query.ts:41): no handle returned at all, so nothing can.close()it;_runBackfillnever starts soinitial()never fires; and the stream opened a line earlier keeps running — connected, reconnecting on its own — with onlyopts.signalable to stop it, buffering every event into_bufferbecause_bufferingnever flipslive-query.ts:87-91catchabsorbs a throw frominitial()or fromnext()mid-flush and a rejection from the fetch itself (a rejectingauth, a relativebaseURL) and clears_buffer, silently dropping the rest.:58-61drops it a second way — an errorResultreturns early with_bufferstill populated and never flushed, bypassing thecatchentirelySymptom worth naming because it presents as its own puzzle: a throwing
statushandler registered before.connected()makes that promise reject with its timeout while.statusalready readslive—connected()'s watcher (:123) is added after the consumer's subscriber, so the throw aborts the fan-out before it observeslive. Same mechanism as 1; no separate fix needed.Scope
Isolate per subscriber in all three loops, so one bad callback costs only its own delivery:
That covers 1 and 2. It is not sufficient on its own — three things need separate decisions:
_done = trueand the waiter resolution ahead of the fan-out, or into afinally. Iterator termination should not depend on subscriber behaviour.controller.ts:130, not atsubscribe()'s return. "Return the unsubscribe function anyway" does not help theliveQuery()case, becauseLiveQuery's constructor discards the return value on the way out (live-query.ts:32assigns_unsubStreamonly aftersubscribe()returns). Guarding the call site fixes both at once.catchat:87misses the error-Resultearly return at:58-61.Worth deciding at the same time:
errorrisks a loop if theerrorhandler is the one throwing.nextshould differ fromstatus/error. A throw innextis the likeliest in practice (it runs consumer rendering/business logic) and the most costly to swallow silently.Resultshould flush the buffer rather than drop it. Arguably the events are still valid even though the historical query failed.Acceptance
next/status/errorfor awaitits eventstatushandler on the terminalclosedstatus still terminates a concurrentfor awaitstatushandler does not escape.subscribe()or.liveQuery(); both still return their handleinitial()/next()during the backfill flush does not discard the remaining buffered events, and a backfill errorResultis handled deliberately either wayfor awaitconcurrent with a throwing subscribersdk/reference.md("If your own callback throws") andsdk/streaming.mdupdated — they currently document all five as live behaviour and link hereRelated: #470 (fixed the transport-side crash, documented the rest), #389 (StreamController buffering, same file), #449 (live-query dedup boundary, same file).