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