git.iko.soy/opencode-android
on 5d56632 help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/network/EventStreamClient.kt
279 lines 15 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 private suspend fun readSseEvents(
102 scope: CoroutineScope,
103 events: kotlinx.coroutines.channels.ReceiveChannel<ServerSentEvent>,
104 send: suspend (BusEvent) -> Unit,
105 ) {
106 while (scope.isActive) {
107 // The OkHttp engine has readTimeout=0 (the SSE stream must never time out on
108 // read), so a socket the peer never closed would otherwise hang until OS TCP
109 // keepalive kicks in — minutes to hours on stock Linux. If nothing arrives
110 // within the idle timeout, treat the connection as half-open and drop it.
111 val result: ChannelResult<ServerSentEvent>? = select {
112 events.onReceiveCatching { it }
113 onTimeout(idleTimeoutMs) { null }
114 // A triggerReconnect() issued while a read is active must interrupt it now:
115 // return the same null sentinel as the idle timeout so a half-open socket
116 // (no TCP RST) is dropped and reconnected promptly instead of waiting out
117 // idleTimeoutMs (~90s). Consuming the signal here also stops the backoff
118 // select below from re-triggering on the stale request.
119 reconnectSignal.onReceive { null }
120 }
121 if (result == null) throw IOException("SSE read dropped (idle timeout or reconnect requested), reconnecting")
122 val sse = result.getOrNull()
123 if (sse == null) {
124 // The producer channel is closed. A non-null cause means `incoming`
125 // failed (e.g. a socket reset bridged from the producer) — rethrow it so
126 // the outer loop logs and retries with backoff. A clean close (null
127 // cause) just ends this attempt and lets the loop reconnect.
128 result.exceptionOrNull()?.let { throw it }
129 break
130 }
131 val data = sse.data
132 if (data == null) {
133 // A comment or keep-alive event (no data field) resets the idle watchdog
134 // — servers send `: ping` comments to keep the connection alive through
135 // proxies. Without this, a server that only sends comments would still
136 // hit the idle timeout and reconnect unnecessarily.
137 continue
138 }
139 val event = try {
140 OpencodeJson.decodeFromString(BusEvent.serializer(), data)
141 } catch (e: kotlinx.serialization.SerializationException) {
142 Log.d("EventStream", "Skipping unparseable SSE event", e)
143 continue
144 }
145 if (_state.value != ConnectionState.Connected) {
146 _state.value = ConnectionState.Connected
147 }
148 send(event)
149 }
150 }
152 private fun stream(): Flow<BusEvent> = channelFlow {
153 val scope = this
154 // Drain any stale reconnect signal from a previous collection cycle so it
155 // doesn't spuriously reset the backoff on startup (WhileSubscribed may have
156 // stopped the upstream while triggerReconnect() queued a signal).
157 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
158 var backoffMs = initialBackoffMs
159 while (isActive) {
160 _state.value = ConnectionState.Connecting
161 try {
162 // The SSE stream is long-lived: disable the request-level and socket
163 // timeouts (the client default is 60s for REST) in the request config
164 // so the connection isn't killed mid-stream. The idle-timeout watchdog
165 // below handles half-open connections instead.
166 client.sse(
167 "event",
168 request = {
169 timeout {
170 requestTimeoutMillis = Long.MAX_VALUE
171 // Disable the socket (read) timeout entirely for the SSE stream.
172 // A finite value would kill the connection whenever the server
173 // is quiet for longer than that (idle sessions, slow agent runs),
174 // causing a reconnect storm. Half-open detection is handled by
175 // the idle-timeout watchdog below plus OkHttp's pingInterval on
176 // HTTP/2 connections.
177 socketTimeoutMillis = Long.MAX_VALUE
178 }
179 },
180 ) {
181 // The SSE block is open — the connection is established. Mark as
182 // Connected immediately so downstream re-seed logic fires even when
183 // the server only sends keep-alive comments (no data events).
184 if (_state.value != ConnectionState.Connected) {
185 _state.value = ConnectionState.Connected
186 }
187 // Clear any stale reconnect signal from a triggerReconnect() call
188 // made while the previous stream was healthy, so it doesn't suppress
189 // the next legitimate backoff.
190 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
191 // Bridge ktor's `incoming` flow to a channel so each element can be
192 // raced against the idle-timeout watchdog via select. We collect it
193 // ourselves instead of `incoming.produceIn(scope)`: produceIn launches
194 // the collector as a child of this channelFlow, so when `incoming`
195 // throws on an abnormal socket close — a TCP reset, i.e. the common
196 // mobile case: Wi-Fi⇄cellular handoff, server or proxy RST — the child's
197 // failure cancels the whole channelFlow and bypasses the retry `catch`
198 // below. That killed reconnect outright and surfaced the IOException as
199 // an uncaught crash through the (handler-less) shareIn scope. Collecting
200 // into our own channel and closing it with the cause keeps the failure
201 // local: readSseEvents sees the closed channel, rethrows the cause, and
202 // the normal retry/backoff path handles it. (A clean EOF closes the
203 // channel with no cause → readSseEvents breaks and reconnects as before.)
204 val events = Channel<ServerSentEvent>(Channel.BUFFERED)
205 val producer = scope.launch {
206 try {
207 incoming.collect { events.send(it) }
208 events.close()
209 } catch (c: CancellationException) {
210 events.close(c)
211 throw c
212 } catch (t: Throwable) {
213 // Deliver the failure via the channel rather than letting it
214 // propagate up and cancel the parent channelFlow.
215 events.close(t)
216 }
217 }
218 // Reset the backoff only once real data actually arrives, not on
219 // block entry: a server that accepts the connection then immediately
220 // drops it would otherwise loop forever at the initial backoff and
221 // never escalate the exponential ramp.
222 var receivedAny = false
223 try {
224 readSseEvents(scope, events) {
225 if (!receivedAny) {
226 receivedAny = true
227 backoffMs = initialBackoffMs
228 }
229 send(it)
230 }
231 } finally {
232 producer.cancel()
233 events.cancel()
234 }
235 }
236 } catch (c: CancellationException) {
237 _state.value = ConnectionState.Disconnected
238 throw c
239 } catch (e: Exception) {
240 // If the scope was cancelled (e.g. connection close), the closed-client
241 // exception is expected — don't log a spurious "stream error" warning.
242 if (isNonRetryableSseFailure(e)) {
243 // The server rejected the request with a non-retryable 4xx (bad
244 // credentials, missing endpoint, ...) — retrying won't help. Log a
245 // scrubbed summary (class + status) instead of the full exception, whose
246 // message carries the request URL and may include auth or paths.
247 Log.w("EventStream", "SSE non-retryable 4xx, awaiting explicit reconnect: ${safeExceptionSummary(e)}")
248 _state.value = ConnectionState.Failed
249 if (!isActive) break
250 // Don't complete the flow: a subscriber may fix the profile and call
251 // triggerReconnect(). Suspend until an explicit reconnect signal
252 // arrives (no backoff delay), then re-attempt the connection. This receive
253 // is outside the try above, so reset state to Disconnected if the last
254 // subscriber leaves (cancellation) while we're parked here — otherwise the
255 // stream would be torn down with state stuck at Failed.
256 try {
257 reconnectSignal.receive()
258 } catch (c: CancellationException) {
259 _state.value = ConnectionState.Disconnected
260 throw c
261 }
262 continue
263 } else if (isActive) {
264 Log.w("EventStream", "SSE stream error, will retry: ${safeExceptionSummary(e)}")
265 }
266 }
267 _state.value = ConnectionState.Disconnected
268 if (!isActive) break
269 // Wait for the backoff, but allow a reconnect signal to cut it short.
270 var signaled = false
271 val jitter = ((backoffMs * NetworkConfig.retryJitterFactor) * (Random.nextDouble() * 2 - 1)).toLong()
272 select {
273 reconnectSignal.onReceive { signaled = true }
274 onTimeout((backoffMs + jitter).coerceAtLeast(0)) { /* normal backoff elapsed */ }
275 }
276 backoffMs = if (signaled) initialBackoffMs else (backoffMs * 2).coerceAtMost(maxBackoffMs)
277 }
278 }