From 132a437c16a78a93626e179baf4dc3ba176853c2 Mon Sep 17 00:00:00 2001 From: Denver Coneybeare Date: Mon, 20 Jul 2026 19:59:47 -0400 Subject: [PATCH 1/5] DataConnectBidiConnectStream.kt: minor prefactor before the real work --- .../dataconnect/core/DataConnectBidiConnectStream.kt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/core/DataConnectBidiConnectStream.kt b/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/core/DataConnectBidiConnectStream.kt index 6b9c159ddd3..fffca8a104f 100644 --- a/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/core/DataConnectBidiConnectStream.kt +++ b/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/core/DataConnectBidiConnectStream.kt @@ -233,7 +233,10 @@ internal class DataConnectBidiConnectStream( .buffer(capacity = 64) // Use a finite buffer to activate gRPC flow control, when needed .shareIn( coroutineScope, - started = SharingStarted.WhileSubscribed(replayExpirationMillis = 0), + started = + SharingStarted.WhileSubscribed( + replayExpirationMillis = 0, + ), replay = 0, ) From 34f8c9015736376959a817a19e28f238db4f726b Mon Sep 17 00:00:00 2001 From: Denver Coneybeare Date: Mon, 20 Jul 2026 20:00:05 -0400 Subject: [PATCH 2/5] DataConnectBidiConnectStream.kt: set stopTimeoutMillis=15.seconds, to allow a grace period before killing the active connection --- .../core/DataConnectBidiConnectStream.kt | 1 + .../core/QuerySubscriptionImplUnitTest.kt | 122 ++++++++++++++++++ 2 files changed, 123 insertions(+) diff --git a/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/core/DataConnectBidiConnectStream.kt b/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/core/DataConnectBidiConnectStream.kt index fffca8a104f..a8d8b40f93b 100644 --- a/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/core/DataConnectBidiConnectStream.kt +++ b/firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/core/DataConnectBidiConnectStream.kt @@ -235,6 +235,7 @@ internal class DataConnectBidiConnectStream( coroutineScope, started = SharingStarted.WhileSubscribed( + stopTimeoutMillis = 15_000, replayExpirationMillis = 0, ), replay = 0, diff --git a/firebase-dataconnect/src/test/kotlin/com/google/firebase/dataconnect/core/QuerySubscriptionImplUnitTest.kt b/firebase-dataconnect/src/test/kotlin/com/google/firebase/dataconnect/core/QuerySubscriptionImplUnitTest.kt index 7f37015c4a1..eaaa04b0e6a 100644 --- a/firebase-dataconnect/src/test/kotlin/com/google/firebase/dataconnect/core/QuerySubscriptionImplUnitTest.kt +++ b/firebase-dataconnect/src/test/kotlin/com/google/firebase/dataconnect/core/QuerySubscriptionImplUnitTest.kt @@ -19,6 +19,7 @@ import android.content.Context.CONNECTIVITY_SERVICE import android.net.ConnectivityManager import androidx.test.ext.junit.runners.AndroidJUnit4 import app.cash.turbine.ReceiveTurbine +import app.cash.turbine.TurbineContext import app.cash.turbine.test import app.cash.turbine.turbineScope import com.google.firebase.appcheck.interop.InteropAppCheckTokenProvider @@ -26,6 +27,7 @@ import com.google.firebase.auth.internal.InternalAuthProvider import com.google.firebase.dataconnect.DataConnectSettings import com.google.firebase.dataconnect.FirebaseDataConnect.CallerSdkType import com.google.firebase.dataconnect.QueryRef +import com.google.firebase.dataconnect.QuerySubscriptionResult import com.google.firebase.dataconnect.core.DataConnectAuth.GetAuthTokenResult import com.google.firebase.dataconnect.core.DataConnectBidiConnectStream.Companion.setReconnectPendingAuthTokenForTesting import com.google.firebase.dataconnect.core.DataConnectBidiConnectStream.Companion.unsetReconnectPendingAuthTokenForTesting @@ -111,6 +113,7 @@ import io.kotest.property.arbitrary.az import io.kotest.property.arbitrary.distinct import io.kotest.property.arbitrary.enum import io.kotest.property.arbitrary.int +import io.kotest.property.arbitrary.long import io.kotest.property.arbitrary.map import io.kotest.property.arbitrary.next import io.kotest.property.arbitrary.of @@ -2036,6 +2039,125 @@ class QuerySubscriptionImplUnitTest { } } + @Test + fun `connection is kept alive for exactly 15 seconds after last subscriber unsubscribes`() = + testConnectionGracePeriod(Arb.long(0L until 15_000L)) { context -> + context.clientCollector1.cancelAndIgnoreRemainingEvents() + + // Wait a random duration less than 15 seconds + delay(context.delayMillis.milliseconds) + + // Verify that the connection is still open (no close event received on serverCollector) + context.serverCollector.asChannel().tryReceive().getOrNull() shouldBe null + + // Measure the remaining time until the client closes the connection + val time1 = @OptIn(ExperimentalCoroutinesApi::class) context.testScheduler.currentTime + context.serverCollector.awaitUntilClientClosesConnection() + val time2 = @OptIn(ExperimentalCoroutinesApi::class) context.testScheduler.currentTime + + (time2 - time1) shouldBe (15_000L - context.delayMillis) + } + + @Test + fun `re-subscribing within 15-second grace period keeps the connection alive and reuses it`() = + testConnectionGracePeriod(Arb.long(100L until 15_000L)) { context -> + context.clientCollector1.cancelAndIgnoreRemainingEvents() + + // Wait a random duration less than 15 seconds + delay(context.delayMillis.milliseconds) + + // Re-subscribe clientCollector2 + val clientCollector2 = + context.subscription.flow.testIn(context.backgroundScope, name = "clientCollector2") + + // Verify that the server receives a subscribe request for the connection, but the connection + // ID remains the same (proving reuse) + val subscribeRequest = context.serverCollector.awaitUntilSubscribeStreamRequest() + subscribeRequest.connectionId shouldBe context.initialConnectionId + + clientCollector2.cancelAndIgnoreRemainingEvents() + // Wait for the new grace period to expire so we don't leak connection closure errors/events + context.serverCollector.awaitUntilClientClosesConnection() + } + + @Test + fun `re-subscribing after 15-second grace period establishes a new connection`() = + testConnectionGracePeriod(Arb.long(15_000L..100_000L)) { context -> + context.clientCollector1.cancelAndIgnoreRemainingEvents() + + // Wait a random duration of at least 15 seconds + delay(context.delayMillis.milliseconds) + + // Verify that the connection is closed + context.serverCollector.awaitUntilClientClosesConnection() + + // Re-subscribe clientCollector2 + val clientCollector2 = + context.subscription.flow.testIn(context.backgroundScope, name = "clientCollector2") + + // Verify that a new connection is established + val connection2 = context.serverCollector.awaitConnectRpcStarted() + connection2.connectionId shouldNotBe context.initialConnectionId + context.serverCollector.awaitUntilInitStreamRequest() + context.serverCollector.awaitUntilSubscribeStreamRequest() + + clientCollector2.cancelAndIgnoreRemainingEvents() + // Wait for connection to close to keep clean state + context.serverCollector.awaitUntilClientClosesConnection() + } + + @OptIn(ExperimentalCoroutinesApi::class) + private fun testConnectionGracePeriod( + delayMillisArb: Arb, + block: suspend TurbineContext.(TestConnectionGracePeriodContext) -> Unit + ) = runTest { + val server = runningInProcessDataConnectServer() + checkAll(propTestConfig, delayMillisArb, Arb.dataConnect.operationName(), testVariablesArb()) { + delayMillis, + operationName, + variables -> + runWithDataConnect(server) { dataConnect -> + val subscription = querySubscription(dataConnect, operationName, variables) + + turbineScope { + val serverCollector = server.events.testIn(backgroundScope, name = "serverCollector") + val clientCollector1 = + subscription.flow.testIn(backgroundScope, name = "clientCollector1") + + // Wait for initial connection and subscribe + val connection = serverCollector.awaitConnectRpcStarted() + serverCollector.awaitUntilInitStreamRequest() + serverCollector.awaitUntilSubscribeStreamRequest() + + val context = + TestConnectionGracePeriodContext( + delayMillis = delayMillis, + serverCollector = serverCollector, + clientCollector1 = clientCollector1, + initialConnectionId = connection.connectionId, + subscription = subscription, + backgroundScope = backgroundScope, + testScheduler = testScheduler, + ) + + block(context) + + serverCollector.cancelAndIgnoreRemainingEvents() + } + } + } + } + + private data class TestConnectionGracePeriodContext( + val delayMillis: Long, + val serverCollector: ReceiveTurbine, + val clientCollector1: ReceiveTurbine>, + val initialConnectionId: InProcessDataConnectGrpcStreamingServer.ConnectionId, + val subscription: QuerySubscriptionImpl, + val backgroundScope: kotlinx.coroutines.CoroutineScope, + val testScheduler: TestCoroutineScheduler, + ) + private fun runningInProcessDataConnectServer(): InProcessDataConnectGrpcStreamingServer { val server = InProcessDataConnectGrpcStreamingServer() cleanups.register(server) From eb91fe1de4b994af4e8d78f0242f9832fdd69f17 Mon Sep 17 00:00:00 2001 From: Denver Coneybeare Date: Wed, 22 Jul 2026 21:20:14 -0400 Subject: [PATCH 3/5] CHANGELOG.md: entry added [no ci] --- firebase-dataconnect/CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/firebase-dataconnect/CHANGELOG.md b/firebase-dataconnect/CHANGELOG.md index 4bd4dc28301..27cac5d953b 100644 --- a/firebase-dataconnect/CHANGELOG.md +++ b/firebase-dataconnect/CHANGELOG.md @@ -7,6 +7,10 @@ [#8446](https://github.com/firebase/firebase-android-sdk/pull/8446), [#8456](https://github.com/firebase/firebase-android-sdk/pull/8456), [#8460](https://github.com/firebase/firebase-android-sdk/pull/8460)) +- [changed] Wait for 15 seconds before closing realtime streaming connection + with backend after last subscriber unsubscribes (instead of closing the + connection immediately). + ([#NNNN](https://github.com/firebase/firebase-android-sdk/pull/NNNN)) # 17.3.2 From 8260c6a2532b4904632e0197287e58566d962338 Mon Sep 17 00:00:00 2001 From: Denver Coneybeare Date: Fri, 24 Jul 2026 13:02:26 -0400 Subject: [PATCH 4/5] QuerySubscriptionImplUnitTest.kt: Create CONNECTION_GRACE_PERIOD_MS rather than hardcoding 15 seconds to improve readabiltiy of the test code. --- .../core/QuerySubscriptionImplUnitTest.kt | 33 +++++++++++++------ 1 file changed, 23 insertions(+), 10 deletions(-) diff --git a/firebase-dataconnect/src/test/kotlin/com/google/firebase/dataconnect/core/QuerySubscriptionImplUnitTest.kt b/firebase-dataconnect/src/test/kotlin/com/google/firebase/dataconnect/core/QuerySubscriptionImplUnitTest.kt index eaaa04b0e6a..159918022b3 100644 --- a/firebase-dataconnect/src/test/kotlin/com/google/firebase/dataconnect/core/QuerySubscriptionImplUnitTest.kt +++ b/firebase-dataconnect/src/test/kotlin/com/google/firebase/dataconnect/core/QuerySubscriptionImplUnitTest.kt @@ -2040,11 +2040,11 @@ class QuerySubscriptionImplUnitTest { } @Test - fun `connection is kept alive for exactly 15 seconds after last subscriber unsubscribes`() = - testConnectionGracePeriod(Arb.long(0L until 15_000L)) { context -> + fun `connection is kept alive for exactly the grace period after last subscriber unsubscribes`() = + testConnectionGracePeriod(Arb.long(0L until CONNECTION_GRACE_PERIOD_MS)) { context -> context.clientCollector1.cancelAndIgnoreRemainingEvents() - // Wait a random duration less than 15 seconds + // Wait a random duration less than CONNECTION_GRACE_PERIOD_MS delay(context.delayMillis.milliseconds) // Verify that the connection is still open (no close event received on serverCollector) @@ -2055,15 +2055,15 @@ class QuerySubscriptionImplUnitTest { context.serverCollector.awaitUntilClientClosesConnection() val time2 = @OptIn(ExperimentalCoroutinesApi::class) context.testScheduler.currentTime - (time2 - time1) shouldBe (15_000L - context.delayMillis) + (time2 - time1) shouldBe (CONNECTION_GRACE_PERIOD_MS - context.delayMillis) } @Test - fun `re-subscribing within 15-second grace period keeps the connection alive and reuses it`() = - testConnectionGracePeriod(Arb.long(100L until 15_000L)) { context -> + fun `re-subscribing within grace period keeps the connection alive and reuses it`() = + testConnectionGracePeriod(Arb.long(100L until CONNECTION_GRACE_PERIOD_MS)) { context -> context.clientCollector1.cancelAndIgnoreRemainingEvents() - // Wait a random duration less than 15 seconds + // Wait a random duration less than CONNECTION_GRACE_PERIOD_MS delay(context.delayMillis.milliseconds) // Re-subscribe clientCollector2 @@ -2081,11 +2081,13 @@ class QuerySubscriptionImplUnitTest { } @Test - fun `re-subscribing after 15-second grace period establishes a new connection`() = - testConnectionGracePeriod(Arb.long(15_000L..100_000L)) { context -> + fun `re-subscribing after grace period establishes a new connection`() = + testConnectionGracePeriod( + Arb.long(CONNECTION_GRACE_PERIOD_MS..(CONNECTION_GRACE_PERIOD_MS * 4)) + ) { context -> context.clientCollector1.cancelAndIgnoreRemainingEvents() - // Wait a random duration of at least 15 seconds + // Wait a random duration of at least CONNECTION_GRACE_PERIOD_MS delay(context.delayMillis.milliseconds) // Verify that the connection is closed @@ -2395,3 +2397,14 @@ private class SequenceRandom(jitters: Sequence) : Random() { return lock.withLock { iterator.next() + 0.5 } } } + +/** + * The amount of time, in milliseconds, that [DataConnectBidiConnectStream] keeps the physical + * connection with the backend alive after the last subscriber unsubscribes. By keeping the + * connection alive for a short amount of time rather than closing it immediately it improves the + * latency and reduces the backend load if a new subscriber were to subscribe within this grace + * period. This rapid unsubscription and resubscription could happen, for example, between activity + * or fragment transitions in an application where the old activity/fragment unsubscribes in its + * onDestory() and the new activity/fragment subscribes in its onCreate(). + */ +const val CONNECTION_GRACE_PERIOD_MS = 15_000L From 88b81364c52280c400d77381dbe64589ed19718a Mon Sep 17 00:00:00 2001 From: Denver Coneybeare Date: Fri, 24 Jul 2026 14:21:04 -0400 Subject: [PATCH 5/5] QuerySubscriptionImplUnitTest.kt: fix flaky test "connection is kept alive for exactly the grace period after last subscriber unsubscribes" by ignoring spurious "cancel" StreamRequest --- .../core/QuerySubscriptionImplUnitTest.kt | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/firebase-dataconnect/src/test/kotlin/com/google/firebase/dataconnect/core/QuerySubscriptionImplUnitTest.kt b/firebase-dataconnect/src/test/kotlin/com/google/firebase/dataconnect/core/QuerySubscriptionImplUnitTest.kt index 159918022b3..3f4bb15746c 100644 --- a/firebase-dataconnect/src/test/kotlin/com/google/firebase/dataconnect/core/QuerySubscriptionImplUnitTest.kt +++ b/firebase-dataconnect/src/test/kotlin/com/google/firebase/dataconnect/core/QuerySubscriptionImplUnitTest.kt @@ -93,9 +93,11 @@ import io.kotest.assertions.print.print import io.kotest.assertions.withClue import io.kotest.common.DelicateKotest import io.kotest.common.ExperimentalKotest +import io.kotest.matchers.booleans.shouldBeTrue import io.kotest.matchers.collections.shouldBeIn import io.kotest.matchers.collections.shouldContainExactly import io.kotest.matchers.maps.shouldBeEmpty +import io.kotest.matchers.nulls.shouldBeNull import io.kotest.matchers.result.shouldBeSuccess import io.kotest.matchers.shouldBe import io.kotest.matchers.shouldNotBe @@ -2047,8 +2049,14 @@ class QuerySubscriptionImplUnitTest { // Wait a random duration less than CONNECTION_GRACE_PERIOD_MS delay(context.delayMillis.milliseconds) - // Verify that the connection is still open (no close event received on serverCollector) - context.serverCollector.asChannel().tryReceive().getOrNull() shouldBe null + // Verify that the connection is still open (no close event received on serverCollector). + // Based on the timing, we _may_ receive the "cancel" event, which is expected. + val event = context.serverCollector.asChannel().tryReceive().getOrNull() + if (event != null) { + val streamRequest = event.shouldBeInstanceOf().streamRequest + streamRequest.hasCancel().shouldBeTrue() + context.serverCollector.asChannel().tryReceive().getOrNull().shouldBeNull() + } // Measure the remaining time until the client closes the connection val time1 = @OptIn(ExperimentalCoroutinesApi::class) context.testScheduler.currentTime