From ef85f5fd0bc47b6e09f1068de0856a4e811d18e2 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 27 May 2026 23:52:24 -0400 Subject: [PATCH] =?UTF-8?q?feat(android):=20EventsStream=20reconnect-with-?= =?UTF-8?q?backoff=20(audit=20v3=20=C2=A74.13)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously a mid-session SSE drop (server restart / network blip) left the stream closed until the next sign-in — cross-device reactivity (likes, playlist updates, request status, quarantine changes) silently died until app relaunch. Now onClosed / onFailure schedule a reconnect with exponential backoff (1s → 2s → … → 30s cap), reset to 1s on a successful onOpen. Reconnect only fires while signed in; sign-out's disconnect() cancels any pending retry. State swaps (connect/disconnect/scheduleReconnect) are @Synchronized since the OkHttp listener thread and the auth-cookie collector both touch currentSource + backoff. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../minstrel/events/EventsStream.kt | 71 ++++++++++++++++--- 1 file changed, 60 insertions(+), 11 deletions(-) diff --git a/android/app/src/main/java/com/fabledsword/minstrel/events/EventsStream.kt b/android/app/src/main/java/com/fabledsword/minstrel/events/EventsStream.kt index 5ed0fc90..b375756a 100644 --- a/android/app/src/main/java/com/fabledsword/minstrel/events/EventsStream.kt +++ b/android/app/src/main/java/com/fabledsword/minstrel/events/EventsStream.kt @@ -3,7 +3,9 @@ package com.fabledsword.minstrel.events import com.fabledsword.minstrel.auth.AuthStore import com.fabledsword.minstrel.di.ApplicationScope import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.asSharedFlow @@ -26,6 +28,9 @@ import javax.inject.Singleton private const val SSE_PATH = "/api/events/stream" private const val EVENTS_BUFFER_CAPACITY = 64 +private const val BASE_BACKOFF_MS = 1_000L +private const val MAX_BACKOFF_MS = 30_000L +private const val BACKOFF_FACTOR = 2 /** * Long-lived SSE subscription to `GET /api/events/stream`. Exposes @@ -40,9 +45,12 @@ private const val EVENTS_BUFFER_CAPACITY = 64 * back to null (sign-out). * - No client-side timeout — the server emits 15s heartbeats which * okhttp-sse handles transparently. - * - No explicit reconnect-with-backoff in v1; if the stream drops - * mid-session (server restart, network blip) it stays closed until - * the next auth-cookie transition. Matches Flutter's behavior. + * - Reconnect-with-backoff: if the stream drops mid-session (server + * restart, network blip) it reconnects with exponential backoff + * (1s → 2s → … → 30s cap), reset to 1s on a successful open. Only + * reconnects while still signed in; a sign-out cancels the pending + * retry. Without this a single blip silently kills cross-device + * reactivity until the next app launch. * * The URL passes through the placeholder host that * `BaseUrlInterceptor` rewrites — same mechanism the rest of the @@ -66,19 +74,59 @@ class EventsStream @Inject constructor( val events: SharedFlow = emitter.asSharedFlow() private var currentSource: EventSource? = null + @Volatile private var signedIn = false + private var reconnectJob: Job? = null + private var backoffMs = BASE_BACKOFF_MS init { scope.launch { authStore.sessionCookie .map { !it.isNullOrEmpty() } .distinctUntilChanged() - .collect { signedIn -> - currentSource?.cancel() - currentSource = if (signedIn) openSubscription() else null + .collect { isSignedIn -> + signedIn = isSignedIn + if (isSignedIn) { + backoffMs = BASE_BACKOFF_MS + connect() + } else { + disconnect() + } } } } + @Synchronized + private fun connect() { + reconnectJob?.cancel() + currentSource?.cancel() + currentSource = openSubscription() + } + + @Synchronized + private fun disconnect() { + reconnectJob?.cancel() + currentSource?.cancel() + currentSource = null + } + + /** + * Schedule a reconnect after the current backoff, then double it + * (capped). No-op when signed out — sign-out's [disconnect] + * cancels the pending job. A successful [Listener.onOpen] resets + * the backoff to the floor. + */ + @Synchronized + private fun scheduleReconnect() { + if (!signedIn) return + reconnectJob?.cancel() + val waitMs = backoffMs + backoffMs = (backoffMs * BACKOFF_FACTOR).coerceAtMost(MAX_BACKOFF_MS) + reconnectJob = scope.launch { + delay(waitMs) + if (signedIn) connect() + } + } + private fun openSubscription(): EventSource { val request = Request.Builder() .url("http://placeholder.invalid$SSE_PATH") @@ -88,6 +136,10 @@ class EventsStream @Inject constructor( } private inner class Listener : EventSourceListener() { + override fun onOpen(eventSource: EventSource, response: Response) { + backoffMs = BASE_BACKOFF_MS + } + override fun onEvent( eventSource: EventSource, id: String?, @@ -98,8 +150,7 @@ class EventsStream @Inject constructor( } override fun onClosed(eventSource: EventSource) { - // Stream ended (server-side close or network drop). No - // explicit reconnect — auth-cookie transitions re-open. + scheduleReconnect() } override fun onFailure( @@ -107,9 +158,7 @@ class EventsStream @Inject constructor( t: Throwable?, response: Response?, ) { - // Same as onClosed — auth transition will re-open. The - // throwable/response carry diagnostics; we don't surface - // them in v1 since the stream is "best effort" by design. + scheduleReconnect() } }