From a9f4c7df6f6f0c07df828e1687019008835e869c Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Wed, 1 Jul 2026 10:39:34 -0400 Subject: [PATCH 1/2] chore(logging): report all errors to Bugsnag, downgrade non-notifiable severity Previously only notifiable errors reached Bugsnag; everything else was dropped. Now all non-noise errors are reported, with severity derived from isNotifiable: WARNING for notifiable (needs attention), INFO for non-notifiable (reference only). Crashes remain ERROR. Slack routing should filter on severity != INFO to preserve today's notifiable-only behavior. Signed-off-by: Brandon McAnsh --- .../app/internal/startup/BugsnagErrorReporter.kt | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/apps/flipcash/app/src/main/kotlin/com/flipcash/app/internal/startup/BugsnagErrorReporter.kt b/apps/flipcash/app/src/main/kotlin/com/flipcash/app/internal/startup/BugsnagErrorReporter.kt index d2a4e6432..e45e57f2a 100644 --- a/apps/flipcash/app/src/main/kotlin/com/flipcash/app/internal/startup/BugsnagErrorReporter.kt +++ b/apps/flipcash/app/src/main/kotlin/com/flipcash/app/internal/startup/BugsnagErrorReporter.kt @@ -1,12 +1,17 @@ package com.flipcash.app.internal.startup import com.bugsnag.android.Bugsnag +import com.bugsnag.android.Severity import com.getcode.utils.ErrorReporter class BugsnagErrorReporter : ErrorReporter { override fun report(error: Throwable, cause: Throwable, isNotifiable: Boolean) { - if (!isNotifiable) return if (!Bugsnag.isStarted()) return - Bugsnag.notify(error) + Bugsnag.notify(error) { event -> + // WARNING = needs active attention (also drives the Slack filter); + // INFO = recorded for reference only, excluded from Slack. + event.severity = if (isNotifiable) Severity.WARNING else Severity.INFO + true + } } } From 1529f6ad9c892a7692c334458d46be402023db59 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Wed, 1 Jul 2026 12:53:18 -0400 Subject: [PATCH 2/2] fix(services/bidi): gate reconnect reset on stream health, add exponential backoff Signed-off-by: Brandon McAnsh --- .../network/services/EventStreamingService.kt | 1 + .../opencode/internal/bidi/OpenStream.kt | 43 ++++++- .../bidi/OpenBidirectionalStreamTest.kt | 118 +++++++++++++++++- 3 files changed, 155 insertions(+), 7 deletions(-) diff --git a/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/services/EventStreamingService.kt b/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/services/EventStreamingService.kt index ca75a5158..61718a45f 100644 --- a/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/services/EventStreamingService.kt +++ b/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/services/EventStreamingService.kt @@ -73,6 +73,7 @@ internal class EventStreamingService @Inject constructor( reconnectOnCancelled = true, reconnectOnAborted = true, reconnectDelayMs = 1_000L, + maxReconnectDelayMs = 30_000L, onError = { onError(it) }, responseHandler = { response, sendRequest -> when (response.typeCase) { diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt index 147a562e2..fcd2beca4 100644 --- a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt +++ b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt @@ -13,6 +13,10 @@ import kotlinx.coroutines.flow.catch import kotlinx.coroutines.flow.consumeAsFlow import kotlinx.coroutines.launch import kotlin.coroutines.cancellation.CancellationException +import kotlin.time.Duration +import kotlin.time.Duration.Companion.seconds +import kotlin.time.TimeMark +import kotlin.time.TimeSource fun openBidirectionalStream( streamRef: StreamRef, @@ -27,7 +31,13 @@ fun openBidirectionalStream( reconnectOnCancelled: Boolean = false, reconnectOnAborted: Boolean = false, reconnectDelayMs: Long = 300L, + maxReconnectDelayMs: Long = 30_000L, maxReconnectAttempts: Int = 8, + // Minimum time a stream must stay connected (past first response) before we + // treat it as healthy and reset the backoff counter. Prevents an + // activate-then-immediately-fail stream from resetting forever. + healthyThreshold: Duration = 30.seconds, + timeSource: TimeSource = TimeSource.Monotonic, onReconnectAttempt: ((attempt: Int, reason: Throwable?) -> Unit)? = null ) where StreamRef : BidirectionalStreamReference<*, *> { @@ -55,8 +65,9 @@ fun openBidirectionalStream( var attempt = 0 while (attempt++ <= maxReconnectAttempts) { - if (attempt > 1) { - delay(reconnectDelayMs) + val backoffMs = computeBackoffMs(attempt, reconnectDelayMs, maxReconnectDelayMs) + if (backoffMs > 0) { + delay(backoffMs) } trace(tag = tag, message = "Opening bidirectional stream (attempt $attempt)") @@ -84,6 +95,7 @@ fun openBidirectionalStream( } var activated = false + var activationMark: TimeMark? = null val collectionJob = launch { responseFlow @@ -96,6 +108,7 @@ fun openBidirectionalStream( .collect { response -> if (!activated) { activated = true + activationMark = timeSource.markNow() streamRef.activateStream() trace(tag = tag, message = "Stream activated on first response") } @@ -154,14 +167,21 @@ fun openBidirectionalStream( val code = error.grpcStatusCode() if (isRetryable(error)) { + val healthy = activationMark?.elapsedNow()?.let { it >= healthyThreshold } ?: false trace( tag = tag, - message = "Stream failed (${code?.name ?: error.javaClass.simpleName}) → reconnecting" + message = "Stream failed (${code?.name ?: error.javaClass.simpleName}) " + + "→ reconnecting (attempt $attempt, healthy=$healthy)" ) onReconnectAttempt?.invoke(attempt, error) - // Reset attempt counter on successful activation — only count - // consecutive failures, not total lifetime attempts. - if (activated) attempt = 0 + // Reset the backoff counter ONLY when the stream was genuinely + // healthy (stayed connected past healthyThreshold). A stream that + // activates then immediately fails (e.g. ABORTED "stream already + // exists") must count toward maxReconnectAttempts so the loop + // terminates instead of hammering the server. + // Reset to 0: after a healthy stream dies, reconnect immediately + // (computeBackoffMs returns 0 for attempt=1). + if (healthy) attempt = 0 continue } @@ -189,3 +209,14 @@ internal fun Throwable.grpcStatusCode(): Status.Code? = when (this) { is StatusException -> status.code else -> null } + +/** + * Exponential backoff with a hard cap. Returns 0 for the first attempt (no delay) + * and for a zero base delay (used in tests). Doubles per consecutive failed attempt. + */ +internal fun computeBackoffMs(attempt: Int, baseDelayMs: Long, maxDelayMs: Long): Long { + if (attempt <= 1 || baseDelayMs <= 0) return 0L + val exponent = (attempt - 2).coerceIn(0, 16) + val scaled = baseDelayMs shl exponent + return scaled.coerceAtMost(maxDelayMs) +} diff --git a/services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt b/services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt index d29efe8a6..f57319503 100644 --- a/services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt +++ b/services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt @@ -1,15 +1,19 @@ -@file:OptIn(ExperimentalCoroutinesApi::class) +@file:OptIn(ExperimentalCoroutinesApi::class, kotlin.time.ExperimentalTime::class) package com.getcode.opencode.internal.bidi +import io.grpc.Status import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.channels.ClosedSendChannelException +import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.flow import kotlinx.coroutines.test.advanceUntilIdle import kotlinx.coroutines.test.runTest import org.junit.Test import kotlin.test.assertEquals import kotlin.test.assertTrue +import kotlin.time.Duration.Companion.seconds +import kotlin.time.TestTimeSource class OpenBidirectionalStreamTest { @@ -89,4 +93,116 @@ class OpenBidirectionalStreamTest { streamRef.destroy() } + + @Test + fun `computeBackoffMs grows exponentially and caps at max`() { + assertEquals(0L, computeBackoffMs(attempt = 1, baseDelayMs = 1000, maxDelayMs = 30_000)) + assertEquals(1000L, computeBackoffMs(attempt = 2, baseDelayMs = 1000, maxDelayMs = 30_000)) + assertEquals(2000L, computeBackoffMs(attempt = 3, baseDelayMs = 1000, maxDelayMs = 30_000)) + assertEquals(4000L, computeBackoffMs(attempt = 4, baseDelayMs = 1000, maxDelayMs = 30_000)) + assertEquals(8000L, computeBackoffMs(attempt = 5, baseDelayMs = 1000, maxDelayMs = 30_000)) + assertEquals(30_000L, computeBackoffMs(attempt = 20, baseDelayMs = 1000, maxDelayMs = 30_000)) + // Zero base (used by tests) must never delay. + assertEquals(0L, computeBackoffMs(attempt = 5, baseDelayMs = 0, maxDelayMs = 30_000)) + } + + /** + * Regression for the infinite ABORTED reconnect loop (Bugsnag 6a4528ca). + * The stream activates (emits one response) then aborts with ABORTED on every + * attempt. Because it never stays healthy past `healthyThreshold`, the attempt + * counter must NOT reset, so the loop terminates at maxReconnectAttempts. + * + * Without the fix (reset on mere activation), this loops forever and the test + * times out. + */ + @Test + fun `activate-then-abort stream stops at maxReconnectAttempts`() = runTest { + var attemptCount = 0 + val errors = mutableListOf() + val timeSource = TestTimeSource() // stays at 0 → never healthy + + val streamRef = BidirectionalStreamReference(this, "test-stream") + streamRef.retain() + + openBidirectionalStream>( + streamRef = streamRef, + apiCall = { requestFlow -> + attemptCount++ + flow { + // RENDEZVOUS channel: requestChannel.send() in the coordinator suspends until + // a consumer is ready. Call first() here to rendezvous so send can proceed. + requestFlow.first() + emit("activation") // activates the stream (first response) + throw Status.ABORTED.withDescription("stream already exists").asRuntimeException() + } + }, + initialRequest = { "req" }, + responseHandler = { _: String, _: (String) -> Unit -> }, + onError = { errors.add(it) }, + reconnectOnAborted = true, + maxReconnectAttempts = 3, + reconnectDelayMs = 0, + healthyThreshold = 30.seconds, + timeSource = timeSource, + ) + + advanceUntilIdle() + + // maxReconnectAttempts=3 → 4 total attempts, then give up. + assertEquals(4, attemptCount, "Should stop after maxReconnectAttempts, got $attemptCount") + assertTrue( + errors.any { it is IllegalStateException }, + "Should terminate with max-attempts IllegalStateException" + ) + + streamRef.destroy() + } + + /** + * A stream that stays healthy (survives past healthyThreshold) between failures + * must reset the backoff counter and keep reconnecting well beyond + * maxReconnectAttempts, until a non-retryable error stops it. + */ + @Test + fun `healthy stream resets backoff counter across failures`() = runTest { + var attemptCount = 0 + val errors = mutableListOf() + val timeSource = TestTimeSource() + + val streamRef = BidirectionalStreamReference(this, "test-stream") + streamRef.retain() + + openBidirectionalStream>( + streamRef = streamRef, + apiCall = { requestFlow -> + attemptCount++ + flow { + // RENDEZVOUS channel: requestChannel.send() in the coordinator suspends until + // a consumer is ready. Call first() here to rendezvous so send can proceed. + requestFlow.first() + emit("activation") // activates + marks activation time + timeSource += 31.seconds // stream stayed healthy + if (attemptCount >= 5) { + throw IllegalArgumentException("stop") // non-retryable → terminate + } + throw Status.ABORTED.asRuntimeException() // healthy abort → reset counter + } + }, + initialRequest = { "req" }, + responseHandler = { _: String, _: (String) -> Unit -> }, + onError = { errors.add(it) }, + reconnectOnAborted = true, + maxReconnectAttempts = 2, // would stop at 3 attempts WITHOUT resets + reconnectDelayMs = 0, + healthyThreshold = 30.seconds, + timeSource = timeSource, + ) + + advanceUntilIdle() + + assertEquals(5, attemptCount, "Healthy resets should let it survive past maxReconnectAttempts") + assertTrue(errors.last() is IllegalArgumentException) + + streamRef.destroy() + } }