feat(android): EventsStream reconnect-with-backoff (audit v3 §4.13)
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) <noreply@anthropic.com>
This commit is contained in:
@@ -3,7 +3,9 @@ package com.fabledsword.minstrel.events
|
|||||||
import com.fabledsword.minstrel.auth.AuthStore
|
import com.fabledsword.minstrel.auth.AuthStore
|
||||||
import com.fabledsword.minstrel.di.ApplicationScope
|
import com.fabledsword.minstrel.di.ApplicationScope
|
||||||
import kotlinx.coroutines.CoroutineScope
|
import kotlinx.coroutines.CoroutineScope
|
||||||
|
import kotlinx.coroutines.Job
|
||||||
import kotlinx.coroutines.channels.BufferOverflow
|
import kotlinx.coroutines.channels.BufferOverflow
|
||||||
|
import kotlinx.coroutines.delay
|
||||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||||
import kotlinx.coroutines.flow.SharedFlow
|
import kotlinx.coroutines.flow.SharedFlow
|
||||||
import kotlinx.coroutines.flow.asSharedFlow
|
import kotlinx.coroutines.flow.asSharedFlow
|
||||||
@@ -26,6 +28,9 @@ import javax.inject.Singleton
|
|||||||
|
|
||||||
private const val SSE_PATH = "/api/events/stream"
|
private const val SSE_PATH = "/api/events/stream"
|
||||||
private const val EVENTS_BUFFER_CAPACITY = 64
|
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
|
* 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).
|
* back to null (sign-out).
|
||||||
* - No client-side timeout — the server emits 15s heartbeats which
|
* - No client-side timeout — the server emits 15s heartbeats which
|
||||||
* okhttp-sse handles transparently.
|
* okhttp-sse handles transparently.
|
||||||
* - No explicit reconnect-with-backoff in v1; if the stream drops
|
* - Reconnect-with-backoff: if the stream drops mid-session (server
|
||||||
* mid-session (server restart, network blip) it stays closed until
|
* restart, network blip) it reconnects with exponential backoff
|
||||||
* the next auth-cookie transition. Matches Flutter's behavior.
|
* (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
|
* The URL passes through the placeholder host that
|
||||||
* `BaseUrlInterceptor` rewrites — same mechanism the rest of the
|
* `BaseUrlInterceptor` rewrites — same mechanism the rest of the
|
||||||
@@ -66,19 +74,59 @@ class EventsStream @Inject constructor(
|
|||||||
val events: SharedFlow<LiveEvent> = emitter.asSharedFlow()
|
val events: SharedFlow<LiveEvent> = emitter.asSharedFlow()
|
||||||
|
|
||||||
private var currentSource: EventSource? = null
|
private var currentSource: EventSource? = null
|
||||||
|
@Volatile private var signedIn = false
|
||||||
|
private var reconnectJob: Job? = null
|
||||||
|
private var backoffMs = BASE_BACKOFF_MS
|
||||||
|
|
||||||
init {
|
init {
|
||||||
scope.launch {
|
scope.launch {
|
||||||
authStore.sessionCookie
|
authStore.sessionCookie
|
||||||
.map { !it.isNullOrEmpty() }
|
.map { !it.isNullOrEmpty() }
|
||||||
.distinctUntilChanged()
|
.distinctUntilChanged()
|
||||||
.collect { signedIn ->
|
.collect { isSignedIn ->
|
||||||
currentSource?.cancel()
|
signedIn = isSignedIn
|
||||||
currentSource = if (signedIn) openSubscription() else null
|
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 {
|
private fun openSubscription(): EventSource {
|
||||||
val request = Request.Builder()
|
val request = Request.Builder()
|
||||||
.url("http://placeholder.invalid$SSE_PATH")
|
.url("http://placeholder.invalid$SSE_PATH")
|
||||||
@@ -88,6 +136,10 @@ class EventsStream @Inject constructor(
|
|||||||
}
|
}
|
||||||
|
|
||||||
private inner class Listener : EventSourceListener() {
|
private inner class Listener : EventSourceListener() {
|
||||||
|
override fun onOpen(eventSource: EventSource, response: Response) {
|
||||||
|
backoffMs = BASE_BACKOFF_MS
|
||||||
|
}
|
||||||
|
|
||||||
override fun onEvent(
|
override fun onEvent(
|
||||||
eventSource: EventSource,
|
eventSource: EventSource,
|
||||||
id: String?,
|
id: String?,
|
||||||
@@ -98,8 +150,7 @@ class EventsStream @Inject constructor(
|
|||||||
}
|
}
|
||||||
|
|
||||||
override fun onClosed(eventSource: EventSource) {
|
override fun onClosed(eventSource: EventSource) {
|
||||||
// Stream ended (server-side close or network drop). No
|
scheduleReconnect()
|
||||||
// explicit reconnect — auth-cookie transitions re-open.
|
|
||||||
}
|
}
|
||||||
|
|
||||||
override fun onFailure(
|
override fun onFailure(
|
||||||
@@ -107,9 +158,7 @@ class EventsStream @Inject constructor(
|
|||||||
t: Throwable?,
|
t: Throwable?,
|
||||||
response: Response?,
|
response: Response?,
|
||||||
) {
|
) {
|
||||||
// Same as onClosed — auth transition will re-open. The
|
scheduleReconnect()
|
||||||
// throwable/response carry diagnostics; we don't surface
|
|
||||||
// them in v1 since the stream is "best effort" by design.
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user