Skip to content

Commit 0e2a94b

Browse files
YashJainSCclaude
YashJainSC
and
claude
authored
fix: prevent reconnect from hanging when server sends LEAVE or never responds (livekit#934)
* fix: resume joinContinuation when LEAVE received during reconnect handshake When the server sends a LEAVE as the initial response to a soft reconnect (e.g. STATE_MISMATCH), handleSignalResponse processed it without resuming joinContinuation, leaving client.reconnect() suspended forever. This prevented the reconnect loop from advancing to a full reconnect. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix: add timeout on signal connect to prevent reconnect hang Add a 10s timeout on the signal connect handshake so the reconnect loop can advance when the server never sends a join/reconnect response. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * added changeset * fix: use withDeadline instead of withTimeout to pass detekt Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * style: spotlessApply Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix: update reconnect tests to complete signal handshake before clock advance Tests using advanceUntilIdle() now advance past the signal connect timeout, causing the reconnect loop to exhaust retries. Fix by completing the reconnect handshake before advancing the clock. Also fix softReconnectResendsPackets to properly test resend behavior by adding a publisher offer handler. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * style: spotlessApply Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent 3e8034a commit 0e2a94b

6 files changed

Lines changed: 131 additions & 15 deletions

File tree

‎.changeset/neat-rabbits-draw.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"client-sdk-android": patch
3+
---
4+
5+
fix: resume joinContinuation when LEAVE received during reconnect handshake to avoid reconnection loop hanging issue

‎livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt‎

Lines changed: 20 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ import io.livekit.android.util.Either
3030
import io.livekit.android.util.LKLog
3131
import io.livekit.android.util.toHttpUrl
3232
import io.livekit.android.util.toWebsocketUrl
33+
import io.livekit.android.util.withDeadline
3334
import io.livekit.android.webrtc.toProtoSessionDescription
3435
import kotlinx.coroutines.CancellableContinuation
3536
import kotlinx.coroutines.CoroutineDispatcher
@@ -63,6 +64,7 @@ import javax.inject.Inject
6364
import javax.inject.Named
6465
import javax.inject.Singleton
6566
import kotlin.coroutines.resumeWithException
67+
import kotlin.time.Duration.Companion.milliseconds
6668

6769
/**
6870
* SignalClient to LiveKit WS servers
@@ -186,17 +188,19 @@ constructor(
186188
.addHeader("Authorization", "Bearer $token")
187189
.build()
188190

189-
return suspendCancellableCoroutine { cont ->
190-
// Wait for join response through WebSocketListener
191-
joinContinuation = cont
192-
cont.invokeOnCancellation {
193-
// If the coroutine is cancelled, websocket needs to be cancelled.
194-
// onFailure will handle cleanup.
195-
LKLog.v { "connect cancelled, abort websocket" }
196-
joinContinuation = null
197-
currentWs?.cancel()
191+
return withDeadline(SIGNAL_CONNECT_TIMEOUT.milliseconds) {
192+
suspendCancellableCoroutine { cont ->
193+
// Wait for join response through WebSocketListener
194+
joinContinuation = cont
195+
cont.invokeOnCancellation {
196+
// If the coroutine is cancelled, websocket needs to be cancelled.
197+
// onFailure will handle cleanup.
198+
LKLog.v { "connect cancelled, abort websocket" }
199+
joinContinuation = null
200+
currentWs?.cancel()
201+
}
202+
currentWs = websocketFactory.newWebSocket(request, this@SignalClient)
198203
}
199-
currentWs = websocketFactory.newWebSocket(request, this@SignalClient)
200204
}
201205
}
202206

@@ -691,6 +695,11 @@ constructor(
691695
} else if (response.hasLeave()) {
692696
// Some reconnects may immediately send leave back without a join response first.
693697
handleSignalResponseImpl(ws, response)
698+
val cont = joinContinuation
699+
joinContinuation = null
700+
cont?.resumeWithException(
701+
RoomException.ConnectException("Received leave during reconnect: ${response.leave.reason}"),
702+
)
694703
} else if (isReconnecting) {
695704
// When reconnecting, any message received means signal reconnected.
696705
// Newer servers will send a reconnect response first
@@ -977,6 +986,7 @@ constructor(
977986
// iceServer("stun:stun3.l.google.com:19302"),
978987
// iceServer("stun:stun4.l.google.com:19302"),
979988
)
989+
private const val SIGNAL_CONNECT_TIMEOUT = 10000
980990
const val CLOSE_REASON_NORMAL_CLOSURE = 1000
981991
const val CLOSE_REASON_PING_TIMEOUT = 3000
982992
const val CLOSE_REASON_WEBSOCKET_FAILURE = 3500

‎livekit-android-test/src/test/java/io/livekit/android/room/RTCEngineMockE2ETest.kt‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -408,7 +408,10 @@ class RTCEngineMockE2ETest : MockE2ETest() {
408408
connect()
409409
wsFactory.listener.onFailure(wsFactory.ws, Exception(), null)
410410

411-
testScheduler.advanceUntilIdle()
411+
testScheduler.advanceTimeBy(1000)
412+
wsFactory.listener.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request))
413+
simulateMessageFromServer(TestData.RECONNECT)
414+
412415
val sid = wsFactory.request.url.queryParameter(SignalClient.CONNECT_QUERY_PARTICIPANT_SID)
413416
assertEquals(TestData.JOIN.join.participant.sid, sid)
414417
}

‎livekit-android-test/src/test/java/io/livekit/android/room/RoomMockE2ETest.kt‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2023-2025 LiveKit, Inc.
2+
* Copyright 2023-2026 LiveKit, Inc.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -345,13 +345,17 @@ class RoomMockE2ETest : MockE2ETest() {
345345
callback.onAvailable(network)
346346
}
347347

348-
coroutineRule.dispatcher.scheduler.advanceUntilIdle()
348+
coroutineRule.dispatcher.scheduler.advanceTimeBy(1000)
349+
wsFactory.listener.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request))
350+
simulateMessageFromServer(TestData.RECONNECT)
351+
349352
val events = eventCollector.stopCollecting()
350353

351354
assertEquals(
352355
listOf(
353356
ConnectionState.CONNECTED,
354357
ConnectionState.RESUMING,
358+
ConnectionState.CONNECTED,
355359
),
356360
events,
357361
)

‎livekit-android-test/src/test/java/io/livekit/android/room/RoomReconnectionMockE2ETest.kt‎

Lines changed: 24 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2023-2025 LiveKit, Inc.
2+
* Copyright 2023-2026 LiveKit, Inc.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -24,6 +24,7 @@ import io.livekit.android.test.mock.MockDataChannel
2424
import io.livekit.android.test.mock.MockMediaStream
2525
import io.livekit.android.test.mock.MockRtpReceiver
2626
import io.livekit.android.test.mock.MockVideoStreamTrack
27+
import io.livekit.android.test.mock.SignalRequestHandler
2728
import io.livekit.android.test.mock.TestData
2829
import io.livekit.android.test.mock.createMediaStreamId
2930
import io.livekit.android.test.mock.room.track.createMockLocalAudioTrack
@@ -197,8 +198,28 @@ class RoomReconnectionMockE2ETest : MockE2ETest() {
197198
fun softReconnectResendsPackets() = runTest {
198199
room.setReconnectionType(ReconnectType.FORCE_SOFT_RECONNECT)
199200

201+
val publisherOfferHandler: SignalRequestHandler = { request ->
202+
if (request.hasOffer()) {
203+
val answer = with(LivekitRtc.SignalResponse.newBuilder()) {
204+
answer = with(LivekitRtc.SessionDescription.newBuilder()) {
205+
sdp = "remote_answer"
206+
type = "answer"
207+
id = request.offer.id
208+
build()
209+
}
210+
build()
211+
}
212+
wsFactory.receiveMessage(answer)
213+
true
214+
} else {
215+
false
216+
}
217+
}
218+
wsFactory.registerSignalRequestHandler(publisherOfferHandler)
200219
connect()
201220

221+
val lastMessageSeq = TestData.RECONNECT.reconnect.lastMessageSeq
222+
202223
for (i in 1..5) {
203224
assertTrue(room.localParticipant.publishData(ByteArray(i), reliability = DataPublishReliability.RELIABLE).isSuccess)
204225
}
@@ -212,7 +233,8 @@ class RoomReconnectionMockE2ETest : MockE2ETest() {
212233
val pubPeerConnection = getPublisherPeerConnection()
213234
val pubDataChannel = pubPeerConnection.dataChannels[RTCEngine.RELIABLE_DATA_CHANNEL_LABEL] as MockDataChannel
214235

215-
assertEquals(5, pubDataChannel.sentBuffers.size)
236+
val expectedResentCount = (1..5).count { it > lastMessageSeq }
237+
assertEquals(5 + expectedResentCount, pubDataChannel.sentBuffers.size)
216238
}
217239

218240
@Test

‎livekit-android-test/src/test/java/io/livekit/android/room/SignalClientTest.kt‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ import kotlinx.coroutines.async
3333
import kotlinx.coroutines.supervisorScope
3434
import kotlinx.coroutines.yield
3535
import kotlinx.serialization.json.Json
36+
import livekit.LivekitModels
3637
import livekit.LivekitRtc
3738
import livekit.org.webrtc.SessionDescription
3839
import okhttp3.OkHttpClient
@@ -429,6 +430,77 @@ class SignalClientTest : BaseTest() {
429430
assertTrue(originalWs.isClosed)
430431
}
431432

433+
/**
434+
* Reconnect fails cleanly when server sends LEAVE as the initial response and closes WS cleanly.
435+
*/
436+
@Test
437+
fun reconnectDoesNotHangOnLeaveStateMismatchWithCleanClose() = runTest {
438+
val joinJob = async { client.join(EXAMPLE_URL, "") }
439+
connectWebsocketAndJoin()
440+
joinJob.await()
441+
442+
wsFactory.ws.cancel()
443+
444+
supervisorScope {
445+
val reconnectJob = async {
446+
client.reconnect(EXAMPLE_URL, "", "participant_sid")
447+
}
448+
yield()
449+
450+
client.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request))
451+
val leaveResponse = with(LivekitRtc.SignalResponse.newBuilder()) {
452+
leave = with(LivekitRtc.LeaveRequest.newBuilder()) {
453+
reason = LivekitModels.DisconnectReason.STATE_MISMATCH
454+
action = LivekitRtc.LeaveRequest.Action.RECONNECT
455+
build()
456+
}
457+
build()
458+
}
459+
wsFactory.receiveMessage(leaveResponse)
460+
461+
wsFactory.ws.close(1000, "normal")
462+
463+
val result = runCatching { reconnectJob.await() }
464+
assertTrue("reconnect should have failed", result.isFailure)
465+
}
466+
}
467+
468+
/**
469+
* Reconnect fails cleanly when server sends LEAVE as the initial response and drops the connection.
470+
*/
471+
@Test
472+
fun reconnectDoesNotHangOnLeaveStateMismatchWithFailure() = runTest {
473+
val joinJob = async { client.join(EXAMPLE_URL, "") }
474+
connectWebsocketAndJoin()
475+
joinJob.await()
476+
477+
wsFactory.ws.cancel()
478+
479+
supervisorScope {
480+
val reconnectJob = async {
481+
client.reconnect(EXAMPLE_URL, "", "participant_sid")
482+
}
483+
yield()
484+
485+
client.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request))
486+
val leaveResponse = with(LivekitRtc.SignalResponse.newBuilder()) {
487+
leave = with(LivekitRtc.LeaveRequest.newBuilder()) {
488+
reason = LivekitModels.DisconnectReason.STATE_MISMATCH
489+
action = LivekitRtc.LeaveRequest.Action.RECONNECT
490+
build()
491+
}
492+
build()
493+
}
494+
wsFactory.receiveMessage(leaveResponse)
495+
496+
// 5. Server drops connection (unclean close)
497+
wsFactory.ws.cancel()
498+
499+
val result = runCatching { reconnectJob.await() }
500+
assertTrue("reconnect should have failed", result.isFailure)
501+
}
502+
}
503+
432504
// mock data
433505
companion object
434506
}

0 commit comments

Comments
 (0)