From f961eb38484f1ae328505024878d81c447e801ce Mon Sep 17 00:00:00 2001 From: Adrian Niculescu <15037449+adrian-niculescu@users.noreply.github.com> Date: Thu, 13 Aug 2026 14:09:12 +0300 Subject: [PATCH] Fixed local publication jobs leaking on every full reconnect --- .../fix-reconnect-publication-job-leak.md | 5 + .../detekt-baseline-release.xml | 12 +- .../java/io/livekit/android/room/RTCEngine.kt | 42 +++- .../io/livekit/android/room/SignalClient.kt | 72 +++++-- .../room/participant/LocalParticipant.kt | 104 ++++++--- .../android/room/RTCEngineMockE2ETest.kt | 2 +- .../room/RoomReconnectionMockE2ETest.kt | 203 ++++++++++++++++++ 7 files changed, 379 insertions(+), 61 deletions(-) create mode 100644 .changeset/fix-reconnect-publication-job-leak.md diff --git a/.changeset/fix-reconnect-publication-job-leak.md b/.changeset/fix-reconnect-publication-job-leak.md new file mode 100644 index 000000000..3c1994a90 --- /dev/null +++ b/.changeset/fix-reconnect-publication-job-leak.md @@ -0,0 +1,5 @@ +--- +"client-sdk-android": patch +--- + +Fixed local track publications leaking their jobs on every full reconnect, which left the audio feature collectors of the old publications running and sending feature updates for stale track sids. diff --git a/livekit-android-sdk/detekt-baseline-release.xml b/livekit-android-sdk/detekt-baseline-release.xml index 54ebf273d..f1dbfe99b 100644 --- a/livekit-android-sdk/detekt-baseline-release.xml +++ b/livekit-android-sdk/detekt-baseline-release.xml @@ -10,7 +10,7 @@ ComplexCondition:RemoteTrackPublication.kt$RemoteTrackPublication$isAutoManaged || !subscribed || this.fps == fps || track !is VideoTrack ComplexCondition:RemoteTrackPublication.kt$RemoteTrackPublication$isAutoManaged || !subscribed || videoDimensions == dimensions || track !is VideoTrack ComplexCondition:TextureViewRenderer.kt$TextureViewRenderer$enableFixedSize && rotatedFrameWidth != 0 && rotatedFrameHeight != 0 && width != 0 && height != 0 - CyclomaticComplexMethod:LocalParticipant.kt$LocalParticipant$@Throws(TrackException.PublishException::class) private suspend fun publishTrackImpl( track: Track, options: TrackPublishOptions, requestConfig: AddTrackRequest.Builder.() -> Unit, encodings: List<RtpParameters.Encoding> = emptyList(), publishListener: PublishListener? = null, ): LocalTrackPublication? + CyclomaticComplexMethod:LocalParticipant.kt$LocalParticipant$@Throws(TrackException.PublishException::class) private suspend fun publishTrackImpl( track: Track, options: TrackPublishOptions, requestConfig: AddTrackRequest.Builder.() -> Unit, encodings: List<RtpParameters.Encoding> = emptyList(), publishListener: PublishListener? = null, ): PublishResult? CyclomaticComplexMethod:LocalParticipant.kt$LocalParticipant$private fun computeVideoEncodings( isScreenShare: Boolean, dimensions: Track.Dimensions, options: VideoTrackPublishOptions, ): List<RtpParameters.Encoding> CyclomaticComplexMethod:LocalParticipant.kt$LocalParticipant$private suspend fun setTrackEnabled( source: Track.Source, enabled: Boolean, screenCaptureParams: ScreenCaptureParams? = null, ): Boolean CyclomaticComplexMethod:LocalParticipant.kt$LocalParticipant$suspend fun publishVideoTrack( track: LocalVideoTrack, options: VideoTrackPublishOptions = VideoTrackPublishOptions( null, if (track.options.isScreencast) screenShareTrackPublishDefaults else videoTrackPublishDefaults, ), publishListener: PublishListener? = null, ): Boolean @@ -27,13 +27,13 @@ CyclomaticComplexMethod:Room.kt$Room$@Throws(Exception::class) suspend fun connect(url: String, token: String, options: ConnectOptions = ConnectOptions()) CyclomaticComplexMethod:RoomEvent.kt$fun LivekitModels.DisconnectReason?.convert(): DisconnectReason CyclomaticComplexMethod:SignalClient.kt$SignalClient$override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) - CyclomaticComplexMethod:SignalClient.kt$SignalClient$private fun handleSignalResponseImpl(ws: WebSocket, response: LivekitRtc.SignalResponse) + CyclomaticComplexMethod:SignalClient.kt$SignalClient$private fun handleSignalResponseImpl(connection: SignalConnection, response: LivekitRtc.SignalResponse) EmptyFunctionBlock:RTCEngine.kt$RTCEngine${ } HasPlatformType:DataChannelManager.kt$DataChannelManager$@get:FlowObservable var state by flowDelegate(dataChannel.state()) private set IgnoredReturnValue:RpcServerManager.kt$RpcServerManager$publishRpcAck(callerIdentity, requestId) InstanceOfCheckForException:RpcServerManager.kt$RpcServerManager$e is RpcError LargeClass:LocalParticipant.kt$LocalParticipant : ParticipantOutgoingDataStreamManagerRpcManager - LargeClass:RTCEngine.kt$RTCEngine : Listener + LargeClass:RTCEngine.kt$RTCEngine : ListenerSignalSessionListener LargeClass:Room.kt$Room : ListenerParticipantListenerRpcManagerIncomingDataStreamManager LargeClass:SignalClient.kt$SignalClient : WebSocketListener LongMethod:RTCEngine.kt$RTCEngine$@Synchronized @VisibleForTesting(otherwise = VisibleForTesting.PACKAGE_PRIVATE) fun reconnect() @@ -61,7 +61,7 @@ LongParameterList:Room.kt$Room$( @Assisted private val context: Context, internal val engine: RTCEngine, private val eglBase: EglBase, localParticipantFactory: LocalParticipant.Factory, private val defaultsManager: DefaultsManager, @Named(InjectionNames.DISPATCHER_DEFAULT) private val defaultDispatcher: CoroutineDispatcher, @Named(InjectionNames.DISPATCHER_IO) private val ioDispatcher: CoroutineDispatcher, /** * The [AudioHandler] for setting up the audio as need. * * By default, this is an instance of [AudioSwitchHandler]. * * This can be substituted for your own custom implementation through * [LiveKitOverrides.audioOptions] when creating the room with [LiveKit.create]. * * @see [audioSwitchHandler] * @see [AudioSwitchHandler] */ val audioHandler: AudioHandler, private val closeableManager: CloseableManager, private val e2EEManagerFactory: E2EEManager.Factory, private val communicationWorkaround: CommunicationWorkaround, val audioProcessingController: AudioProcessingController, /** * A holder for objects that are used internally within LiveKit. */ val lkObjects: LKObjects, networkCallbackManagerFactory: NetworkCallbackManagerFactory, private val audioDeviceModule: AudioDeviceModule, private val regionUrlProviderFactory: RegionUrlProvider.Factory, private val connectionWarmer: ConnectionWarmer, private val audioRecordPrewarmer: AudioRecordPrewarmer, private val incomingDataStreamManager: IncomingDataStreamManager, private val rpcClientManager: RpcClientManager, private val rpcServerManager: RpcServerManager, private val remoteParticipantFactory: RemoteParticipant.Factory, ) MapGetWithNotNullAssertionOperator:LocalParticipant.kt$LocalParticipant$sourcePubLocks[source]!! NestedBlockDepth:ByteStreamSender.kt$@CheckResult suspend fun ByteStreamSender.write(source: Source): Result<Unit> - NestedBlockDepth:LocalParticipant.kt$LocalParticipant$@Throws(TrackException.PublishException::class) private suspend fun publishTrackImpl( track: Track, options: TrackPublishOptions, requestConfig: AddTrackRequest.Builder.() -> Unit, encodings: List<RtpParameters.Encoding> = emptyList(), publishListener: PublishListener? = null, ): LocalTrackPublication? + NestedBlockDepth:LocalParticipant.kt$LocalParticipant$@Throws(TrackException.PublishException::class) private suspend fun publishTrackImpl( track: Track, options: TrackPublishOptions, requestConfig: AddTrackRequest.Builder.() -> Unit, encodings: List<RtpParameters.Encoding> = emptyList(), publishListener: PublishListener? = null, ): PublishResult? NestedBlockDepth:LocalParticipant.kt$LocalParticipant$fun cleanup() NestedBlockDepth:LocalVideoTrack.kt$LocalVideoTrack$internal fun setPublishingCodecs(codecs: List<SubscribedCodec>): List<VideoCodec> NestedBlockDepth:LocalVideoTrack.kt$LocalVideoTrack$private fun setPublishingLayersForSender( sender: RtpSender, qualities: List<LivekitRtc.SubscribedQuality>, ) @@ -71,7 +71,7 @@ NestedBlockDepth:RTCEngine.kt$RTCEngine$private fun makeRTCConfig( serverResponse: Either<JoinResponse, ReconnectResponse>, connectOptions: ConnectOptions, ): RTCConfiguration NestedBlockDepth:Room.kt$Room$override suspend fun onPostReconnect(isFullReconnect: Boolean) NestedBlockDepth:SignalClient.kt$SignalClient$override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) - NestedBlockDepth:SignalClient.kt$SignalClient$private fun handleSignalResponse(ws: WebSocket, response: LivekitRtc.SignalResponse) + NestedBlockDepth:SignalClient.kt$SignalClient$private fun handleSignalResponse(connection: SignalConnection, response: LivekitRtc.SignalResponse) SwallowedException:FlowExt.kt$e: CancellationException SwallowedException:LocalVideoTrack.kt$LocalVideoTrack$e: Exception SwallowedException:TextureViewRenderer.kt$TextureViewRenderer$e: NotFoundException @@ -86,7 +86,7 @@ TooManyFunctions:Participant.kt$ParticipantListener TooManyFunctions:PeerConnectionTransport.kt$PeerConnectionTransport TooManyFunctions:PublisherTransportObserver.kt$PublisherTransportObserver : ObserverListenerPeerConnectionStateObservable - TooManyFunctions:RTCEngine.kt$RTCEngine : Listener + TooManyFunctions:RTCEngine.kt$RTCEngine : ListenerSignalSessionListener TooManyFunctions:RTCEngine.kt$RTCEngine$Listener TooManyFunctions:RTCMetricsManager.kt$io.livekit.android.room.metrics.RTCMetricsManager.kt TooManyFunctions:RTCModule.kt$RTCModule 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 e0e64e7db..88de19296 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 @@ -97,6 +97,7 @@ import livekit.org.webrtc.RtpTransceiver import livekit.org.webrtc.RtpTransceiver.RtpTransceiverInit import livekit.org.webrtc.SessionDescription import java.nio.ByteBuffer +import java.util.concurrent.atomic.AtomicLong import javax.inject.Inject import javax.inject.Named import javax.inject.Singleton @@ -119,7 +120,8 @@ internal constructor( private val ioDispatcher: CoroutineDispatcher, private val rtcThreadToken: RTCThreadToken, private val dataPacketCryptorFactory: DataPacketCryptorManager.Factory, -) : SignalClient.Listener { +) : SignalClient.Listener, + SignalClient.SignalSessionListener { internal var listener: Listener? = null /** @@ -175,9 +177,20 @@ internal constructor( internal var reconnectPolicy: ReconnectPolicy = DefaultReconnectPolicy() - private val pendingTrackResolvers: MutableMap> = + private val pendingTrackResolvers: MutableMap> = mutableMapOf() + // Counts full reconnect preparations, which invalidate the server-side state of every + // publish accepted before them. Soft reconnects preserve publishes and do not advance + // it. + private val fullReconnectEpoch = AtomicLong(0) + + internal fun currentFullReconnectEpoch(): Long = fullReconnectEpoch.get() + + internal fun advanceFullReconnectEpoch() { + fullReconnectEpoch.incrementAndGet() + } + internal var regionUrlProvider: RegionUrlProvider? = null private var sessionUrl: String? = null private var sessionToken: String? = null @@ -266,6 +279,7 @@ internal constructor( if (connectionState == ConnectionState.DISCONNECTED) { connectionState = ConnectionState.CONNECTING } + client.prepareSignalConnection(fullReconnectEpoch.get()) val joinResponse = client.join(url, token, options, roomOptions) ensureActive() @@ -394,13 +408,13 @@ internal constructor( /** * @param builder an optional builder to include other parameters related to the track */ - suspend fun addTrack( + internal suspend fun addTrack( cid: String, name: String, kind: LivekitModels.TrackType, stream: String?, builder: LivekitRtc.AddTrackRequest.Builder = LivekitRtc.AddTrackRequest.newBuilder(), - ): LivekitModels.TrackInfo { + ): PublishAcceptance { synchronized(pendingTrackResolvers) { if (pendingTrackResolvers[cid] != null) { throw TrackException.DuplicateTrackException("Track with same ID $cid has already been published!") @@ -639,6 +653,7 @@ internal constructor( LKLog.v { "Attempting soft reconnect." } subscriber?.prepareForIceRestart() try { + client.prepareSignalConnection(fullReconnectEpoch.get()) val response = client.reconnect(url!!, token, participantSid) if (response is Either.Left) { val reconnectResponse = response.value @@ -1193,6 +1208,14 @@ internal constructor( } override fun onLocalTrackPublished(response: LivekitRtc.TrackPublishedResponse) { + handleLocalTrackPublished(response, fullReconnectEpoch.get()) + } + + override fun onLocalTrackPublishedInSession(response: LivekitRtc.TrackPublishedResponse, fullReconnectEpoch: Long) { + handleLocalTrackPublished(response, fullReconnectEpoch) + } + + private fun handleLocalTrackPublished(response: LivekitRtc.TrackPublishedResponse, fullReconnectEpoch: Long) { val cid = response.cid ?: run { LKLog.e { "local track published with null cid?" } return @@ -1211,7 +1234,7 @@ internal constructor( LKLog.d { "missing track resolver for: $cid" } return } - cont.resume(response.track) + cont.resume(PublishAcceptance(response.track, fullReconnectEpoch)) } override fun onLocalTrackSubscribed(trackSubscribed: LivekitRtc.TrackSubscribed) { @@ -1607,6 +1630,15 @@ internal class SenderTransceiverHandle( internal val signalSessionState: SignalSessionState, ) +/** + * A server-accepted publish: the track info from the TrackPublished response, stamped with + * the full reconnect epoch of the signal session that delivered it. + */ +internal class PublishAcceptance( + val trackInfo: LivekitModels.TrackInfo, + val fullReconnectEpoch: Long, +) + /** * @suppress */ diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt index 3f2432705..dd2fbeb38 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt @@ -87,7 +87,12 @@ constructor( private set @Volatile - private var currentWs: WebSocket? = null + private var currentSignalConnection: SignalConnection? = null + + // Copied into the next WebSocket's connection context so each response is attributed + // to the full reconnect epoch of the session that delivered it. + @Volatile + private var nextFullReconnectEpoch = 0L @Volatile private var isReconnecting: Boolean = false @@ -112,7 +117,7 @@ constructor( /** * @see [onReadyForResponses] */ - private val responseFlow = MutableSharedFlow>(Int.MAX_VALUE) + private val responseFlow = MutableSharedFlow>(Int.MAX_VALUE) private val responseFlowJobLock = Object() private var responseFlowJob: Job? = null @@ -124,6 +129,10 @@ constructor( var connectionState: ConnectionState = ConnectionState.DISCONNECTED + internal fun prepareSignalConnection(fullReconnectEpoch: Long) { + nextFullReconnectEpoch = fullReconnectEpoch + } + /** * @throws Exception if fails to connect. */ @@ -197,9 +206,10 @@ constructor( // onFailure will handle cleanup. LKLog.v { "connect cancelled, abort websocket" } joinContinuation = null - currentWs?.cancel() + currentSignalConnection?.webSocket?.cancel() } - currentWs = websocketFactory.newWebSocket(request, this@SignalClient) + val webSocket = websocketFactory.newWebSocket(request, this@SignalClient) + currentSignalConnection = SignalConnection(webSocket, nextFullReconnectEpoch) } } } @@ -258,9 +268,9 @@ constructor( synchronized(responseFlowJobLock) { if (responseFlowJob == null) { responseFlowJob = coroutineScope.launch { - responseFlow.collect { (ws, response) -> + responseFlow.collect { (connection, response) -> responseFlow.resetReplayCache() - handleSignalResponseImpl(ws, response) + handleSignalResponseImpl(connection, response) } } } @@ -294,7 +304,7 @@ constructor( // --------------------------------- WebSocket Listener --------------------------------------// override fun onMessage(webSocket: WebSocket, text: String) { - if (webSocket != currentWs) { + if (webSocket !== currentSignalConnection?.webSocket) { // Possibly message from old websocket, discard. return } @@ -303,7 +313,8 @@ constructor( } override fun onMessage(webSocket: WebSocket, bytes: ByteString) { - if (webSocket != currentWs) { + val connection = currentSignalConnection + if (webSocket !== connection?.webSocket) { // Possibly message from old websocket, discard. return } @@ -312,11 +323,11 @@ constructor( .mergeFrom(byteArray) val response = signalResponseBuilder.build() - handleSignalResponse(webSocket, response) + handleSignalResponse(connection, response) } override fun onClosed(webSocket: WebSocket, code: Int, reason: String) { - if (webSocket != currentWs) { + if (webSocket !== currentSignalConnection?.webSocket) { return } handleWebSocketClose(reason, code) @@ -327,7 +338,7 @@ constructor( } override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) { - if (webSocket != currentWs) { + if (webSocket !== currentSignalConnection?.webSocket) { return } var reason: String? = null @@ -653,20 +664,21 @@ constructor( private fun sendRequestImpl(request: LivekitRtc.SignalRequest) { LKLog.v { "sending request: $request" } - if (!isConnected || currentWs == null) { + val connection = currentSignalConnection + if (!isConnected || connection == null) { LKLog.w { "not connected, could not send request $request" } return } val message = request.toByteArray().toByteString() - val sent = currentWs?.send(message) ?: false + val sent = connection.webSocket.send(message) if (!sent) { LKLog.e { "error sending request: $request" } } } - private fun handleSignalResponse(ws: WebSocket, response: LivekitRtc.SignalResponse) { - if (ws != currentWs) { + private fun handleSignalResponse(connection: SignalConnection, response: LivekitRtc.SignalResponse) { + if (connection !== currentSignalConnection) { return } @@ -695,7 +707,7 @@ constructor( joinContinuation = null } else if (response.hasLeave()) { // Some reconnects may immediately send leave back without a join response first. - handleSignalResponseImpl(ws, response) + handleSignalResponseImpl(connection, response) val cont = joinContinuation joinContinuation = null cont?.resumeWithException( @@ -737,11 +749,11 @@ constructor( return } } - responseFlow.tryEmit(ws to response) + responseFlow.tryEmit(connection to response) } - private fun handleSignalResponseImpl(ws: WebSocket, response: LivekitRtc.SignalResponse) { - if (ws != currentWs) { + private fun handleSignalResponseImpl(connection: SignalConnection, response: LivekitRtc.SignalResponse) { + if (connection !== currentSignalConnection) { LKLog.v { "received message from old websocket, discarding." } return } @@ -779,7 +791,12 @@ constructor( } LivekitRtc.SignalResponse.MessageCase.TRACK_PUBLISHED -> { - listener?.onLocalTrackPublished(response.trackPublished) + val listener = listener + if (listener is SignalSessionListener) { + listener.onLocalTrackPublishedInSession(response.trackPublished, connection.fullReconnectEpoch) + } else { + listener?.onLocalTrackPublished(response.trackPublished) + } } LivekitRtc.SignalResponse.MessageCase.SPEAKERS_CHANGED -> { @@ -903,7 +920,7 @@ constructor( pongJob = coroutineScope.launch { delay(pingTimeoutDurationMillis) LKLog.d { "Ping timeout reached for ping sent at $timestamp." } - currentWs?.close(CLOSE_REASON_PING_TIMEOUT, "Ping timeout") + currentSignalConnection?.webSocket?.close(CLOSE_REASON_PING_TIMEOUT, "Ping timeout") } } @@ -933,8 +950,8 @@ constructor( pingJob = null pongJob?.cancel() pongJob = null - currentWs?.close(code, reason) - currentWs = null + currentSignalConnection?.webSocket?.close(code, reason) + currentSignalConnection = null // Same ordering as [onFailure]: clear the field before cancel() for synchronous listener callbacks. val joinCont = joinContinuation joinContinuation = null @@ -950,6 +967,15 @@ constructor( serverInfo = null } + private class SignalConnection( + val webSocket: WebSocket, + val fullReconnectEpoch: Long, + ) + + internal interface SignalSessionListener { + fun onLocalTrackPublishedInSession(response: LivekitRtc.TrackPublishedResponse, fullReconnectEpoch: Long) + } + interface Listener { fun onServerAnswer(sessionDescription: SessionDescription, offerId: Int) fun onServerOffer(sessionDescription: SessionDescription, offerId: Int) diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt index 75ac69c11..96000f48b 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt @@ -33,6 +33,7 @@ import io.livekit.android.events.ParticipantEvent import io.livekit.android.room.ClientProtocolVersion import io.livekit.android.room.ConnectionState import io.livekit.android.room.DefaultsManager +import io.livekit.android.room.PublishAcceptance import io.livekit.android.room.RTCEngine import io.livekit.android.room.Room import io.livekit.android.room.SenderTransceiverHandle @@ -64,6 +65,7 @@ import io.livekit.android.util.flow import io.livekit.android.util.rethrowIfCancellationSignal import io.livekit.android.webrtc.sortVideoCodecPreferences import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.CoroutineStart import kotlinx.coroutines.Job import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.async @@ -77,7 +79,6 @@ import livekit.LivekitModels import livekit.LivekitModels.AudioTrackFeature import livekit.LivekitModels.Codec import livekit.LivekitModels.DataPacket -import livekit.LivekitModels.TrackInfo import livekit.LivekitRtc import livekit.LivekitRtc.AddTrackRequest import livekit.LivekitRtc.SimulcastCodec @@ -134,6 +135,13 @@ internal constructor( .mapNotNull { it as? LocalTrackPublication } .toList() + // Feature-collector jobs of the published audio tracks, guarded by its own + // monitor. Retiring paths remove publications before draining their jobs, and a + // full reconnect preparation advances the engine's publish epoch. Registration + // requires the publication to still be present and the epoch stamped at the + // server's acceptance of the publish to still be current, so a collector never + // starts for a publication a reconnect invalidated, while one accepted by the + // post-reconnect session registers normally. private val jobs = mutableMapOf() // For ensuring that only one caller can execute setTrackEnabled at a time. @@ -489,9 +497,9 @@ internal constructor( } }, ) - var publication: LocalTrackPublication? = null + var result: PublishResult? = null try { - publication = publishTrackImpl( + result = publishTrackImpl( track = track, options = options, requestConfig = { @@ -507,16 +515,35 @@ internal constructor( LKLog.e(e) { "Error thrown when publishing track:" } } - if (publication != null) { - val job = scope.launch { + if (result != null) { + val publication = result.publication + val job = scope.launch(start = CoroutineStart.LAZY) { track::features.flow.collect { engine.updateLocalAudioTrack(publication.sid, it + options.getFeaturesList()) } } - jobs[publication] = job + // The publication may be retired by unpublishTrack or a full reconnect + // while this publish completes; registering afterwards would leave the + // job running with a stale sid, so it only starts if no full reconnect + // intervened since the server accepted the publish and the publication + // is still current. + val registered = synchronized(jobs) { + val current = + engine.currentFullReconnectEpoch() == result.acceptanceEpoch && + trackPublications[publication.sid] === publication + if (current) { + jobs[publication] = job + } + current + } + if (registered) { + job.start() + } else { + job.cancel() + } } - return publication != null + return result != null } /** @@ -615,7 +642,7 @@ internal constructor( }, encodings = encodings, publishListener = publishListener, - ) + )?.publication } catch (e: TrackException.PublishException) { LKLog.e(e) { "Error thrown when publishing track:" } } @@ -644,7 +671,7 @@ internal constructor( /** * @throws TrackException.PublishException thrown when the publish fails. see [TrackException.PublishException.message] for details. - * @return true if the track publish was successful. + * @return the publish result, or null if the publish failed. */ @Throws(TrackException.PublishException::class) private suspend fun publishTrackImpl( @@ -653,7 +680,7 @@ internal constructor( requestConfig: AddTrackRequest.Builder.() -> Unit, encodings: List = emptyList(), publishListener: PublishListener? = null, - ): LocalTrackPublication? { + ): PublishResult? { if (track.isDisposed) { LKLog.w { "Attempting to publish a disposed track, ignoring." } return null @@ -775,7 +802,7 @@ internal constructor( // so no need to call negotiate manually. } - suspend fun requestAddTrack(): TrackInfo? { + suspend fun requestAddTrack(): PublishAcceptance? { return try { engine.addTrack( cid = cid, @@ -792,12 +819,13 @@ internal constructor( } var publication: LocalTrackPublication? = null + var result: PublishResult? = null try { - val trackInfo: TrackInfo? + val acceptance: PublishAcceptance? if (enabledPublishVideoCodecs.isNotEmpty()) { // Can simultaneous publish and negotiate. // codec is pre-verified in publishVideoTrack - trackInfo = coroutineScope { + acceptance = coroutineScope { val negotiateJob = launch { negotiate() } val publishJob = async { requestAddTrack() } @@ -806,12 +834,12 @@ internal constructor( } } else { // legacy path. - trackInfo = requestAddTrack() - if (trackInfo != null) { + acceptance = requestAddTrack() + if (acceptance != null) { if (options is VideoTrackPublishOptions) { // server might not support the codec the client has requested, in that case, fallback // to a supported codec - val primaryCodecMime = trackInfo.codecsList.firstOrNull()?.mimeType + val primaryCodecMime = acceptance.trackInfo.codecsList.firstOrNull()?.mimeType if (primaryCodecMime != null) { val updatedCodec = primaryCodecMime.mimeTypeToVideoCodec() @@ -832,13 +860,14 @@ internal constructor( } } - if (trackInfo != null) { + if (acceptance != null) { publication = LocalTrackPublication( - info = trackInfo, + info = acceptance.trackInfo, track = track, participant = this, options = options, ) + result = PublishResult(publication, acceptance.fullReconnectEpoch) addTrackPublication(publication) LKLog.v { "add track publication $publication" } @@ -874,9 +903,19 @@ internal constructor( } } - return publication + return result } + /** + * A completed publish: the created publication and the full reconnect epoch stamped + * when the server accepted it, used to decide whether bookkeeping keyed to the + * publication is still valid at registration time. + */ + private class PublishResult( + val publication: LocalTrackPublication, + val acceptanceEpoch: Long, + ) + private fun computeVideoEncodings( isScreenShare: Boolean, dimensions: Track.Dimensions, @@ -1021,15 +1060,13 @@ internal constructor( return } - val publicationJob = jobs[publication] - if (publicationJob != null) { - publicationJob.cancel() - jobs.remove(publication) - } - val sid = publication.sid trackPublications = trackPublications.toMutableMap().apply { remove(sid) } + // The publication is retired above, so a racing publishAudioTrack fails its + // registration re-check and cannot register a job after this drain. + synchronized(jobs) { jobs.remove(publication) }?.cancel() + if (engine.connectionState == ConnectionState.CONNECTED) { engine.removeTrack(track.rtcTrack) @@ -1334,7 +1371,7 @@ internal constructor( } negotiateJob.join() try { - val trackInfo = publishJob.await() + val trackInfo = publishJob.await().trackInfo LKLog.d { "published $codec for track ${track.sid}, $trackInfo" } } catch (e: Exception) { e.rethrowIfCancellationSignal() @@ -1364,6 +1401,21 @@ internal constructor( trackPublications = trackPublications.toMutableMap().apply { clear() } + // The publications are cleared, so the unpublishTrack calls during + // republishing can't find them to cancel their jobs, and republishing + // creates fresh jobs for the new publications. Advancing the epoch fails + // the registration re-check of any publish accepted before this + // preparation, whose publication would otherwise re-insert itself after + // this drain and satisfy the presence check. Only an acceptance from a + // signal session opened for the current epoch can register afterwards. + val staleJobs = synchronized(jobs) { + engine.advanceFullReconnectEpoch() + val stale = jobs.values.toList() + jobs.clear() + stale + } + staleJobs.forEach { it.cancel() } + for (publication in pubs) { internalListener?.onTrackUnpublished(publication, this) eventBus.postEvent(ParticipantEvent.LocalTrackUnpublished(this, publication), scope) 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 6225fb1ec..c3b590487 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 @@ -567,7 +567,7 @@ class RTCEngineMockE2ETest : MockE2ETest() { assertTrue("Retry should have succeeded, got $secondFailure", secondFailure == null) assertEquals( TestData.LOCAL_TRACK_PUBLISHED.trackPublished.track.sid, - secondPublish.getCompleted().sid, + secondPublish.getCompleted().trackInfo.sid, ) } finally { supervisor.cancel() diff --git a/livekit-android-test/src/test/java/io/livekit/android/room/RoomReconnectionMockE2ETest.kt b/livekit-android-test/src/test/java/io/livekit/android/room/RoomReconnectionMockE2ETest.kt index 1b1742bd3..4f6135b72 100644 --- a/livekit-android-test/src/test/java/io/livekit/android/room/RoomReconnectionMockE2ETest.kt +++ b/livekit-android-test/src/test/java/io/livekit/android/room/RoomReconnectionMockE2ETest.kt @@ -20,6 +20,7 @@ import io.livekit.android.room.track.DataPublishReliability import io.livekit.android.room.track.RemoteTrackPublication import io.livekit.android.room.track.Track import io.livekit.android.test.MockE2ETest +import io.livekit.android.test.mock.MockAudioStreamTrack import io.livekit.android.test.mock.MockDataChannel import io.livekit.android.test.mock.MockMediaStream import io.livekit.android.test.mock.MockRtpReceiver @@ -30,15 +31,23 @@ import io.livekit.android.test.mock.createMediaStreamId import io.livekit.android.test.mock.room.track.createMockLocalAudioTrack import io.livekit.android.test.util.toPBByteString import io.livekit.android.util.toOkioByteString +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.async +import kotlinx.coroutines.test.StandardTestDispatcher import kotlinx.coroutines.test.advanceUntilIdle +import kotlinx.coroutines.test.runCurrent import livekit.LivekitRtc import livekit.org.webrtc.PeerConnection import org.junit.Assert.assertEquals +import org.junit.Assert.assertNotSame import org.junit.Assert.assertTrue import org.junit.Test import org.junit.runner.RunWith +import org.mockito.Mockito import org.robolectric.RobolectricTestRunner +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit /** * For tests that only target one reconnection type. @@ -284,4 +293,198 @@ class RoomReconnectionMockE2ETest : MockE2ETest() { println(sentRequests) assertTrue(sentAddTrack) } + + @Test + fun softReconnectKeepsFeatureUpdatesFromPublication() = runTest { + room.setReconnectionType(ReconnectType.FORCE_SOFT_RECONNECT) + connect() + + val audioTrack = createMockLocalAudioTrack() + room.localParticipant.publishAudioTrack(track = audioTrack) + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + reconnectWebsocket() + connectPeerConnection() + advanceUntilIdle() + + val baseline = countAudioTrackFeatureUpdates() + assertTrue(audioTrack.applyOptions(audioTrack.options.copy(echoCancellation = false)).isSuccess) + advanceUntilIdle() + + assertEquals(1, countAudioTrackFeatureUpdates() - baseline) + } + + @Test + fun fullReconnectStopsFeatureUpdatesFromOldPublication() = runTest { + room.setReconnectionType(ReconnectType.FORCE_FULL_RECONNECT) + connect() + + val audioTrack = createMockLocalAudioTrack() + room.localParticipant.publishAudioTrack(track = audioTrack) + + disconnectPeerConnection() + // Wait so that the reconnect job properly starts first. + testScheduler.advanceTimeBy(1000) + reconnectWebsocket() + connectPeerConnection() + testScheduler.advanceUntilIdle() + + val baseline = countAudioTrackFeatureUpdates() + + // A features change should be reported once, by the republished publication. + // A second update means the old publication's feature collector is still + // running and reporting a stale track sid. + assertTrue(audioTrack.applyOptions(audioTrack.options.copy(echoCancellation = false)).isSuccess) + testScheduler.advanceUntilIdle() + + assertEquals(1, countAudioTrackFeatureUpdates() - baseline) + } + + @Test + fun fullReconnectStopsFeatureUpdatesFromPublishCompletingDuringPreparation() = runTest { + connect() + + var deferredAddTrack: LivekitRtc.AddTrackRequest? = null + wsFactory.registerSignalRequestHandler { request -> + if (request.hasAddTrack() && deferredAddTrack == null) { + deferredAddTrack = request.addTrack + true + } else { + false + } + } + + val audioTrack = createMockLocalAudioTrack() + val publish = async(StandardTestDispatcher(testScheduler)) { + room.localParticipant.publishAudioTrack(audioTrack) + } + runCurrent() + + val addTrack = requireNotNull(deferredAddTrack) + wsFactory.receiveMessage( + LivekitRtc.SignalResponse.newBuilder() + .setTrackPublished( + LivekitRtc.TrackPublishedResponse.newBuilder() + .setCid(addTrack.cid) + .setTrack(TestData.LOCAL_AUDIO_TRACK), + ) + .build(), + ) + // The old session accepted the response, but the publish continuation has + // not created its publication or feature collector yet. + room.localParticipant.prepareForFullReconnect() + runCurrent() + assertTrue(publish.await()) + + val baseline = countAudioTrackFeatureUpdates() + assertTrue(audioTrack.applyOptions(audioTrack.options.copy(echoCancellation = false)).isSuccess) + runCurrent() + + assertEquals(0, countAudioTrackFeatureUpdates() - baseline) + } + + @Test + fun fullReconnectStopsFeatureUpdatesFromPublishAcceptedByOldSessionAfterPreparation() = runTest { + connect() + + var deferredAddTrack: LivekitRtc.AddTrackRequest? = null + wsFactory.registerSignalRequestHandler { request -> + if (request.hasAddTrack() && deferredAddTrack == null) { + deferredAddTrack = request.addTrack + true + } else { + false + } + } + + val audioTrack = createMockLocalAudioTrack() + val publish = async(StandardTestDispatcher(testScheduler)) { + room.localParticipant.publishAudioTrack(audioTrack) + } + runCurrent() + + room.localParticipant.prepareForFullReconnect() + + // The old signal session can still deliver responses until its replacement starts. + val addTrack = requireNotNull(deferredAddTrack) + wsFactory.receiveMessage( + LivekitRtc.SignalResponse.newBuilder() + .setTrackPublished( + LivekitRtc.TrackPublishedResponse.newBuilder() + .setCid(addTrack.cid) + .setTrack(TestData.LOCAL_AUDIO_TRACK), + ) + .build(), + ) + advanceUntilIdle() + assertTrue(publish.getCompleted()) + + val baseline = countAudioTrackFeatureUpdates() + assertTrue(audioTrack.applyOptions(audioTrack.options.copy(echoCancellation = false)).isSuccess) + advanceUntilIdle() + + assertEquals(0, countAudioTrackFeatureUpdates() - baseline) + } + + @Test + fun fullReconnectKeepsFeatureUpdatesFromPublishAcceptedByNewSession() = runTest { + room.setReconnectionType(ReconnectType.FORCE_FULL_RECONNECT) + connect() + + val oldWebSocket = wsFactory.ws + val idRequested = CountDownLatch(1) + val resumeId = CountDownLatch(1) + val publishCompleted = CountDownLatch(1) + val mediaTrack = Mockito.spy( + MockAudioStreamTrack(id = TestData.LOCAL_TRACK_PUBLISHED.trackPublished.cid), + ) + Mockito.doAnswer { + idRequested.countDown() + check(resumeId.await(5, TimeUnit.SECONDS)) { "Timed out waiting for full reconnect" } + TestData.LOCAL_TRACK_PUBLISHED.trackPublished.cid + }.`when`(mediaTrack).id() + + val audioTrack = createMockLocalAudioTrack(mediaTrack = mediaTrack) + val publish = async(Dispatchers.Default) { + room.localParticipant.publishAudioTrack(audioTrack) + } + // Completion handlers run after the Deferred reaches its final state, so the + // latch guarantees getCompleted() below cannot race the state transition. The + // test body must not suspend on publish.await(): it would resume inside the + // publish coroutine's frame, where the unconfined event loop defers the + // applyOptions feature propagation past the assertions. + publish.invokeOnCompletion { publishCompleted.countDown() } + + try { + // The publish has passed its connection-state check but has not called addTrack. + assertTrue(idRequested.await(5, TimeUnit.SECONDS)) + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + assertNotSame(oldWebSocket, wsFactory.ws) + reconnectWebsocket() + connectPeerConnection() + + // addTrack is sent through the replacement signal session and accepted there. + resumeId.countDown() + assertTrue(publishCompleted.await(5, TimeUnit.SECONDS)) + assertTrue(publish.getCompleted()) + + val baseline = countAudioTrackFeatureUpdates() + assertTrue(audioTrack.applyOptions(audioTrack.options.copy(echoCancellation = false)).isSuccess) + advanceUntilIdle() + + assertEquals(1, countAudioTrackFeatureUpdates() - baseline) + } finally { + resumeId.countDown() + } + } + + private fun countAudioTrackFeatureUpdates() = wsFactory.ws.sentRequests.count { requestString -> + LivekitRtc.SignalRequest.newBuilder() + .mergeFrom(requestString.toPBByteString()) + .build() + .hasUpdateAudioTrack() + } }