@@ -97,10 +97,16 @@ open class EventStreamClient(
9797 * Read SSE events from [events], racing each read against an idle-timeout watchdog.
9898 * Throws [IOException] on idle timeout so the outer reconnect loop fires; breaks on
9999 * 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.
100105 */
101106 private suspend fun readSseEvents(
102107 scope: CoroutineScope,
103108 events: kotlinx.coroutines.channels.ReceiveChannel<ServerSentEvent>,
109+ onReconnectConsumed: () -> Unit,
104110 send: suspend (BusEvent) -> Unit,
105111 ) {
106112 while (scope.isActive) {
@@ -108,17 +114,20 @@ open class EventStreamClient(
108114 // read), so a socket the peer never closed would otherwise hang until OS TCP
109115 // keepalive kicks in — minutes to hours on stock Linux. If nothing arrives
110116 // within the idle timeout, treat the connection as half-open and drop it.
117+ var consumedReconnectSignal = false
111118 val result: ChannelResult<ServerSentEvent>? = select {
112119 events.onReceiveCatching { it }
113120 onTimeout(idleTimeoutMs) { null }
114121 // A triggerReconnect() issued while a read is active must interrupt it now:
115122 // return the same null sentinel as the idle timeout so a half-open socket
116123 // (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 }
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")
120130 }
121− if (result == null) throw IOException("SSE read dropped (idle timeout or reconnect requested), reconnecting")
122131 val sse = result.getOrNull()
123132 if (sse == null) {
124133 // The producer channel is closed. A non-null cause means `incoming`
@@ -158,6 +167,12 @@ open class EventStreamClient(
158167 var backoffMs = initialBackoffMs
159168 while (isActive) {
160169 _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
161176 try {
162177 // The SSE stream is long-lived: disable the request-level and socket
163178 // timeouts (the client default is 60s for REST) in the request config
@@ -221,7 +236,7 @@ open class EventStreamClient(
221236 // never escalate the exponential ramp.
222237 var receivedAny = false
223238 try {
224− readSseEvents(scope, events) {
239+ readSseEvents(scope, events, { reconnectConsumedDuringRead = true }) {
225240 if (!receivedAny) {
226241 receivedAny = true
227242 backoffMs = initialBackoffMs
@@ -272,14 +287,30 @@ open class EventStreamClient(
272287 }
273288 _state.value = ConnectionState.Disconnected
274289 if (!isActive) break
275− // Wait for the backoff, but allow a reconnect signal to cut it short.
276− var signaled = false
277− val jitter = ((backoffMs * NetworkConfig.retryJitterFactor) * (Random.nextDouble() * 2 - 1)).toLong()
278− select {
279− reconnectSignal.onReceive { signaled = true }
280− onTimeout((backoffMs + jitter).coerceAtLeast(0)) { /* normal backoff elapsed */ }
281− }
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)
282296 backoffMs = if (signaled) initialBackoffMs else (backoffMs * 2).coerceAtMost(maxBackoffMs)
283297 }
284298 }
299+
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+ }
285316}