diff --git a/.changeset/quiet-pandas-marry.md b/.changeset/quiet-pandas-marry.md new file mode 100644 index 00000000..5c52144c --- /dev/null +++ b/.changeset/quiet-pandas-marry.md @@ -0,0 +1,18 @@ +--- +"client-sdk-android": patch +--- + +Fix the subscriber silently buffering remote ICE candidates after a resume + +A soft reconnect put the subscriber into an ice restart state, but only `setRemoteDescription` +clears that and the server re-offers the subscriber only when the reconnect moved the participant +to another node. After an ordinary resume no offer arrives, so the state stayed set for the life of +the transport and every later remote candidate was queued instead of applied, leaving the +subscriber unable to adopt any new path the server proposed. + +Candidates now wait on the description they belong to rather than on the reconnect. `onServerOffer` +says a description is coming before it schedules the work that applies it, so a candidate that +arrives in between is held rather than tried against the description being replaced. Each wait is +owned by what answers it, an ice restart offer by its own offer id and a server offer by its own +turn, and it ends when that attempt ends whether the description lands, is refused, or answers an +offer that has since been superseded. diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/PeerConnectionTransport.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/PeerConnectionTransport.kt index efae279a..e93f32ef 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/PeerConnectionTransport.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/PeerConnectionTransport.kt @@ -89,7 +89,16 @@ constructor( ) ?: throw IllegalStateException("peer connection creation failed?") }!! private val pendingCandidates = mutableListOf() - private var restartingIce: Boolean = false + + /** + * The offers whose answers are still owed, by offer id, and how many server offers are still to + * be applied. Candidates wait while either stands, so one belonging to a description that has + * not landed is never tried against the description it replaces. Only touched on the RTC thread. + */ + private val awaitedAnswers = mutableSetOf() + private var awaitedOffers = 0 + + private fun awaitingDescription() = awaitedAnswers.isNotEmpty() || awaitedOffers > 0 private var renegotiate = false @@ -104,7 +113,7 @@ constructor( fun addIceCandidate(candidate: IceCandidate) { executeRTCIfNotClosed { - if (peerConnection.remoteDescription != null && !restartingIce) { + if (peerConnection.remoteDescription != null && !awaitingDescription()) { peerConnection.addIceCandidate(candidate) } else { pendingCandidates.add(candidate) @@ -122,16 +131,13 @@ constructor( val result = launchRTCIfNotClosed { val currentOfferId = latestOfferId.get() if (sd.type == SessionDescription.Type.ANSWER && currentOfferId > 0 && offerId > 0 && currentOfferId > offerId) { + // The offer this answers has been superseded, so only that offer's wait ends here; + // the offer that replaced it keeps its own. + descriptionSettled(sd, offerId, applied = false) return@launchRTCIfNotClosed Either.Right("Old offer, ignoring. Expected: $currentOfferId, actual: $offerId") } val result = peerConnection.setRemoteDescription(sd) - if (result is Either.Left) { - pendingCandidates.forEach { pending -> - peerConnection.addIceCandidate(pending) - } - pendingCandidates.clear() - restartingIce = false - } + descriptionSettled(sd, offerId, applied = result is Either.Left) return@launchRTCIfNotClosed result } ?: Either.Right("PCT is closed.") @@ -167,7 +173,6 @@ constructor( constraints.findConstraint(MediaConstraintKeys.ICE_RESTART) == MediaConstraintKeys.TRUE if (iceRestart) { LKLog.d { "restarting ice" } - restartingIce = true } if (peerConnection.signalingState() == SignalingState.HAVE_LOCAL_OFFER) { @@ -191,10 +196,18 @@ constructor( // this may skip some ids, but is not an issue. offerId = latestOfferId.incrementAndGet() + // An ice restart mints new local credentials, so candidates wait for the answer + // that carries the far side's. The wait is owned by this offer id alone. + if (iceRestart) { + awaitedAnswers.add(offerId) + } + val sdpOffer = when (val outcome = peerConnection.createOffer(constraints)) { is Either.Left -> outcome.value is Either.Right -> { LKLog.d { "error creating offer: ${outcome.value}" } + // No offer goes out, so the answer it would have waited for is not coming. + awaitedAnswers.remove(offerId) return@launchRTCIfNotClosed } } @@ -298,8 +311,40 @@ constructor( return sdp } - fun prepareForIceRestart() { - restartingIce = true + /** + * Says a remote description is on its way, so candidates for it wait rather than land against + * the description it replaces. Called before the work that applies it is scheduled, and + * answered by the attempt to set that description whether it lands or is refused. + */ + fun expectRemoteDescription() { + executeRTCIfNotClosed { awaitedOffers++ } + } + + /** + * Ends the wait this description answers, however it ended: an answer ends the wait its own + * offer opened and no other, a server offer ends one of the waits opened for them. Only a + * description that landed takes the candidates held, and only once nothing else is awaited. A + * wait that ended without one leaves them queued rather than dropping them, since one that no + * longer fits is refused by the connection anyway and one that still fits would be lost. + */ + private fun descriptionSettled(sd: SessionDescription, offerId: Int, applied: Boolean) { + if (sd.type == SessionDescription.Type.ANSWER) { + // A legacy answer carries no id to match itself to, so it ends whatever was owed. + if (offerId > 0) { + awaitedAnswers.remove(offerId) + } else { + awaitedAnswers.clear() + } + } else if (awaitedOffers > 0) { + awaitedOffers-- + } + if (!applied || awaitingDescription()) { + return + } + pendingCandidates.forEach { pending -> + peerConnection.addIceCandidate(pending) + } + pendingCandidates.clear() } fun isClosed() = isClosed.get() diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt index e0e64e7d..3c04ff34 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt @@ -637,7 +637,10 @@ internal constructor( } connectionState = ConnectionState.RESUMING LKLog.v { "Attempting soft reconnect." } - subscriber?.prepareForIceRestart() + // The subscriber is not told to hold candidates here. The server re-offers it + // only when the reconnect moved us to another node, so a resume that stays put + // would never clear the hold. onServerOffer sets it instead, which covers + // exactly the offer it arrives with and nothing after. try { val response = client.reconnect(url!!, token, participantSid) if (response is Either.Left) { @@ -1128,6 +1131,9 @@ internal constructor( override fun onServerOffer(sessionDescription: SessionDescription, offerId: Int) { LKLog.v { "received server offer: ${sessionDescription.type}, ${runBlocking { publisher?.signalingState() }}" } + // Set before the work is scheduled: responses keep their order but this coroutine does not + // run inline, so a candidate for this offer can arrive while the old description is still on. + subscriber?.expectRemoteDescription() coroutineScope.launch { run { when (val outcome = subscriber?.setRemoteDescription(sessionDescription, offerId).nullSafe()) { diff --git a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt index 6a51cfeb..28c2e1d2 100644 --- a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt +++ b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt @@ -129,7 +129,11 @@ class MockPeerConnection( return true } + /** Every candidate actually handed to the connection, so a test can tell added from queued. */ + val addedIceCandidates = mutableListOf() + override fun addIceCandidate(candidate: IceCandidate?): Boolean { + candidate?.let { addedIceCandidates.add(it) } return true } diff --git a/livekit-android-test/src/test/java/io/livekit/android/room/RTCEngineMockE2ETest.kt b/livekit-android-test/src/test/java/io/livekit/android/room/RTCEngineMockE2ETest.kt index 6225fb1e..b1de4f88 100644 --- a/livekit-android-test/src/test/java/io/livekit/android/room/RTCEngineMockE2ETest.kt +++ b/livekit-android-test/src/test/java/io/livekit/android/room/RTCEngineMockE2ETest.kt @@ -683,4 +683,150 @@ class RTCEngineMockE2ETest : MockE2ETest() { ) } } + + /** + * A resume must leave the subscriber able to take the paths the server proposes. It used to + * enter an ice restart state that only a remote description clears, and the server re-offers + * the subscriber only when the reconnect moved us to another node, so after an ordinary resume + * every later candidate was queued and never added. Same defect as client-sdk-js#2054. + */ + @Test + fun softReconnectKeepsSubscriberApplyingRemoteCandidates() = runTest { + room.setReconnectionType(ReconnectType.FORCE_SOFT_RECONNECT) + connect() + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + wsFactory.listener.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request)) + simulateMessageFromServer(TestData.RECONNECT) + connectPeerConnection() + advanceUntilIdle() + + val subPeerConnection = getSubscriberPeerConnection() + val before = subPeerConnection.addedIceCandidates.size + simulateMessageFromServer(subscriberTrickle()) + advanceUntilIdle() + + assertEquals(before + 1, subPeerConnection.addedIceCandidates.size) + } + + /** + * A candidate belonging to a server offer waits for that offer to be applied. Offer handling is + * launched rather than run inline, so without the hold the candidate would land against the + * description the offer replaces and be rejected against its ice credentials. + */ + @Test + fun candidateArrivingWithAServerOfferWaitsForItsDescription() = runTest { + connect() + + val subPeerConnection = getSubscriberPeerConnection() + val before = subPeerConnection.addedIceCandidates.size + simulateMessageFromServer(TestData.OFFER) + simulateMessageFromServer(subscriberTrickle()) + + assertEquals(before, subPeerConnection.addedIceCandidates.size) + + advanceUntilIdle() + + assertEquals(before + 1, subPeerConnection.addedIceCandidates.size) + } + + /** + * A refused offer ends its own wait. The description it would have installed never came into + * being, so the connection carries on with the one in force and later candidates reach it + * rather than piling up behind an offer that is never going to land. + */ + @Test + fun aRefusedOfferStopsHoldingCandidates() = runTest { + connect() + val subPeerConnection = getSubscriberPeerConnection() + + simulateMessageFromServer(refusedOffer()) + advanceUntilIdle() + + val before = subPeerConnection.addedIceCandidates.size + simulateMessageFromServer(subscriberTrickle()) + advanceUntilIdle() + + assertEquals(before + 1, subPeerConnection.addedIceCandidates.size) + } + + /** + * A stale answer ends only its own offer's wait. An ordinary offer opens none, so an answer to + * it arriving late must not release the wait an ice restart opened afterwards, or candidates + * for the restart land against the credentials it replaced. + */ + @Test + fun aStaleAnswerDoesNotEndAnotherOffersWait() = runTest { + room.setReconnectionType(ReconnectType.FORCE_SOFT_RECONNECT) + connect() + + // Offers go unanswered from here, so the restart's wait is still outstanding below. + val heldAnswers: SignalRequestHandler = { request -> request.hasOffer() } + wsFactory.registerSignalRequestHandler(heldAnswers) + val pubPeerConnection = getPublisherPeerConnection() + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + wsFactory.listener.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request)) + simulateMessageFromServer(TestData.RECONNECT) + connectPeerConnection() + advanceUntilIdle() + + val before = pubPeerConnection.addedIceCandidates.size + simulateMessageFromServer(staleAnswer()) + simulateMessageFromServer(publisherTrickle()) + advanceUntilIdle() + + assertEquals(before, pubPeerConnection.addedIceCandidates.size) + wsFactory.unregisterSignalRequestHandler(heldAnswers) + } + + /** An answer to the first offer, long since replaced by the ones after it. */ + private fun staleAnswer(): LivekitRtc.SignalResponse { + val answer = LivekitRtc.SessionDescription.newBuilder() + .setSdp("remote_answer") + .setType("answer") + .setId(1) + .build() + return LivekitRtc.SignalResponse.newBuilder() + .setAnswer(answer) + .build() + } + + private fun publisherTrickle(): LivekitRtc.SignalResponse { + val trickle = LivekitRtc.TrickleRequest.newBuilder() + .setCandidateInit( + """{"candidate":"candidate:2 1 UDP 1 127.0.0.1 9 typ host","sdpMLineIndex":0,"sdpMid":"0"}""", + ) + .setTarget(LivekitRtc.SignalTarget.PUBLISHER) + .build() + return LivekitRtc.SignalResponse.newBuilder() + .setTrickle(trickle) + .build() + } + + /** An empty description is the one the mock connection refuses. */ + private fun refusedOffer(): LivekitRtc.SignalResponse { + val offer = LivekitRtc.SessionDescription.newBuilder() + .setSdp("") + .setType("offer") + .setId(100) + .build() + return LivekitRtc.SignalResponse.newBuilder() + .setOffer(offer) + .build() + } + + private fun subscriberTrickle(): LivekitRtc.SignalResponse { + val trickle = LivekitRtc.TrickleRequest.newBuilder() + .setCandidateInit( + """{"candidate":"candidate:1 1 UDP 1 127.0.0.1 9 typ host","sdpMLineIndex":0,"sdpMid":"0"}""", + ) + .setTarget(LivekitRtc.SignalTarget.SUBSCRIBER) + .build() + return LivekitRtc.SignalResponse.newBuilder() + .setTrickle(trickle) + .build() + } }