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