Fix multicaster race that leaves a new downstream without a producer#740
Open
MartinStrambach wants to merge 1 commit into
Open
Conversation
When the last downstream is removed, StoreChannelManager cancels the producer but keeps the stale reference until the producer's async UpstreamFinished message is processed. An AddChannel message that arrives in between sees producer != null and does not restart the upstream. If nothing was dispatched and piggybackingDownstream is enabled, doHandleUpstreamClose then parks that downstream as piggybacked without reactivation, so it never receives a value. Store hits this via FetcherController (piggybackingDownstream = true): cancelling a collector of store.stream() and immediately resubscribing to the same key can hang the new collector forever until an unrelated subscriber on the same key restarts the fetcher. Clear the producer reference in doRemove right after cancelAndJoin so a subsequent AddChannel starts a fresh producer. The stale UpstreamFinished of the cancelled producer is already ignored by the identity check in doHandleUpstreamClose. Signed-off-by: Martin.Strambach <martin.strambach@gmail.com>
93732b0 to
0c1cf62
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
The bug
StoreChannelManagerkeeps a staleproducerreference after the producer has been cancelled. When the last downstream is removed,doRemovecallsproducer?.cancelAndJoin()but does not clear the field — it waits for the producer'sUpstreamFinishedmessage to do that. That message is sent asynchronously fromSharedFlowProducer.start()'s joiner coroutine, so there is a window in which anAddChannelmessage is processed first:AddChannel(#2)→doAdd→activateIfNecessary()seesproducer != null(dead) → upstream is not restarted.UpstreamFinished→doHandleUpstreamClose: nothing was dispatched (dispatchedValue == false), so withpiggybackingDownstream = truethe new downstream is parked as piggybacked,produceris set tonull, and reactivation only happens forleftovers(empty).The new downstream now hangs forever with zero emissions. It only recovers if some later downstream on the same multicaster restarts the producer, which dispatches to piggybacked channels too.
The ordering is easy to hit in practice:
Multicaster.newDownstream()sendsRemoveChannelfromonCompletionandAddChannelfromonStart, so cancelling a collector and immediately re-collecting (a routine "restart the subscription" pattern, e.g.flatMapLatest-style UI code overstore.stream()) races the resubscribe against the in-flightUpstreamFinished. SinceFetcherControllerbuilds its multicaster withpiggybackingDownstream = true, the visible symptom at the Store level isstore.stream()collectors that never receive anything — we tracked down a production "infinite loading" bug in our app to exactly this race.The fix
Clear the
producerreference indoRemoveimmediately aftercancelAndJoin(), so a subsequentAddChannelstarts a fresh producer. This is safe:UpstreamFinishedis already ignored by the identity check at the top ofdoHandleUpstreamClose(if (this.producer !== producer) return).Testing
Added a deterministic regression test in
StoreChannelManagerTests. It runs on a single-threaded test dispatcher and exploits the actor's rendezvous channel: while the actor is suspended indoRemove'scancelAndJoin(), the nextaddDownstreamparks as a pending sender ahead ofUpstreamFinished, reproducing the exact production ordering. The upstream suspends on its first collection (fetch in flight, then cancelled) and emits on the second.expected:<1> but was:<null>— the second downstream never receives a value.:multicast:jvmTest,:store:jvmTest, and:multicast:ktlintCheckall pass.