Skip to content

Commit 779eda2

Browse files
bmc08gtbmc08gt
authored
fix: debounce network-state subscription to connected boolean only (#958)
* fix(contacts): debounce network-state subscription to connected boolean only The ContactCoordinator subscribed to the full NetworkState (connected + signalStrength + type). Signal-strength or connection-type fluctuations passed through distinctUntilChanged() and each triggered a full contact sync — up to 44 times in 270ms during a charger-connect event. Fix: map to just the connected boolean before distinctUntilChanged so only actual connectivity transitions trigger sync. Bugsnag: 6a3686f3d2c3b94118b56442 * fix(services/bidi): treat ClosedSendChannelException as retryable on initial send When the gRPC response flow errors immediately after creation, the .catch block closes the requestChannel before send(initialRequest()) completes, throwing ClosedSendChannelException. This is a transient condition (the stream died during setup) and should always be retried. Previously this was treated as a terminal error, reported to Bugsnag, and the stream gave up. Now it continues the retry loop, bounded by maxReconnectAttempts. Bugsnag: 6a3686f3d2c3b94118b56442 Signed-off-by: Brandon McAnsh <git@bmcreations.dev> * fix(tokens,accounts): apply same network-state debounce fix AccountController and TokenCoordinator had the same bug as ContactCoordinator — subscribing to the full NetworkState instead of just the connected boolean, causing redundant work on signal-strength or connection-type changes. AccountController was worse: no distinctUntilChanged at all, so every NetworkState emission triggered fetchAdditionalAccountInfo(). Bugsnag: 6a3686f3d2c3b94118b56442 Signed-off-by: Brandon McAnsh <git@bmcreations.dev> --------- Signed-off-by: Brandon McAnsh <git@bmcreations.dev>
1 parent 1b0a7f6 commit 779eda2

7 files changed

Lines changed: 316 additions & 12 deletions

File tree

‎apps/flipcash/shared/contacts/src/main/kotlin/com/flipcash/app/contacts/ContactCoordinator.kt‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -143,9 +143,9 @@ class ContactCoordinator @Inject constructor(
143143
.launchIn(scope)
144144

145145
cluster.filterNotNull()
146-
.flatMapLatest { networkObserver.state }
146+
.flatMapLatest { networkObserver.state.map { it.connected } }
147147
.distinctUntilChanged()
148-
.filter { it.connected }
148+
.filter { it }
149149
.onEach {
150150
trace(tag = TAG, message = "Network connected, triggering contact sync", type = TraceType.Process)
151151
launchSync()

‎apps/flipcash/shared/contacts/src/test/kotlin/com/flipcash/app/contacts/ContactCoordinatorNetworkSyncTest.kt‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,81 @@
1+
@file:OptIn(ExperimentalCoroutinesApi::class)
2+
3+
package com.flipcash.app.contacts
4+
5+
import com.getcode.utils.network.ConnectionType
6+
import com.getcode.utils.network.NetworkConnectivityListener
7+
import com.getcode.utils.network.NetworkState
8+
import com.getcode.utils.network.SignalStrength
9+
import kotlinx.coroutines.ExperimentalCoroutinesApi
10+
import kotlinx.coroutines.flow.MutableStateFlow
11+
import kotlinx.coroutines.flow.StateFlow
12+
import kotlinx.coroutines.flow.asStateFlow
13+
import kotlinx.coroutines.flow.distinctUntilChanged
14+
import kotlinx.coroutines.flow.filter
15+
import kotlinx.coroutines.flow.filterNotNull
16+
import kotlinx.coroutines.flow.flatMapLatest
17+
import kotlinx.coroutines.flow.launchIn
18+
import kotlinx.coroutines.flow.map
19+
import kotlinx.coroutines.flow.onEach
20+
import kotlinx.coroutines.test.advanceUntilIdle
21+
import kotlinx.coroutines.test.runTest
22+
import org.junit.Test
23+
import kotlin.test.assertEquals
24+
25+
class ContactCoordinatorNetworkSyncTest {
26+
27+
private class FakeNetworkObserver : NetworkConnectivityListener {
28+
val _state = MutableStateFlow(NetworkState.Default)
29+
override val state: StateFlow<NetworkState> = _state.asStateFlow()
30+
override val isConnected: Boolean get() = _state.value.connected
31+
override val type: ConnectionType get() = _state.value.type
32+
}
33+
34+
@Test
35+
fun `signal strength changes while connected do not re-trigger sync`() = runTest {
36+
val networkObserver = FakeNetworkObserver()
37+
var syncCount = 0
38+
39+
val cluster = MutableStateFlow<Unit?>(null)
40+
41+
// Mirrors the FIXED subscription from ContactCoordinator.init
42+
val job = cluster
43+
.filterNotNull()
44+
.flatMapLatest {
45+
networkObserver.state
46+
.map { it.connected }
47+
.distinctUntilChanged()
48+
}
49+
.filter { it }
50+
.onEach { syncCount++ }
51+
.launchIn(this)
52+
53+
cluster.value = Unit
54+
advanceUntilIdle()
55+
56+
// Initial connected
57+
networkObserver._state.value = NetworkState(true, SignalStrength.Good, ConnectionType.Wifi)
58+
advanceUntilIdle()
59+
assertEquals(1, syncCount, "First connected event should trigger sync")
60+
61+
// Rapid signal/type changes — all still connected
62+
networkObserver._state.value = NetworkState(true, SignalStrength.Great, ConnectionType.Wifi)
63+
networkObserver._state.value = NetworkState(true, SignalStrength.Strong, ConnectionType.Wifi)
64+
networkObserver._state.value = NetworkState(true, SignalStrength.Good, ConnectionType.Cellular)
65+
networkObserver._state.value = NetworkState(true, SignalStrength.Poor, ConnectionType.Cellular)
66+
networkObserver._state.value = NetworkState(true, SignalStrength.Great, ConnectionType.Wifi)
67+
advanceUntilIdle()
68+
69+
assertEquals(1, syncCount, "Signal strength / type changes should NOT re-trigger sync")
70+
71+
// Disconnect → reconnect
72+
networkObserver._state.value = NetworkState(false, SignalStrength.Unknown, ConnectionType.Unknown)
73+
advanceUntilIdle()
74+
networkObserver._state.value = NetworkState(true, SignalStrength.Good, ConnectionType.Wifi)
75+
advanceUntilIdle()
76+
77+
assertEquals(2, syncCount, "Reconnect after disconnect should trigger sync")
78+
79+
job.cancel()
80+
}
81+
}

‎apps/flipcash/shared/tokens/src/main/kotlin/com/flipcash/app/tokens/TokenCoordinator.kt‎

Lines changed: 18 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -151,9 +151,9 @@ class TokenCoordinator @Inject constructor(
151151
ProcessLifecycleOwner.get().lifecycle.addObserver(this)
152152

153153
cluster.filterNotNull()
154-
.flatMapLatest { networkObserver.state }
154+
.flatMapLatest { networkObserver.state.map { it.connected } }
155155
.distinctUntilChanged()
156-
.filter { it.connected }
156+
.filter { it }
157157
.onEach {
158158
trace(tag = TAG, message = "Network connected, triggering token update", type = TraceType.Process)
159159
retryable { update() }
@@ -221,15 +221,30 @@ class TokenCoordinator @Inject constructor(
221221
modifyBalance(token, amount) { current, delta -> current + delta }
222222
}
223223

224+
suspend fun add(mint: Mint, nativeAmount: Fiat) {
225+
val token = getTokenMetadata(mint).getOrNull()?.token ?: return
226+
add(token, nativeAmount.toLocalFiat(mint))
227+
}
228+
224229
suspend fun subtract(token: Token, fiat: LocalFiat) {
225230
val rate = exchange.rateToUsd(fiat.rate.currency)
226231
val amount = rate?.let { fiat.nativeAmount.convertingTo(it) }
227232
if (amount != null) {
228-
trace(tag = TAG, message = "Subtracting ${amount.formatted()} to ${token.symbol}", type = TraceType.Process)
233+
trace(tag = TAG, message = "Subtracting ${amount.formatted()} from ${token.symbol}", type = TraceType.Process)
229234
}
230235
modifyBalance(token, amount) { current, delta -> current - delta }
231236
}
232237

238+
suspend fun subtract(mint: Mint, nativeAmount: Fiat) {
239+
val token = getTokenMetadata(mint).getOrNull()?.token ?: return
240+
subtract(token, nativeAmount.toLocalFiat(mint))
241+
}
242+
243+
private fun Fiat.toLocalFiat(mint: Mint): LocalFiat {
244+
val rate = exchange.rateFor(currencyCode) ?: Rate.oneToOne
245+
return LocalFiat.fromNativeAmount(this, rate, mint)
246+
}
247+
233248
// endregion
234249

235250
// region Public API — Token Metadata (implements TokenMetadataProvider)

‎scripts/stress-test-event-stream.sh‎

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
#!/bin/bash
2+
# Event Stream Stress Test
3+
#
4+
# Prerequisites: adb connected to device, app installed
5+
6+
set -euo pipefail
7+
8+
LABEL="${1:-test}"
9+
PKG="com.flipcash.app.android"
10+
LOGFILE="event-stream-${LABEL}-$(date +%Y%m%d-%H%M%S).log"
11+
TAGS="gRPC:V BIDI:V ChatCoordinator:V EventStream:V event-streaming:V EventStreamingController:V *:S"
12+
13+
echo "=== Event Stream Stress Test ($LABEL) ==="
14+
echo "Logging to: $LOGFILE"
15+
echo ""
16+
17+
# Clear logcat
18+
adb logcat -c
19+
20+
# Start logcat capture in background
21+
adb logcat $TAGS | tee "$LOGFILE" &
22+
LOGCAT_PID=$!
23+
trap "kill $LOGCAT_PID 2>/dev/null; echo; echo 'Logs saved to $LOGFILE'" EXIT
24+
25+
pause() {
26+
echo ""
27+
echo ">>> $1"
28+
echo " Press ENTER when ready..."
29+
read -r
30+
}
31+
32+
wait_and_log() {
33+
local duration=$1
34+
local label=$2
35+
echo " [$label] Waiting ${duration}s..."
36+
sleep "$duration"
37+
echo " [$label] Done."
38+
}
39+
40+
echo "--- Test 1: Cold Start ---"
41+
echo " Force stopping app..."
42+
adb shell am force-stop "$PKG"
43+
sleep 2
44+
echo " Launching app..."
45+
adb shell am start -n "$PKG/.MainActivity" -a android.intent.action.MAIN -c android.intent.category.LAUNCHER
46+
wait_and_log 15 "cold-start"
47+
echo " CHECK: Look for 'flipcash-stream => READY' and 'Stream activated on first response'"
48+
echo ""
49+
50+
echo "--- Test 2: Airplane Mode Toggle (network loss/recovery) ---"
51+
echo " Enabling airplane mode..."
52+
adb shell cmd connectivity airplane-mode enable
53+
wait_and_log 5 "airplane-on"
54+
echo " Disabling airplane mode..."
55+
adb shell cmd connectivity airplane-mode disable
56+
wait_and_log 20 "airplane-off"
57+
echo " CHECK: Stream should reconnect. Look for 'Opening bidirectional stream' then 'Stream activated'"
58+
echo ""
59+
60+
echo "--- Test 3: Rapid Network Flapping (3 cycles) ---"
61+
for i in 1 2 3; do
62+
echo " Cycle $i: airplane ON..."
63+
adb shell cmd connectivity airplane-mode enable
64+
sleep 3
65+
echo " Cycle $i: airplane OFF..."
66+
adb shell cmd connectivity airplane-mode disable
67+
sleep 8
68+
done
69+
wait_and_log 15 "flap-settle"
70+
echo " CHECK: Should settle to one active stream, no zombie states"
71+
echo ""
72+
73+
echo "--- Test 4: Background/Foreground Cycling ---"
74+
echo " Sending to background..."
75+
adb shell input keyevent KEYCODE_HOME
76+
wait_and_log 5 "background"
77+
echo " Bringing to foreground..."
78+
adb shell am start -n "$PKG/.MainActivity" -a android.intent.action.MAIN -c android.intent.category.LAUNCHER
79+
wait_and_log 10 "foreground"
80+
echo " Repeat..."
81+
adb shell input keyevent KEYCODE_HOME
82+
sleep 3
83+
adb shell am start -n "$PKG/.MainActivity" -a android.intent.action.MAIN -c android.intent.category.LAUNCHER
84+
wait_and_log 10 "foreground-2"
85+
echo " CHECK: Stream should close on background, reopen on foreground"
86+
echo ""
87+
88+
echo "--- Test 5: Long Idle (simulates carrier NAT timeout) ---"
89+
echo " Waiting 90s with app in foreground (covers typical 30-90s NAT timeout)..."
90+
wait_and_log 90 "long-idle"
91+
echo " CHECK: Heartbeat should detect dead stream and reconnect if needed"
92+
echo ""
93+
94+
echo "--- Test 6: Wi-Fi Toggle (different network path) ---"
95+
echo " Disabling Wi-Fi..."
96+
adb shell svc wifi disable
97+
wait_and_log 10 "wifi-off"
98+
echo " Enabling Wi-Fi..."
99+
adb shell svc wifi enable
100+
wait_and_log 15 "wifi-on"
101+
echo " CHECK: Stream reconnects on new network"
102+
echo ""
103+
104+
echo ""
105+
echo "=== Stress test complete ==="
106+
echo ""
107+
echo "Review logs: less $LOGFILE"
108+
echo ""
109+
echo "Key things to grep for:"
110+
echo " grep 'flipcash-stream =>' $LOGFILE # channel state changes"
111+
echo " grep 'activateStream' $LOGFILE # when stream becomes 'connected'"
112+
echo " grep 'Stream active:' $LOGFILE # isActive state changes"
113+
echo " grep 'zombie\\|Giving up' $LOGFILE # failure modes (should be absent)"
114+
echo " grep 'Event stream error' $LOGFILE # retry triggers"
115+
echo " grep 'Pong' $LOGFILE # successful ping/pong = healthy stream"

‎services/opencode/src/main/kotlin/com/getcode/opencode/controllers/AccountController.kt‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import kotlinx.coroutines.Dispatchers
2020
import kotlinx.coroutines.SupervisorJob
2121
import kotlinx.coroutines.flow.Flow
2222
import kotlinx.coroutines.flow.MutableStateFlow
23+
import kotlinx.coroutines.flow.filter
2324
import kotlinx.coroutines.flow.filterNotNull
2425
import kotlinx.coroutines.flow.flatMapLatest
2526
import kotlinx.coroutines.flow.launchIn
@@ -65,12 +66,11 @@ class AccountController @Inject constructor(
6566

6667
init {
6768
cluster.filterNotNull()
68-
.flatMapLatest { networkObserver.state }
69-
.map { it.connected }
70-
.onEach { connected ->
71-
if (connected) {
72-
retryable { fetchAdditionalAccountInfo() }
73-
}
69+
.flatMapLatest { networkObserver.state.map { it.connected } }
70+
.distinctUntilChanged()
71+
.filter { it }
72+
.onEach {
73+
retryable { fetchAdditionalAccountInfo() }
7474
}.launchIn(scope)
7575
}
7676

‎services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import io.grpc.Status
66
import io.grpc.StatusException
77
import io.grpc.StatusRuntimeException
88
import kotlinx.coroutines.channels.Channel
9+
import kotlinx.coroutines.channels.ClosedSendChannelException
910
import kotlinx.coroutines.delay
1011
import kotlinx.coroutines.flow.Flow
1112
import kotlinx.coroutines.flow.catch
@@ -129,7 +130,7 @@ fun <Request, Response, StreamRef> openBidirectionalStream(
129130
)
130131
collectionJob.cancel()
131132
requestChannel.close()
132-
if (isRetryable(e)) {
133+
if (isRetryable(e) || e is ClosedSendChannelException) {
133134
onReconnectAttempt?.invoke(attempt, e)
134135
continue
135136
}

‎services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
@file:OptIn(ExperimentalCoroutinesApi::class)
2+
3+
package com.getcode.opencode.internal.bidi
4+
5+
import kotlinx.coroutines.ExperimentalCoroutinesApi
6+
import kotlinx.coroutines.channels.ClosedSendChannelException
7+
import kotlinx.coroutines.flow.flow
8+
import kotlinx.coroutines.test.advanceUntilIdle
9+
import kotlinx.coroutines.test.runTest
10+
import org.junit.Test
11+
import kotlin.test.assertEquals
12+
import kotlin.test.assertTrue
13+
14+
class OpenBidirectionalStreamTest {
15+
16+
/**
17+
* ClosedSendChannelException thrown during the initial send should be
18+
* treated as retryable. We throw on every attempt and set maxReconnectAttempts=2
19+
* so the loop exhausts retries and the coroutine terminates cleanly.
20+
*
21+
* Without the fix, the first ClosedSendChannelException would call onError
22+
* and return immediately (attemptCount == 1). With the fix, it retries
23+
* until maxReconnectAttempts is exceeded (attemptCount == 3).
24+
*/
25+
@Test
26+
fun `ClosedSendChannelException on initial send triggers retry`() = runTest {
27+
var attemptCount = 0
28+
val errors = mutableListOf<Throwable>()
29+
30+
val streamRef = BidirectionalStreamReference<String, String>(this, "test-stream")
31+
streamRef.retain()
32+
33+
openBidirectionalStream<String, String, BidirectionalStreamReference<String, String>>(
34+
streamRef = streamRef,
35+
apiCall = { _ ->
36+
attemptCount++
37+
flow { }
38+
},
39+
initialRequest = {
40+
// Always throw — we're testing that the retry loop continues
41+
throw ClosedSendChannelException("Channel was closed")
42+
},
43+
responseHandler = { _: String, _: (String) -> Unit -> },
44+
onError = { errors.add(it) },
45+
maxReconnectAttempts = 2,
46+
reconnectDelayMs = 0,
47+
)
48+
49+
advanceUntilIdle()
50+
51+
// With fix: retries until max attempts exceeded (3 attempts for maxReconnectAttempts=2)
52+
// Without fix: gives up on first attempt (attemptCount == 1)
53+
assertTrue(
54+
attemptCount > 1,
55+
"Should have retried after ClosedSendChannelException, but only $attemptCount attempt(s)"
56+
)
57+
// The terminal error should be the max-attempts IllegalStateException, not ClosedSendChannelException
58+
assertTrue(
59+
errors.none { it is ClosedSendChannelException },
60+
"ClosedSendChannelException should not be reported as a terminal error"
61+
)
62+
63+
streamRef.destroy()
64+
}
65+
66+
@Test
67+
fun `non-retryable exceptions on initial send still propagate to onError`() = runTest {
68+
val errors = mutableListOf<Throwable>()
69+
70+
val streamRef = BidirectionalStreamReference<String, String>(this, "test-stream")
71+
streamRef.retain()
72+
73+
openBidirectionalStream<String, String, BidirectionalStreamReference<String, String>>(
74+
streamRef = streamRef,
75+
apiCall = { _ ->
76+
throw IllegalArgumentException("Bad request format")
77+
},
78+
initialRequest = { "hello" },
79+
responseHandler = { _: String, _: (String) -> Unit -> },
80+
onError = { errors.add(it) },
81+
maxReconnectAttempts = 3,
82+
reconnectDelayMs = 0,
83+
)
84+
85+
advanceUntilIdle()
86+
87+
assertEquals(1, errors.size, "Non-retryable error should be reported exactly once")
88+
assertTrue(errors[0] is IllegalArgumentException)
89+
90+
streamRef.destroy()
91+
}
92+
}

0 commit comments

Comments
 (0)