From 9235ab401c3fa28336676751e009681dfec87f7c Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Wed, 8 Jul 2026 16:23:38 -0400 Subject: [PATCH] fix(chat): prevent duplicate event streams from concurrent open open()/close() did a non-atomic check-then-act on streamRef while being invoked from multiple triggers (login, lifecycle onStart, network reconnect, feature-flag, heartbeat) on the multi-threaded IO scope. Two threads could both observe a null ref and each open a gRPC stream, leaving a second orphaned stream whose reconnect loop kept running. The server rejects the duplicate with ABORTED 'stream already exists' and the two streams knock each other offline in a persistent duel. Serialize open/close under a lock so only one stream can exist, and guard onError to clear only its own ref. Adds a 32-thread concurrency test. --- .../controllers/EventStreamingController.kt | 27 +++++++--- .../EventStreamingControllerTest.kt | 51 +++++++++++++++++-- 2 files changed, 65 insertions(+), 13 deletions(-) diff --git a/services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt b/services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt index 9bac417ea..e6e46c3f5 100644 --- a/services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt +++ b/services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt @@ -20,13 +20,20 @@ class EventStreamingController @Inject constructor( private val _chatUpdates = Channel(capacity = Channel.UNLIMITED) val chatUpdates: Flow = _chatUpdates.receiveAsFlow() + // Guards all reads/writes of [streamRef]. open()/close() are invoked + // concurrently from multiple triggers (login, lifecycle onStart, network + // reconnect, feature-flag, heartbeat) on the multi-threaded IO scope, so + // the check-then-act on streamRef must be atomic — otherwise two threads + // both observe a null ref and each open a stream, producing a second + // orphaned gRPC stream and the server's "ABORTED: stream already exists". + private val lock = Any() private var streamRef: EventStreamReference? = null - val isConnected: Boolean get() = streamRef != null + val isConnected: Boolean get() = synchronized(lock) { streamRef != null } - val isStreamActive: Boolean get() = streamRef?.isActive == true + val isStreamActive: Boolean get() = synchronized(lock) { streamRef?.isActive == true } - fun open(scope: CoroutineScope): Boolean { + fun open(scope: CoroutineScope): Boolean = synchronized(lock) { if (streamRef != null) { trace("EventStreamingController: Stream already open, skipping") return true @@ -37,7 +44,8 @@ class EventStreamingController @Inject constructor( return false } - streamRef = repository.openEventStream( + lateinit var openedRef: EventStreamReference + openedRef = repository.openEventStream( scope = scope, owner = owner, onEvent = { update -> @@ -47,15 +55,18 @@ class EventStreamingController @Inject constructor( onError = { error -> trace("EventStreamingController: Stream error: ${error.message}") // Clear the ref so the next heartbeat, lifecycle, or network - // event creates a fresh stream. The framework guarantees this - // fires only once per stream, so no risk of clearing a newer ref. - streamRef = null + // event creates a fresh stream. Only clear if it is still THIS + // stream — never null out a newer ref opened after us. + synchronized(lock) { + if (streamRef === openedRef) streamRef = null + } }, ) + streamRef = openedRef return true } - fun close() { + fun close() = synchronized(lock) { streamRef?.destroy() streamRef = null } diff --git a/services/flipcash/src/test/kotlin/com/flipcash/services/controllers/EventStreamingControllerTest.kt b/services/flipcash/src/test/kotlin/com/flipcash/services/controllers/EventStreamingControllerTest.kt index 24b2eedd8..1d73efd87 100644 --- a/services/flipcash/src/test/kotlin/com/flipcash/services/controllers/EventStreamingControllerTest.kt +++ b/services/flipcash/src/test/kotlin/com/flipcash/services/controllers/EventStreamingControllerTest.kt @@ -11,10 +11,15 @@ import io.mockk.every import io.mockk.mockk import io.mockk.verify import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.test.UnconfinedTestDispatcher import kotlinx.coroutines.test.runTest +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import kotlin.concurrent.thread import kotlin.test.Test +import kotlin.test.assertEquals import kotlin.test.assertNotNull @OptIn(ExperimentalCoroutinesApi::class) @@ -69,11 +74,40 @@ class EventStreamingControllerTest { fun `chatUpdates flow is accessible`() { assertNotNull(controller.chatUpdates) } + + @Test + fun `concurrent open calls create only one stream`() { + stubOwner() + val scope = CoroutineScope(Dispatchers.Default) + val threadCount = 32 + val startLatch = CountDownLatch(1) + val doneLatch = CountDownLatch(threadCount) + + repeat(threadCount) { + thread { + startLatch.await() + controller.open(scope) + doneLatch.countDown() + } + } + + // Release all threads at once to maximize the check-then-act race. + startLatch.countDown() + doneLatch.await(5, TimeUnit.SECONDS) + + // Exactly one gRPC stream must be opened; opening two produces the + // server-side "ABORTED: stream already exists" duel. + assertEquals(1, repository.openCount) + } } private class FakeEventStreamingRepository : EventStreamingRepository { - var opened = false + private val lock = Any() + var openCount = 0 + private set + val opened: Boolean get() = openCount > 0 var lastStreamRef: EventStreamReference? = null + private set override fun openEventStream( scope: CoroutineScope, @@ -81,9 +115,16 @@ private class FakeEventStreamingRepository : EventStreamingRepository { onEvent: (ChatUpdate) -> Unit, onError: (Throwable) -> Unit, ): EventStreamReference { - opened = true - val ref = mockk(relaxed = true) - lastStreamRef = ref - return ref + // Widen the controller's check-then-act window so an unsynchronized + // open() lets multiple threads reach here concurrently. Mock creation + // and counting are serialized here so the assertion measures the + // controller's concurrency, not mockk's. + Thread.sleep(20) + return synchronized(lock) { + openCount++ + val ref = mockk(relaxed = true) + lastStreamRef = ref + ref + } } }