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