git.iko.soy/opencode-android
on d518ef7 help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/network/EventStreamClient.kt
353 lines 20 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 * SSL/TLS and certificate-pinning failures are permanent (a changed/rotated cert won't
83 * heal on retry) and are classified as Other so the user sees "check the server URL and
84 * version / cert pin" instead of an infinite "Reconnecting…" spinner. */
85 private enum class SseFailureKind { Auth, Other }
87 private fun classifyNonRetryableSseFailure(e: Throwable): SseFailureKind? =
88 generateSequence(e as Throwable?) { it.cause.takeIf { c -> c !== it } }
89 .take(16)
90 .mapNotNull { t ->
91 when (t) {
92 is io.ktor.client.plugins.sse.SSEClientException -> t.response?.status?.value
93 is io.ktor.client.plugins.ClientRequestException -> t.response.status.value
94 // A TLS handshake / cert-pinning / SSL peer unverified failure happens
95 // before any HTTP response, so it carries no status code. It is never
96 // transient (the server's cert must change), so classify it directly
97 // instead of falling through to the null (retryable) default — otherwise
98 // the client retries forever with backoff against an unreachable pinned
99 // server, showing "Reconnecting…" with no hint that the failure is fixed.
100 is javax.net.ssl.SSLException -> -1
101 is java.security.cert.CertificateException -> -1
102 else -> null
103 }
104 }
105 // Non-408/429 4xx = permanent; 408 and 429 stay in the transient retry path.
106 // The -1 sentinel (SSL/cert) is also permanent → Other.
107 .map { code -> when {
108 code == -1 -> SseFailureKind.Other
109 code in 400..499 && code != 408 && code != 429 ->
110 if (code == 401 || code == 403) SseFailureKind.Auth else SseFailureKind.Other
111 else -> null
112 } }
113 .firstOrNull { it != null }
115 /** Request an immediate reconnect, skipping any in-progress backoff. */
116 open fun triggerReconnect() { reconnectSignal.trySend(Unit) }
118 /**
119 * Read SSE events from [events], racing each read against an idle-timeout watchdog.
120 * Throws [IOException] on idle timeout so the outer reconnect loop fires; breaks on
121 * clean stream close or scope cancellation.
122 *
123 * If the idle watchdog tripped because a [triggerReconnect] signal was consumed
124 * (rather than a genuine idle timeout), [onReconnectConsumed] is invoked so the outer
125 * reconnect loop can reconnect immediately instead of waiting out the backoff — the
126 * signal is already consumed, so the backoff `select` couldn't be cut short.
127 */
128 private suspend fun readSseEvents(
129 scope: CoroutineScope,
130 events: kotlinx.coroutines.channels.ReceiveChannel<ServerSentEvent>,
131 onReconnectConsumed: () -> Unit,
132 send: suspend (BusEvent) -> Unit,
133 ) {
134 while (scope.isActive) {
135 // The OkHttp engine has readTimeout=0 (the SSE stream must never time out on
136 // read), so a socket the peer never closed would otherwise hang until OS TCP
137 // keepalive kicks in — minutes to hours on stock Linux. If nothing arrives
138 // within the idle timeout, treat the connection as half-open and drop it.
139 var consumedReconnectSignal = false
140 val result: ChannelResult<ServerSentEvent>? = select {
141 events.onReceiveCatching { it }
142 onTimeout(idleTimeoutMs) { null }
143 // A triggerReconnect() issued while a read is active must interrupt it now:
144 // return the same null sentinel as the idle timeout so a half-open socket
145 // (no TCP RST) is dropped and reconnected promptly instead of waiting out
146 // idleTimeoutMs (~90s).
147 reconnectSignal.onReceive { consumedReconnectSignal = true; null }
148 }
149 if (result == null) {
150 if (consumedReconnectSignal) onReconnectConsumed()
151 throw IOException("SSE read dropped (idle timeout or reconnect requested), reconnecting")
152 }
153 val sse = result.getOrNull()
154 if (sse == null) {
155 // The producer channel is closed. A non-null cause means `incoming`
156 // failed (e.g. a socket reset bridged from the producer) — rethrow it so
157 // the outer loop logs and retries with backoff. A clean close (null
158 // cause) just ends this attempt and lets the loop reconnect.
159 result.exceptionOrNull()?.let { throw it }
160 break
161 }
162 val data = sse.data
163 if (data == null) {
164 // A comment or keep-alive event (no data field) resets the idle watchdog
165 // — servers send `: ping` comments to keep the connection alive through
166 // proxies. Without this, a server that only sends comments would still
167 // hit the idle timeout and reconnect unnecessarily.
168 continue
169 }
170 val event = try {
171 OpencodeJson.decodeFromString(BusEvent.serializer(), data)
172 } catch (e: kotlinx.serialization.SerializationException) {
173 Log.d("EventStream", "Skipping unparseable SSE event", e)
174 continue
175 }
176 if (_state.value != ConnectionState.Connected) {
177 _state.value = ConnectionState.Connected
178 }
179 send(event)
180 }
181 }
183 private fun stream(): Flow<BusEvent> = channelFlow {
184 val scope = this
185 // Drain any stale reconnect signal from a previous collection cycle so it
186 // doesn't spuriously reset the backoff on startup (WhileSubscribed may have
187 // stopped the upstream while triggerReconnect() queued a signal).
188 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
189 var backoffMs = initialBackoffMs
190 // Tracks whether this is the first iteration (the initial connect) vs a reconnect,
191 // so the attempt counter is only incremented on reconnects (not the first connect).
192 var firstAttempt = true
193 while (isActive) {
194 // Increment the reconnect-attempt counter on every iteration except the initial
195 // connect, so the UI can show "Reconnecting (attempt N)…". Reset to 0 on success.
196 if (firstAttempt) firstAttempt = false else _reconnectAttempts.value++
197 _state.value = ConnectionState.Connecting
198 // Set by readSseEvents when a triggerReconnect() signal interrupted an active
199 // read. The signal is consumed to drop the socket, so the backoff select below
200 // can't also see it — without this flag the reconnect would wait the full
201 // backoff despite the caller asking for an immediate reconnect (network
202 // handoff, manual refresh, ...).
203 var reconnectConsumedDuringRead = false
204 try {
205 // The SSE stream is long-lived: disable the request-level and socket
206 // timeouts (the client default is 60s for REST) in the request config
207 // so the connection isn't killed mid-stream. The idle-timeout watchdog
208 // below handles half-open connections instead.
209 client.sse(
210 "event",
211 request = {
212 timeout {
213 requestTimeoutMillis = Long.MAX_VALUE
214 // Disable the socket (read) timeout entirely for the SSE stream.
215 // A finite value would kill the connection whenever the server
216 // is quiet for longer than that (idle sessions, slow agent runs),
217 // causing a reconnect storm. Half-open detection is handled by
218 // the idle-timeout watchdog below plus OkHttp's pingInterval on
219 // HTTP/2 connections.
220 socketTimeoutMillis = Long.MAX_VALUE
221 }
222 },
223 ) {
224 // The SSE block is open — the connection is established. Mark as
225 // Connected immediately so downstream re-seed logic fires even when
226 // the server only sends keep-alive comments (no data events).
227 if (_state.value != ConnectionState.Connected) {
228 _state.value = ConnectionState.Connected
229 }
230 // Clear any stale reconnect signal from a triggerReconnect() call
231 // made while the previous stream was healthy, so it doesn't suppress
232 // the next legitimate backoff.
233 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
234 // Bridge ktor's `incoming` flow to a channel so each element can be
235 // raced against the idle-timeout watchdog via select. We collect it
236 // ourselves instead of `incoming.produceIn(scope)`: produceIn launches
237 // the collector as a child of this channelFlow, so when `incoming`
238 // throws on an abnormal socket close — a TCP reset, i.e. the common
239 // mobile case: Wi-Fi⇄cellular handoff, server or proxy RST — the child's
240 // failure cancels the whole channelFlow and bypasses the retry `catch`
241 // below. That killed reconnect outright and surfaced the IOException as
242 // an uncaught crash through the (handler-less) shareIn scope. Collecting
243 // into our own channel and closing it with the cause keeps the failure
244 // local: readSseEvents sees the closed channel, rethrows the cause, and
245 // the normal retry/backoff path handles it. (A clean EOF closes the
246 // channel with no cause → readSseEvents breaks and reconnects as before.)
247 val events = Channel<ServerSentEvent>(Channel.BUFFERED)
248 val producer = scope.launch {
249 try {
250 incoming.collect { events.send(it) }
251 events.close()
252 } catch (c: CancellationException) {
253 events.close(c)
254 throw c
255 } catch (t: Throwable) {
256 // Deliver the failure via the channel rather than letting it
257 // propagate up and cancel the parent channelFlow.
258 events.close(t)
259 }
260 }
261 // Reset the backoff only once real data actually arrives, not on
262 // block entry: a server that accepts the connection then immediately
263 // drops it would otherwise loop forever at the initial backoff and
264 // never escalate the exponential ramp.
265 var receivedAny = false
266 try {
267 readSseEvents(scope, events, { reconnectConsumedDuringRead = true }) {
268 if (!receivedAny) {
269 receivedAny = true
270 backoffMs = initialBackoffMs
271 // A genuine event confirms the connection is healthy, so reset
272 // the reconnect-attempt counter for the next outage's count.
273 _reconnectAttempts.value = 0
274 }
275 send(it)
276 }
277 } finally {
278 producer.cancel()
279 events.cancel()
280 }
281 }
282 } catch (c: CancellationException) {
283 _state.value = ConnectionState.Disconnected
284 throw c
285 } catch (e: Exception) {
286 // If the scope was cancelled (e.g. connection close), the closed-client
287 // exception is expected — don't log a spurious "stream error" warning.
288 val failureKind = classifyNonRetryableSseFailure(e)
289 if (failureKind != null) {
290 // The server rejected the request with a non-retryable 4xx (bad
291 // credentials, missing endpoint, ...) — retrying won't help. Log a
292 // scrubbed summary (class + status) instead of the full exception, whose
293 // message carries the request URL and may include auth or paths.
294 Log.w("EventStream", "SSE non-retryable 4xx (${if (failureKind == SseFailureKind.Auth) "auth" else "endpoint"}), awaiting explicit reconnect: ${safeExceptionSummary(e)}")
295 _state.value = if (failureKind == SseFailureKind.Auth) ConnectionState.AuthFailed else ConnectionState.Failed
296 if (!isActive) {
297 // Final teardown (last subscriber left) while a 4xx surfaced: end in
298 // Disconnected like every other loop-exit path. Failed must stay
299 // observable only while the flow is live/retrying, not after teardown.
300 _state.value = ConnectionState.Disconnected
301 break
302 }
303 // Don't complete the flow: a subscriber may fix the profile and call
304 // triggerReconnect(). Suspend until an explicit reconnect signal
305 // arrives (no backoff delay), then re-attempt the connection. This receive
306 // is outside the try above, so reset state to Disconnected if the last
307 // subscriber leaves (cancellation) while we're parked here — otherwise the
308 // stream would be torn down with state stuck at Failed.
309 try {
310 reconnectSignal.receive()
311 } catch (c: CancellationException) {
312 _state.value = ConnectionState.Disconnected
313 throw c
314 }
315 continue
316 } else if (isActive) {
317 Log.w("EventStream", "SSE stream error, will retry: ${safeExceptionSummary(e)}")
318 }
319 }
320 _state.value = ConnectionState.Disconnected
321 if (!isActive) break
322 // Wait for the backoff, but allow a reconnect signal to cut it short. When
323 // readSseEvents consumed a triggerReconnect() signal to drop an active read,
324 // the signal is already spent so the select below can't receive it — pass a
325 // zero backoff in that case so the reconnect is immediate (the whole point of
326 // triggerReconnect: network handoff, manual refresh, ...).
327 val signaled = waitForBackoffOrSignal(reconnectConsumedDuringRead, backoffMs)
328 backoffMs = if (signaled) initialBackoffMs else (backoffMs * 2).coerceAtMost(maxBackoffMs)
329 }
330 }
332 /** Suspend for the backoff unless [reconnectConsumedDuringRead] (immediate) or a new
333 * [triggerReconnect] arrives during the wait. Returns true when the reconnect should
334 * skip the backoff ramp (signal-driven), false when the full backoff elapsed. */
335 private suspend fun waitForBackoffOrSignal(
336 reconnectConsumedDuringRead: Boolean,
337 backoffMs: Long,
338 ): Boolean {
339 if (reconnectConsumedDuringRead) return true
340 var signaled = false
341 val jitter = ((backoffMs * NetworkConfig.retryJitterFactor) * (Random.nextDouble() * 2 - 1)).toLong()
342 // Floor the wait at half the initial backoff: symmetric jitter can otherwise pull a
343 // small backoff down to 0ms, producing a sub-second reconnect storm right after a
344 // triggerReconnect reset on a connect-RST loop. A non-zero floor preserves the
345 // purpose of backoff (spacing retries) while still allowing jitter to shorten it.
346 val minWait = (initialBackoffMs / 2).coerceAtLeast(1L)
347 select {
348 reconnectSignal.onReceive { signaled = true }
349 onTimeout((backoffMs + jitter).coerceAtLeast(minWait)) { /* normal backoff elapsed */ }
350 }
351 return signaled
352 }