From 0c1cf62e67850f4208d371df0b4cd029d32b8a57 Mon Sep 17 00:00:00 2001
From: "Martin.Strambach" <martin.strambach@gmail.com>
Date: Thu, 9 Jul 2026 14:08:46 +0200
Subject: [PATCH] Fix multicaster race that leaves a new downstream without a
 producer

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>
---
 .../store/multicast5/ChannelManager.kt        |  6 +++
 .../multicast5/StoreChannelManagerTests.kt    | 46 +++++++++++++++++++
 2 files changed, 52 insertions(+)

diff --git a/multicast/src/commonMain/kotlin/org/mobilenativefoundation/store/multicast5/ChannelManager.kt b/multicast/src/commonMain/kotlin/org/mobilenativefoundation/store/multicast5/ChannelManager.kt
index 61948bd93..aa913f577 100644
--- a/multicast/src/commonMain/kotlin/org/mobilenativefoundation/store/multicast5/ChannelManager.kt
+++ b/multicast/src/commonMain/kotlin/org/mobilenativefoundation/store/multicast5/ChannelManager.kt
@@ -321,6 +321,12 @@ internal class StoreChannelManager<T>(
                 channels.removeAt(index)
                 if (!keepUpstreamAlive && channels.isEmpty()) {
                     producer?.cancelAndJoin()
+                    // Clear the dead producer reference right away instead of waiting for its
+                    // UpstreamFinished message. Otherwise a downstream added before that message
+                    // arrives would not restart the upstream and would never receive a value.
+                    // The stale UpstreamFinished is ignored by the identity check in
+                    // doHandleUpstreamClose.
+                    producer = null
                 }
             }
         }
diff --git a/multicast/src/commonTest/kotlin/org/mobilenativefoundation/store/multicast5/StoreChannelManagerTests.kt b/multicast/src/commonTest/kotlin/org/mobilenativefoundation/store/multicast5/StoreChannelManagerTests.kt
index c765b7c11..327693b6e 100644
--- a/multicast/src/commonTest/kotlin/org/mobilenativefoundation/store/multicast5/StoreChannelManagerTests.kt
+++ b/multicast/src/commonTest/kotlin/org/mobilenativefoundation/store/multicast5/StoreChannelManagerTests.kt
@@ -3,6 +3,8 @@ package org.mobilenativefoundation.store.multicast5
 import app.cash.turbine.test
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.awaitCancellation
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.consumeAsFlow
 import kotlinx.coroutines.flow.filterIsInstance
@@ -11,10 +13,12 @@ import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.test.advanceUntilIdle
 import kotlinx.coroutines.test.runTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
 
+@OptIn(ExperimentalCoroutinesApi::class)
 class StoreChannelManagerTests {
     @Test
     fun cancelledDownstreamChannelShouldNotCancelOtherChannels() =
@@ -72,6 +76,48 @@ class StoreChannelManagerTests {
             }
         }
 
+    @Test
+    fun downstreamAddedWhileUpstreamCancellationIsInFlightShouldRestartUpstream() =
+        runTest {
+            var upstreamCollectionCount = 0
+            val upstreamFlow =
+                flow {
+                    upstreamCollectionCount++
+                    if (upstreamCollectionCount == 1) {
+                        awaitCancellation()
+                    } else {
+                        emit(1)
+                    }
+                }
+            val channelManager =
+                StoreChannelManager(
+                    scope = this,
+                    bufferSize = 0,
+                    upstream = upstreamFlow,
+                    piggybackingDownstream = true,
+                    keepUpstreamAlive = false,
+                    onEach = { },
+                )
+            val firstChannel = Channel<ChannelManager.Message.Dispatch.Value<Int>>(Channel.UNLIMITED)
+            val secondChannel = Channel<ChannelManager.Message.Dispatch.Value<Int>>(Channel.UNLIMITED)
+
+            channelManager.addDownstream(firstChannel)
+            advanceUntilIdle()
+
+            // Removing the last downstream makes the actor suspend in doRemove on
+            // producer.cancelAndJoin(). Adding the next downstream right away enqueues its
+            // AddChannel message ahead of the producer's UpstreamFinished message, so the add is
+            // processed while the dead producer reference is still set.
+            channelManager.removeDownstream(firstChannel)
+            channelManager.addDownstream(secondChannel)
+            advanceUntilIdle()
+
+            val dispatchedValue = secondChannel.tryReceive().getOrNull()
+            assertEquals(1, dispatchedValue?.value)
+
+            channelManager.close()
+        }
+
     private fun createChannels(count: Int): List<Channel<ChannelManager.Message.Dispatch<Int>>> {
         return (1..count).map { Channel(Channel.UNLIMITED) }
     }