git.iko.soy/opencode-android
on bc6b45e help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/network/EventStreamClient.kt
214 lines 11 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 private fun isSseAuthFailure(e: Throwable): Boolean =
73 e is io.ktor.client.plugins.ClientRequestException &&
74 (e.response.status.value == 401 || e.response.status.value == 403)
76 /** Request an immediate reconnect, skipping any in-progress backoff. */
77 open fun triggerReconnect() { reconnectSignal.trySend(Unit) }
79 /**
80 * Read SSE events from [events], racing each read against an idle-timeout watchdog.
81 * Throws [IOException] on idle timeout so the outer reconnect loop fires; breaks on
82 * clean stream close or scope cancellation.
83 */
84 private suspend fun readSseEvents(
85 scope: CoroutineScope,
86 events: kotlinx.coroutines.channels.ReceiveChannel<ServerSentEvent>,
87 send: suspend (BusEvent) -> Unit,
88 ) {
89 while (scope.isActive) {
90 // The OkHttp engine has readTimeout=0 (the SSE stream must never time out on
91 // read), so a socket the peer never closed would otherwise hang until OS TCP
92 // keepalive kicks in — minutes to hours on stock Linux. If nothing arrives
93 // within the idle timeout, treat the connection as half-open and drop it.
94 val result: ChannelResult<ServerSentEvent>? = select {
95 events.onReceiveCatching { it }
96 onTimeout(idleTimeoutMs) { null }
97 }
98 if (result == null) throw IOException("SSE idle timeout, reconnecting")
99 val sse = result.getOrNull() ?: break
100 val data = sse.data
101 if (data == null) {
102 // A comment or keep-alive event (no data field) resets the idle watchdog
103 // — servers send `: ping` comments to keep the connection alive through
104 // proxies. Without this, a server that only sends comments would still
105 // hit the idle timeout and reconnect unnecessarily.
106 continue
107 }
108 val event = try {
109 OpencodeJson.decodeFromString(BusEvent.serializer(), data)
110 } catch (e: kotlinx.serialization.SerializationException) {
111 Log.d("EventStream", "Skipping unparseable SSE event", e)
112 continue
113 }
114 if (_state.value != ConnectionState.Connected) {
115 _state.value = ConnectionState.Connected
116 }
117 send(event)
118 }
119 }
121 private fun stream(): Flow<BusEvent> = channelFlow {
122 val scope = this
123 // Drain any stale reconnect signal from a previous collection cycle so it
124 // doesn't spuriously reset the backoff on startup (WhileSubscribed may have
125 // stopped the upstream while triggerReconnect() queued a signal).
126 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
127 var backoffMs = initialBackoffMs
128 while (isActive) {
129 _state.value = ConnectionState.Connecting
130 try {
131 // The SSE stream is long-lived: disable the request-level and socket
132 // timeouts (the client default is 60s for REST) in the request config
133 // so the connection isn't killed mid-stream. The idle-timeout watchdog
134 // below handles half-open connections instead.
135 client.sse(
136 "event",
137 request = {
138 timeout {
139 requestTimeoutMillis = Long.MAX_VALUE
140 // Disable the socket (read) timeout entirely for the SSE stream.
141 // A finite value would kill the connection whenever the server
142 // is quiet for longer than that (idle sessions, slow agent runs),
143 // causing a reconnect storm. Half-open detection is handled by
144 // the idle-timeout watchdog below plus OkHttp's pingInterval on
145 // HTTP/2 connections.
146 socketTimeoutMillis = Long.MAX_VALUE
147 }
148 },
149 ) {
150 // The SSE block is open — the connection is established. Mark as
151 // Connected immediately so downstream re-seed logic fires even when
152 // the server only sends keep-alive comments (no data events).
153 if (_state.value != ConnectionState.Connected) {
154 _state.value = ConnectionState.Connected
155 }
156 // Clear any stale reconnect signal from a triggerReconnect() call
157 // made while the previous stream was healthy, so it doesn't suppress
158 // the next legitimate backoff.
159 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
160 // `incoming` is a cold Flow; bridge it to a ReceiveChannel so each
161 // element can be raced against an idle timeout via select.
162 val events = incoming.produceIn(scope)
163 // Reset the backoff only once real data actually arrives, not on
164 // block entry: a server that accepts the connection then immediately
165 // drops it would otherwise loop forever at the initial backoff and
166 // never escalate the exponential ramp.
167 var receivedAny = false
168 try {
169 readSseEvents(scope, events) {
170 if (!receivedAny) {
171 receivedAny = true
172 backoffMs = initialBackoffMs
173 }
174 send(it)
175 }
176 } finally {
177 events.cancel()
178 }
179 }
180 } catch (c: CancellationException) {
181 _state.value = ConnectionState.Disconnected
182 throw c
183 } catch (e: Exception) {
184 // If the scope was cancelled (e.g. connection close), the closed-client
185 // exception is expected — don't log a spurious "stream error" warning.
186 if (isSseAuthFailure(e)) {
187 // Credentials are wrong — retrying won't help. Log a scrubbed
188 // summary (class + status) instead of the full exception, whose
189 // message carries the request URL and may include auth or paths.
190 Log.w("EventStream", "SSE auth failed (401/403), awaiting explicit reconnect: ${safeExceptionSummary(e)}")
191 _state.value = ConnectionState.Failed
192 if (!isActive) break
193 // Don't complete the flow: a subscriber may fix the profile and call
194 // triggerReconnect(). Suspend until an explicit reconnect signal
195 // arrives (no backoff delay), then re-attempt the connection.
196 reconnectSignal.receive()
197 continue
198 } else if (isActive) {
199 Log.w("EventStream", "SSE stream error, will retry: ${safeExceptionSummary(e)}")
200 }
201 }
202 _state.value = ConnectionState.Disconnected
203 if (!isActive) break
204 // Wait for the backoff, but allow a reconnect signal to cut it short.
205 var signaled = false
206 val jitter = ((backoffMs * NetworkConfig.retryJitterFactor) * (Random.nextDouble() * 2 - 1)).toLong()
207 select {
208 reconnectSignal.onReceive { signaled = true }
209 onTimeout((backoffMs + jitter).coerceAtLeast(0)) { /* normal backoff elapsed */ }
210 }
211 backoffMs = if (signaled) initialBackoffMs else (backoffMs * 2).coerceAtMost(maxBackoffMs)
212 }
213 }