Skip to content

Commit 8a6f83a

Browse files
bmc08gtbmc08gt
committed
fix(services/bidi): improve BIDI stream resilience and event stream recovery
Rewrite openBidirectionalStream with a single coordinator loop replacing the recursive approach that spawned competing reconnect chains. Fresh Channel + consumeAsFlow per attempt avoids the single-use crash. Handle both StatusException and StatusRuntimeException via grpcStatusCode() helper. Add reconnectOnAborted support to the framework (off by default). Event stream: timeout handler now destroys ref and signals error so the heartbeat can reopen cleanly. onError clears streamRef to unblock reopens. Standardize trace tags to "event-stream" with formatted server timestamps. Signed-off-by: Brandon McAnsh <git@bmcreations.dev>
1 parent 9be2716 commit 8a6f83a

8 files changed

Lines changed: 199 additions & 197 deletions

File tree

‎apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt‎

Lines changed: 9 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,6 @@ import kotlinx.coroutines.launch
6565
import javax.inject.Inject
6666
import javax.inject.Singleton
6767
import kotlin.time.Clock
68-
import kotlin.time.Duration.Companion.milliseconds
6968
import kotlin.time.Duration.Companion.seconds
7069

7170
@Singleton
@@ -86,8 +85,6 @@ class ChatCoordinator @Inject constructor(
8685
companion object {
8786
private const val TAG = "ChatCoordinator"
8887
private val HEARTBEAT_INTERVAL = 30.seconds
89-
private const val RETRY_BASE_MS = 2_000L
90-
private const val RETRY_MAX_MS = 60_000L
9188
}
9289

9390
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
@@ -96,10 +93,8 @@ class ChatCoordinator @Inject constructor(
9693
private var syncJob: Job? = null
9794
private var flagObserverJob: Job? = null
9895
private var eventStreamCollectJob: Job? = null
99-
private var eventStreamRetryJob: Job? = null
10096
private var feedObserverJob: Job? = null
10197
private var heartbeatJob: Job? = null
102-
private var retryAttempt = 0
10398
private var backgroundedActiveChat: ChatId? = null
10499

105100
val state: StateFlow<ChatState>
@@ -138,11 +133,9 @@ class ChatCoordinator @Inject constructor(
138133
trace(tag = TAG, message = "User logged in, hydrating chat", type = TraceType.User)
139134
this.cluster.value = cluster
140135
observeFeedFromDb()
141-
if (isChatEnabled()) {
142-
syncFeed()
143-
openEventStream()
144-
startHeartbeat()
145-
}
136+
syncFeed()
137+
openEventStream()
138+
startHeartbeat()
146139
observeFeatureFlag()
147140
}
148141

@@ -437,30 +430,14 @@ class ChatCoordinator @Inject constructor(
437430
}
438431
}
439432

440-
private fun openEventStream(force: Boolean = false) {
441-
if (!force && eventStreamingController.isConnected) {
433+
private fun openEventStream() {
434+
if (eventStreamingController.isConnected) {
442435
trace(tag = TAG, message = "Event stream already connected, skipping open", type = TraceType.Process)
443436
ensureCollector()
444437
return
445438
}
446439

447-
eventStreamRetryJob?.cancel()
448-
val opened = eventStreamingController.open(scope) {
449-
// Stream died — schedule a re-open with exponential backoff
450-
val attempt = retryAttempt++
451-
val delayMs = (RETRY_BASE_MS * (1L shl attempt.coerceAtMost(5)))
452-
.coerceAtMost(RETRY_MAX_MS)
453-
trace(tag = TAG, message = "Event stream error, retry #${attempt + 1} in ${delayMs}ms", type = TraceType.Process)
454-
eventStreamRetryJob = scope.launch {
455-
delay(delayMs.milliseconds)
456-
openEventStream(force = true)
457-
}
458-
}
459-
if (opened) {
460-
retryAttempt = 0
461-
} else {
462-
trace(tag = TAG, message = "Event stream failed to open", type = TraceType.Error)
463-
}
440+
eventStreamingController.open(scope)
464441
ensureCollector()
465442
}
466443

@@ -473,8 +450,6 @@ class ChatCoordinator @Inject constructor(
473450
}
474451

475452
private fun closeEventStream() {
476-
eventStreamRetryJob?.cancel()
477-
eventStreamRetryJob = null
478453
eventStreamCollectJob?.cancel()
479454
eventStreamCollectJob = null
480455
eventStreamingController.close()
@@ -485,9 +460,11 @@ class ChatCoordinator @Inject constructor(
485460
heartbeatJob = scope.launch {
486461
while (true) {
487462
delay(HEARTBEAT_INTERVAL)
488-
if (!eventStreamingController.isConnected) {
463+
if (!eventStreamingController.isStreamActive) {
489464
trace(tag = TAG, message = "Heartbeat: event stream dead, syncing feed and reconnecting", type = TraceType.Process)
490465
syncFeed()
466+
// Close the dead ref so open() creates a fresh one
467+
eventStreamingController.close()
491468
openEventStream()
492469
}
493470
}

‎services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt‎

Lines changed: 13 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -22,19 +22,21 @@ class EventStreamingController @Inject constructor(
2222

2323
private var streamRef: EventStreamReference? = null
2424

25-
val isConnected: Boolean get() = streamRef?.isActive == true
25+
val isConnected: Boolean get() = streamRef != null
26+
27+
val isStreamActive: Boolean get() = streamRef?.isActive == true
28+
29+
fun open(scope: CoroutineScope): Boolean {
30+
if (streamRef != null) {
31+
trace("EventStreamingController: Stream already open, skipping")
32+
return true
33+
}
2634

27-
fun open(
28-
scope: CoroutineScope,
29-
onStreamError: (() -> Unit)? = null,
30-
): Boolean {
3135
val owner = userManager.accountCluster?.authority?.keyPair ?: run {
3236
trace("EventStreamingController: No account cluster, cannot open stream")
3337
return false
3438
}
3539

36-
close()
37-
3840
streamRef = repository.openEventStream(
3941
scope = scope,
4042
owner = owner,
@@ -44,7 +46,10 @@ class EventStreamingController @Inject constructor(
4446
},
4547
onError = { error ->
4648
trace("EventStreamingController: Stream error: ${error.message}")
47-
onStreamError?.invoke()
49+
// Clear the ref so the next heartbeat, lifecycle, or network
50+
// event creates a fresh stream. The framework guarantees this
51+
// fires only once per stream, so no risk of clearing a newer ref.
52+
streamRef = null
4853
},
4954
)
5055
return true

‎services/flipcash/src/main/kotlin/com/flipcash/services/inject/FlipcashModule.kt‎

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package com.flipcash.services.inject
22

33
import android.content.Context
44
import com.flipcash.services.internal.annotations.FlipcashManagedChannel
5+
import com.flipcash.services.internal.annotations.FlipcashManagedStreamingChannel
56
import com.flipcash.services.internal.annotations.FlipcashProtocol
67
import com.flipcash.services.internal.domain.ActivityFeedMessageMapper
78
import com.flipcash.services.internal.domain.ImageModerationResponseMapper
@@ -97,6 +98,25 @@ internal object FlipcashModule {
9798
@ApplicationContext context: Context,
9899
@FlipcashProtocol
99100
config: ProtocolConfig,
101+
): ManagedChannel {
102+
return AndroidChannelBuilder
103+
.usingBuilder(OkHttpChannelBuilder.forAddress(config.baseUrl, config.port))
104+
.context(context)
105+
.userAgent(config.userAgent)
106+
.keepAliveWithoutCalls(false)
107+
.idleTimeout(5, TimeUnit.MINUTES)
108+
.intercept(LoggingClientInterceptor())
109+
.build()
110+
.also { observeChannelState("flipcash", it) }
111+
}
112+
113+
@Singleton
114+
@Provides
115+
@FlipcashManagedStreamingChannel
116+
fun provideManagedStreamingChannel(
117+
@ApplicationContext context: Context,
118+
@FlipcashProtocol
119+
config: ProtocolConfig,
100120
): ManagedChannel {
101121
return AndroidChannelBuilder
102122
.usingBuilder(OkHttpChannelBuilder.forAddress(config.baseUrl, config.port))
@@ -107,7 +127,7 @@ internal object FlipcashModule {
107127
.keepAliveWithoutCalls(true)
108128
.intercept(LoggingClientInterceptor())
109129
.build()
110-
.also { observeChannelState("flipcash", it) }
130+
.also { observeChannelState("flipcash-stream", it) }
111131
}
112132

113133
private fun observeChannelState(name: String, channel: ManagedChannel) {

‎services/flipcash/src/main/kotlin/com/flipcash/services/internal/annotations/FlipcashManagedChannel.kt‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,4 +10,14 @@ import javax.inject.Qualifier
1010
AnnotationTarget.VALUE_PARAMETER,
1111
AnnotationTarget.FIELD
1212
)
13-
annotation class FlipcashManagedChannel
13+
annotation class FlipcashManagedChannel
14+
15+
@Qualifier
16+
@Target(
17+
AnnotationTarget.FUNCTION,
18+
AnnotationTarget.PROPERTY_GETTER,
19+
AnnotationTarget.PROPERTY_SETTER,
20+
AnnotationTarget.VALUE_PARAMETER,
21+
AnnotationTarget.FIELD
22+
)
23+
annotation class FlipcashManagedStreamingChannel

‎services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/api/EventStreamingApi.kt‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ package com.flipcash.services.internal.network.api
22

33
import com.codeinc.flipcash.gen.events.v1.EventStreamingGrpcKt
44
import com.codeinc.flipcash.gen.events.v1.EventStreamingService
5-
import com.flipcash.services.internal.annotations.FlipcashManagedChannel
5+
import com.flipcash.services.internal.annotations.FlipcashManagedStreamingChannel
66
import com.getcode.opencode.internal.network.core.GrpcApi
77
import io.grpc.ManagedChannel
88
import kotlinx.coroutines.flow.Flow
@@ -11,12 +11,11 @@ import javax.inject.Singleton
1111

1212
@Singleton
1313
internal class EventStreamingApi @Inject constructor(
14-
@FlipcashManagedChannel
14+
@FlipcashManagedStreamingChannel
1515
managedChannel: ManagedChannel,
1616
) : GrpcApi(managedChannel) {
1717

1818
private val api = EventStreamingGrpcKt.EventStreamingCoroutineStub(managedChannel)
19-
.withWaitForReady()
2019

2120
fun streamEvents(
2221
requests: Flow<EventStreamingService.StreamEventsRequest>,

‎services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/services/EventStreamingService.kt‎

Lines changed: 13 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import com.getcode.utils.trace
1515
import com.google.protobuf.Timestamp
1616
import kotlin.time.Clock
1717
import kotlinx.coroutines.CoroutineScope
18+
import kotlinx.datetime.Instant
1819
import kotlinx.coroutines.launch
1920
import javax.inject.Inject
2021

@@ -29,15 +30,14 @@ internal class EventStreamingService @Inject constructor(
2930
onEvent: (ChatUpdate) -> Unit,
3031
onError: (Throwable) -> Unit = {},
3132
): EventStreamReference {
32-
trace("EventStream Opening stream.")
33+
trace(tag = "event-stream", message = "Opening stream.")
3334
val streamReference = EventStreamReference(scope, "event-streaming")
3435

3536
streamReference.retain()
3637
streamReference.timeoutHandler = {
37-
trace("EventStream timed out")
38-
streamReference.coroutineScope.launch {
39-
openStream(scope, owner, streamReference, onEvent, onError)
40-
}
38+
trace(tag = "event-stream", message = "Timed out, signaling error for reconnect")
39+
streamReference.destroy()
40+
onError(IllegalStateException("Event stream timed out"))
4141
}
4242

4343
streamReference.coroutineScope.launch {
@@ -82,7 +82,8 @@ internal class EventStreamingService @Inject constructor(
8282

8383
streamRef.receivedPing(updatedTimeout = response.ping.pingDelay.seconds * 1_000L)
8484
sendRequest(pong)
85-
trace(message = "EventStream Pong. Server timestamp: ${response.ping.timestamp}")
85+
val serverTime = Instant.fromEpochSeconds(response.ping.timestamp.seconds, response.ping.timestamp.nanos)
86+
trace(tag = "event-stream", message = "Pong. Server timestamp: $serverTime")
8687
}
8788

8889
RpcEventStreamingService.StreamEventsResponse.TypeCase.EVENTS -> {
@@ -93,7 +94,8 @@ internal class EventStreamingService @Inject constructor(
9394
}
9495
else -> {
9596
trace(
96-
message = "EventStream received unhandled event type: ${event.typeCase}",
97+
tag = "event-stream",
98+
message = "Received unhandled event type: ${event.typeCase}",
9799
type = TraceType.Log,
98100
)
99101
}
@@ -111,14 +113,16 @@ internal class EventStreamingService @Inject constructor(
111113
}
112114
onError(error)
113115
trace(
114-
message = "EventStream error: ${response.error.code}",
116+
tag = "event-stream",
117+
message = "Error: ${response.error.code}",
115118
type = TraceType.Error,
116119
)
117120
}
118121

119122
else -> {
120123
trace(
121-
message = "EventStream received empty message.",
124+
tag = "event-stream",
125+
message = "Received empty message.",
122126
type = TraceType.Error,
123127
)
124128
}

0 commit comments

Comments
 (0)