git.iko.soy/opencode-android
on 8bc2cce help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/network/EventStreamClient.kt
226 lines 12 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.produceIn
25import kotlinx.coroutines.flow.shareIn
26import kotlinx.coroutines.isActive
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() ?: break
112 val data = sse.data
113 if (data == null) {
114 // A comment or keep-alive event (no data field) resets the idle watchdog
115 // — servers send `: ping` comments to keep the connection alive through
116 // proxies. Without this, a server that only sends comments would still
117 // hit the idle timeout and reconnect unnecessarily.
118 continue
119 }
120 val event = try {
121 OpencodeJson.decodeFromString(BusEvent.serializer(), data)
122 } catch (e: kotlinx.serialization.SerializationException) {
123 Log.d("EventStream", "Skipping unparseable SSE event", e)
124 continue
125 }
126 if (_state.value != ConnectionState.Connected) {
127 _state.value = ConnectionState.Connected
128 }
129 send(event)
130 }
131 }
133 private fun stream(): Flow<BusEvent> = channelFlow {
134 val scope = this
135 // Drain any stale reconnect signal from a previous collection cycle so it
136 // doesn't spuriously reset the backoff on startup (WhileSubscribed may have
137 // stopped the upstream while triggerReconnect() queued a signal).
138 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
139 var backoffMs = initialBackoffMs
140 while (isActive) {
141 _state.value = ConnectionState.Connecting
142 try {
143 // The SSE stream is long-lived: disable the request-level and socket
144 // timeouts (the client default is 60s for REST) in the request config
145 // so the connection isn't killed mid-stream. The idle-timeout watchdog
146 // below handles half-open connections instead.
147 client.sse(
148 "event",
149 request = {
150 timeout {
151 requestTimeoutMillis = Long.MAX_VALUE
152 // Disable the socket (read) timeout entirely for the SSE stream.
153 // A finite value would kill the connection whenever the server
154 // is quiet for longer than that (idle sessions, slow agent runs),
155 // causing a reconnect storm. Half-open detection is handled by
156 // the idle-timeout watchdog below plus OkHttp's pingInterval on
157 // HTTP/2 connections.
158 socketTimeoutMillis = Long.MAX_VALUE
159 }
160 },
161 ) {
162 // The SSE block is open — the connection is established. Mark as
163 // Connected immediately so downstream re-seed logic fires even when
164 // the server only sends keep-alive comments (no data events).
165 if (_state.value != ConnectionState.Connected) {
166 _state.value = ConnectionState.Connected
167 }
168 // Clear any stale reconnect signal from a triggerReconnect() call
169 // made while the previous stream was healthy, so it doesn't suppress
170 // the next legitimate backoff.
171 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
172 // `incoming` is a cold Flow; bridge it to a ReceiveChannel so each
173 // element can be raced against an idle timeout via select.
174 val events = incoming.produceIn(scope)
175 // Reset the backoff only once real data actually arrives, not on
176 // block entry: a server that accepts the connection then immediately
177 // drops it would otherwise loop forever at the initial backoff and
178 // never escalate the exponential ramp.
179 var receivedAny = false
180 try {
181 readSseEvents(scope, events) {
182 if (!receivedAny) {
183 receivedAny = true
184 backoffMs = initialBackoffMs
185 }
186 send(it)
187 }
188 } finally {
189 events.cancel()
190 }
191 }
192 } catch (c: CancellationException) {
193 _state.value = ConnectionState.Disconnected
194 throw c
195 } catch (e: Exception) {
196 // If the scope was cancelled (e.g. connection close), the closed-client
197 // exception is expected — don't log a spurious "stream error" warning.
198 if (isSseAuthFailure(e)) {
199 // Credentials are wrong — retrying won't help. Log a scrubbed
200 // summary (class + status) instead of the full exception, whose
201 // message carries the request URL and may include auth or paths.
202 Log.w("EventStream", "SSE auth failed (401/403), awaiting explicit reconnect: ${safeExceptionSummary(e)}")
203 _state.value = ConnectionState.Failed
204 if (!isActive) break
205 // Don't complete the flow: a subscriber may fix the profile and call
206 // triggerReconnect(). Suspend until an explicit reconnect signal
207 // arrives (no backoff delay), then re-attempt the connection.
208 reconnectSignal.receive()
209 continue
210 } else if (isActive) {
211 Log.w("EventStream", "SSE stream error, will retry: ${safeExceptionSummary(e)}")
212 }
213 }
214 _state.value = ConnectionState.Disconnected
215 if (!isActive) break
216 // Wait for the backoff, but allow a reconnect signal to cut it short.
217 var signaled = false
218 val jitter = ((backoffMs * NetworkConfig.retryJitterFactor) * (Random.nextDouble() * 2 - 1)).toLong()
219 select {
220 reconnectSignal.onReceive { signaled = true }
221 onTimeout((backoffMs + jitter).coerceAtLeast(0)) { /* normal backoff elapsed */ }
222 }
223 backoffMs = if (signaled) initialBackoffMs else (backoffMs * 2).coerceAtMost(maxBackoffMs)
224 }
225 }