git.iko.soy/opencode-android
on c4151ca help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/network/EventStreamClient.kt
335 lines 19 kB history raw
1package soy.iko.opencode.data.network
3import soy.iko.opencode.data.model.BusEvent
4import soy.iko.opencode.util.safeExceptionSummary
5import android.util.Log
6import io.ktor.client.HttpClient
7import io.ktor.client.plugins.timeout
8import io.ktor.client.plugins.sse.sse
9import io.ktor.sse.ServerSentEvent
10import kotlinx.coroutines.CancellationException
11import kotlinx.coroutines.CoroutineScope
12import kotlin.coroutines.EmptyCoroutineContext
13import kotlinx.coroutines.channels.BufferOverflow
14import kotlinx.coroutines.channels.Channel
15import kotlinx.coroutines.channels.ChannelResult
16import kotlinx.coroutines.flow.Flow
17import kotlinx.coroutines.flow.MutableStateFlow
18import kotlinx.coroutines.flow.SharedFlow
19import kotlinx.coroutines.flow.StateFlow
20import kotlinx.coroutines.flow.SharingStarted
21import kotlinx.coroutines.flow.asStateFlow
22import kotlinx.coroutines.flow.buffer
23import kotlinx.coroutines.flow.channelFlow
24import kotlinx.coroutines.flow.shareIn
25import kotlinx.coroutines.isActive
26import kotlinx.coroutines.launch
27import kotlinx.coroutines.selects.onTimeout
28import kotlinx.coroutines.selects.select
29import java.io.IOException
30import kotlin.random.Random
32/**
33 * Owns the single long-lived `GET /event` SSE subscription and exposes it as a hot
34 * [SharedFlow] of [BusEvent]. The connection is opened when the first collector
35 * subscribes (WhileSubscribed) and reconnects with backoff on drop.
36 *
37 * Ordering contract: callers that send a prompt must already be collecting [events]
38 * (directly or via the repository) so early `message.part.updated` deltas aren't missed.
39 */
40@OptIn(kotlinx.coroutines.ExperimentalCoroutinesApi::class)
41open class EventStreamClient(
42 private val client: HttpClient,
43 private val scope: CoroutineScope,
44 private val idleTimeoutMs: Long = NetworkConfig.sseIdleTimeoutMs,
45 private val initialBackoffMs: Long = NetworkConfig.sseInitialBackoffMs,
46 private val maxBackoffMs: Long = NetworkConfig.sseMaxBackoffMs,
47 private val bufferCapacity: Int = NetworkConfig.sseEventBufferCapacity,
48) {
49 protected constructor() : this(
50 HttpClient(io.ktor.client.engine.okhttp.OkHttp) {},
51 CoroutineScope(EmptyCoroutineContext),
52 )
53 enum class ConnectionState { Disconnected, Connecting, Connected, Failed, AuthFailed }
55 private val _state = MutableStateFlow(ConnectionState.Disconnected)
56 open val state: StateFlow<ConnectionState> = _state.asStateFlow()
58 /** Number of reconnect attempts made since the last successful [ConnectionState.Connected]
59 * period. Incremented each time the stream loop begins a new connection attempt after a
60 * drop, and reset to 0 on a successful connect. Surfaced so the UI can show "Reconnecting
61 * (attempt N)…" during a sustained outage, giving the user confidence the system is
62 * actively retrying rather than stuck. */
63 private val _reconnectAttempts = MutableStateFlow(0)
64 open val reconnectAttempts: StateFlow<Int> = _reconnectAttempts.asStateFlow()
66 open val events: SharedFlow<BusEvent> by lazy {
67 stream()
68 .buffer(bufferCapacity, BufferOverflow.DROP_OLDEST)
69 .shareIn(
70 scope = scope,
71 started = SharingStarted.WhileSubscribed(stopTimeoutMillis = NetworkConfig.stateFlowSubscriptionTimeoutMs),
72 replay = 0,
73 )
74 }
76 /** Conflated channel used to cut the reconnect backoff short when network returns. */
77 private val reconnectSignal = Channel<Unit>(Channel.CONFLATED)
79 /** Classifies a non-retryable SSE failure so the UI can show the right recovery hint:
80 * an auth failure (401/403) reads "check credentials", while other 4xx (400/404/405…)
81 * reads "check the server URL and version". Returns null for transient/retryable errors. */
82 private enum class SseFailureKind { Auth, Other }
84 private fun classifyNonRetryableSseFailure(e: Throwable): SseFailureKind? =
85 generateSequence(e as Throwable?) { it.cause.takeIf { c -> c !== it } }
86 .take(16)
87 .mapNotNull { t ->
88 when (t) {
89 is io.ktor.client.plugins.sse.SSEClientException -> t.response?.status?.value
90 is io.ktor.client.plugins.ClientRequestException -> t.response.status.value
91 else -> null
92 }
93 }
94 // Non-408/429 4xx = permanent; 408 and 429 stay in the transient retry path.
95 .map { code -> when {
96 code in 400..499 && code != 408 && code != 429 ->
97 if (code == 401 || code == 403) SseFailureKind.Auth else SseFailureKind.Other
98 else -> null
99 } }
100 .firstOrNull { it != null }
102 /** Request an immediate reconnect, skipping any in-progress backoff. */
103 open fun triggerReconnect() { reconnectSignal.trySend(Unit) }
105 /**
106 * Read SSE events from [events], racing each read against an idle-timeout watchdog.
107 * Throws [IOException] on idle timeout so the outer reconnect loop fires; breaks on
108 * clean stream close or scope cancellation.
109 *
110 * If the idle watchdog tripped because a [triggerReconnect] signal was consumed
111 * (rather than a genuine idle timeout), [onReconnectConsumed] is invoked so the outer
112 * reconnect loop can reconnect immediately instead of waiting out the backoff — the
113 * signal is already consumed, so the backoff `select` couldn't be cut short.
114 */
115 private suspend fun readSseEvents(
116 scope: CoroutineScope,
117 events: kotlinx.coroutines.channels.ReceiveChannel<ServerSentEvent>,
118 onReconnectConsumed: () -> Unit,
119 send: suspend (BusEvent) -> Unit,
120 ) {
121 while (scope.isActive) {
122 // The OkHttp engine has readTimeout=0 (the SSE stream must never time out on
123 // read), so a socket the peer never closed would otherwise hang until OS TCP
124 // keepalive kicks in — minutes to hours on stock Linux. If nothing arrives
125 // within the idle timeout, treat the connection as half-open and drop it.
126 var consumedReconnectSignal = false
127 val result: ChannelResult<ServerSentEvent>? = select {
128 events.onReceiveCatching { it }
129 onTimeout(idleTimeoutMs) { null }
130 // A triggerReconnect() issued while a read is active must interrupt it now:
131 // return the same null sentinel as the idle timeout so a half-open socket
132 // (no TCP RST) is dropped and reconnected promptly instead of waiting out
133 // idleTimeoutMs (~90s).
134 reconnectSignal.onReceive { consumedReconnectSignal = true; null }
135 }
136 if (result == null) {
137 if (consumedReconnectSignal) onReconnectConsumed()
138 throw IOException("SSE read dropped (idle timeout or reconnect requested), reconnecting")
139 }
140 val sse = result.getOrNull()
141 if (sse == null) {
142 // The producer channel is closed. A non-null cause means `incoming`
143 // failed (e.g. a socket reset bridged from the producer) — rethrow it so
144 // the outer loop logs and retries with backoff. A clean close (null
145 // cause) just ends this attempt and lets the loop reconnect.
146 result.exceptionOrNull()?.let { throw it }
147 break
148 }
149 val data = sse.data
150 if (data == null) {
151 // A comment or keep-alive event (no data field) resets the idle watchdog
152 // — servers send `: ping` comments to keep the connection alive through
153 // proxies. Without this, a server that only sends comments would still
154 // hit the idle timeout and reconnect unnecessarily.
155 continue
156 }
157 val event = try {
158 OpencodeJson.decodeFromString(BusEvent.serializer(), data)
159 } catch (e: kotlinx.serialization.SerializationException) {
160 Log.d("EventStream", "Skipping unparseable SSE event", e)
161 continue
162 }
163 if (_state.value != ConnectionState.Connected) {
164 _state.value = ConnectionState.Connected
165 }
166 send(event)
167 }
168 }
170 private fun stream(): Flow<BusEvent> = channelFlow {
171 val scope = this
172 // Drain any stale reconnect signal from a previous collection cycle so it
173 // doesn't spuriously reset the backoff on startup (WhileSubscribed may have
174 // stopped the upstream while triggerReconnect() queued a signal).
175 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
176 var backoffMs = initialBackoffMs
177 // Tracks whether this is the first iteration (the initial connect) vs a reconnect,
178 // so the attempt counter is only incremented on reconnects (not the first connect).
179 var firstAttempt = true
180 while (isActive) {
181 // Increment the reconnect-attempt counter on every iteration except the initial
182 // connect, so the UI can show "Reconnecting (attempt N)…". Reset to 0 on success.
183 if (firstAttempt) firstAttempt = false else _reconnectAttempts.value++
184 _state.value = ConnectionState.Connecting
185 // Set by readSseEvents when a triggerReconnect() signal interrupted an active
186 // read. The signal is consumed to drop the socket, so the backoff select below
187 // can't also see it — without this flag the reconnect would wait the full
188 // backoff despite the caller asking for an immediate reconnect (network
189 // handoff, manual refresh, ...).
190 var reconnectConsumedDuringRead = false
191 try {
192 // The SSE stream is long-lived: disable the request-level and socket
193 // timeouts (the client default is 60s for REST) in the request config
194 // so the connection isn't killed mid-stream. The idle-timeout watchdog
195 // below handles half-open connections instead.
196 client.sse(
197 "event",
198 request = {
199 timeout {
200 requestTimeoutMillis = Long.MAX_VALUE
201 // Disable the socket (read) timeout entirely for the SSE stream.
202 // A finite value would kill the connection whenever the server
203 // is quiet for longer than that (idle sessions, slow agent runs),
204 // causing a reconnect storm. Half-open detection is handled by
205 // the idle-timeout watchdog below plus OkHttp's pingInterval on
206 // HTTP/2 connections.
207 socketTimeoutMillis = Long.MAX_VALUE
208 }
209 },
210 ) {
211 // The SSE block is open — the connection is established. Mark as
212 // Connected immediately so downstream re-seed logic fires even when
213 // the server only sends keep-alive comments (no data events).
214 if (_state.value != ConnectionState.Connected) {
215 _state.value = ConnectionState.Connected
216 }
217 // Clear any stale reconnect signal from a triggerReconnect() call
218 // made while the previous stream was healthy, so it doesn't suppress
219 // the next legitimate backoff.
220 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
221 // Bridge ktor's `incoming` flow to a channel so each element can be
222 // raced against the idle-timeout watchdog via select. We collect it
223 // ourselves instead of `incoming.produceIn(scope)`: produceIn launches
224 // the collector as a child of this channelFlow, so when `incoming`
225 // throws on an abnormal socket close — a TCP reset, i.e. the common
226 // mobile case: Wi-Fi⇄cellular handoff, server or proxy RST — the child's
227 // failure cancels the whole channelFlow and bypasses the retry `catch`
228 // below. That killed reconnect outright and surfaced the IOException as
229 // an uncaught crash through the (handler-less) shareIn scope. Collecting
230 // into our own channel and closing it with the cause keeps the failure
231 // local: readSseEvents sees the closed channel, rethrows the cause, and
232 // the normal retry/backoff path handles it. (A clean EOF closes the
233 // channel with no cause → readSseEvents breaks and reconnects as before.)
234 val events = Channel<ServerSentEvent>(Channel.BUFFERED)
235 val producer = scope.launch {
236 try {
237 incoming.collect { events.send(it) }
238 events.close()
239 } catch (c: CancellationException) {
240 events.close(c)
241 throw c
242 } catch (t: Throwable) {
243 // Deliver the failure via the channel rather than letting it
244 // propagate up and cancel the parent channelFlow.
245 events.close(t)
246 }
247 }
248 // Reset the backoff only once real data actually arrives, not on
249 // block entry: a server that accepts the connection then immediately
250 // drops it would otherwise loop forever at the initial backoff and
251 // never escalate the exponential ramp.
252 var receivedAny = false
253 try {
254 readSseEvents(scope, events, { reconnectConsumedDuringRead = true }) {
255 if (!receivedAny) {
256 receivedAny = true
257 backoffMs = initialBackoffMs
258 // A genuine event confirms the connection is healthy, so reset
259 // the reconnect-attempt counter for the next outage's count.
260 _reconnectAttempts.value = 0
261 }
262 send(it)
263 }
264 } finally {
265 producer.cancel()
266 events.cancel()
267 }
268 }
269 } catch (c: CancellationException) {
270 _state.value = ConnectionState.Disconnected
271 throw c
272 } catch (e: Exception) {
273 // If the scope was cancelled (e.g. connection close), the closed-client
274 // exception is expected — don't log a spurious "stream error" warning.
275 val failureKind = classifyNonRetryableSseFailure(e)
276 if (failureKind != null) {
277 // The server rejected the request with a non-retryable 4xx (bad
278 // credentials, missing endpoint, ...) — retrying won't help. Log a
279 // scrubbed summary (class + status) instead of the full exception, whose
280 // message carries the request URL and may include auth or paths.
281 Log.w("EventStream", "SSE non-retryable 4xx (${if (failureKind == SseFailureKind.Auth) "auth" else "endpoint"}), awaiting explicit reconnect: ${safeExceptionSummary(e)}")
282 _state.value = if (failureKind == SseFailureKind.Auth) ConnectionState.AuthFailed else ConnectionState.Failed
283 if (!isActive) {
284 // Final teardown (last subscriber left) while a 4xx surfaced: end in
285 // Disconnected like every other loop-exit path. Failed must stay
286 // observable only while the flow is live/retrying, not after teardown.
287 _state.value = ConnectionState.Disconnected
288 break
289 }
290 // Don't complete the flow: a subscriber may fix the profile and call
291 // triggerReconnect(). Suspend until an explicit reconnect signal
292 // arrives (no backoff delay), then re-attempt the connection. This receive
293 // is outside the try above, so reset state to Disconnected if the last
294 // subscriber leaves (cancellation) while we're parked here — otherwise the
295 // stream would be torn down with state stuck at Failed.
296 try {
297 reconnectSignal.receive()
298 } catch (c: CancellationException) {
299 _state.value = ConnectionState.Disconnected
300 throw c
301 }
302 continue
303 } else if (isActive) {
304 Log.w("EventStream", "SSE stream error, will retry: ${safeExceptionSummary(e)}")
305 }
306 }
307 _state.value = ConnectionState.Disconnected
308 if (!isActive) break
309 // Wait for the backoff, but allow a reconnect signal to cut it short. When
310 // readSseEvents consumed a triggerReconnect() signal to drop an active read,
311 // the signal is already spent so the select below can't receive it — pass a
312 // zero backoff in that case so the reconnect is immediate (the whole point of
313 // triggerReconnect: network handoff, manual refresh, ...).
314 val signaled = waitForBackoffOrSignal(reconnectConsumedDuringRead, backoffMs)
315 backoffMs = if (signaled) initialBackoffMs else (backoffMs * 2).coerceAtMost(maxBackoffMs)
316 }
317 }
319 /** Suspend for the backoff unless [reconnectConsumedDuringRead] (immediate) or a new
320 * [triggerReconnect] arrives during the wait. Returns true when the reconnect should
321 * skip the backoff ramp (signal-driven), false when the full backoff elapsed. */
322 private suspend fun waitForBackoffOrSignal(
323 reconnectConsumedDuringRead: Boolean,
324 backoffMs: Long,
325 ): Boolean {
326 if (reconnectConsumedDuringRead) return true
327 var signaled = false
328 val jitter = ((backoffMs * NetworkConfig.retryJitterFactor) * (Random.nextDouble() * 2 - 1)).toLong()
329 select {
330 reconnectSignal.onReceive { signaled = true }
331 onTimeout((backoffMs + jitter).coerceAtLeast(0)) { /* normal backoff elapsed */ }
332 }
333 return signaled
334 }