From a9d135325d66ad579a0983a4c0a40edb1a18f100 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 1/2] Fixed local publication jobs leaking on every full reconnect --- .../fix-reconnect-publication-job-leak.md | 5 + .../detekt-baseline-release.xml | 14 +- .../java/io/livekit/android/room/RTCEngine.kt | 42 +++- .../io/livekit/android/room/SignalClient.kt | 70 ++++-- .../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 00000000..3c1994a9 --- /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 6bd3cb94..b066f54c 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,18 +27,18 @@ 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, encoded: ByteArray) + CyclomaticComplexMethod:SignalClient.kt$SignalClient$private fun handleSignalResponseImpl(connection: SignalConnection, response: LivekitRtc.SignalResponse, encoded: ByteArray) 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() LongMethod:Room.kt$Room$@Throws(Exception::class) suspend fun connect(url: String, token: String, options: ConnectOptions = ConnectOptions()) - LongMethod:SignalClient.kt$SignalClient$private fun handleSignalResponseImpl(ws: WebSocket, response: LivekitRtc.SignalResponse, encoded: ByteArray) + LongMethod:SignalClient.kt$SignalClient$private fun handleSignalResponseImpl(connection: SignalConnection, response: LivekitRtc.SignalResponse, encoded: ByteArray) LongParameterList:AudioBufferCallbackDispatcher.kt$AudioBufferCallback$(buffer: ByteBuffer, audioFormat: Int, channelCount: Int, sampleRate: Int, bytesRead: Int, captureTimeNs: Long) LongParameterList:KeyProvider.kt$BaseKeyProvider$( ratchetSalt: String = defaultRatchetSalt, uncryptedMagicBytes: String = defaultMagicBytes, ratchetWindowSize: Int = defaultRatchetWindowSize, override var enableSharedKey: Boolean = true, failureTolerance: Int = defaultFailureTolerance, keyRingSize: Int = defaultKeyRingSize, discardFrameWhenCryptorNotReady: Boolean = defaultDiscardFrameWhenCryptorNotReady, keyDerivationAlgorithm: FrameCryptorKeyDerivationAlgorithm = defaultKeyDerivationAlgorithm, ) LongParameterList:LiveKitOverrides.kt$AudioOptions$( /** * Override the default output [AudioType]. * * This affects the audio routing and how the audio is handled. Default is [AudioType.CallAudioType]. * * Note: if [audioHandler] is also passed, the values from [audioOutputType] will not be reflected in it, * and must be set yourself. */ val audioOutputType: AudioType? = null, /** * Override the default [AudioHandler]. * * Default is [AudioSwitchHandler]. * * Use [NoAudioHandler] to turn off automatic audio handling or * [AudioFocusHandler] to get simple audio focus handling. */ val audioHandler: AudioHandler? = null, /** * Override the default [AudioDeviceModule]. * * If a non-null value is passed, the library does not * take ownership of the object and will not release it upon [Room.release]. * It is the responsibility of the owner to call [AudioDeviceModule.release] when finished * with it to prevent memory leaks. */ val audioDeviceModule: AudioDeviceModule? = null, /** * Called after default setup to allow for customizations on the [JavaAudioDeviceModule]. * * Not used if [audioDeviceModule] is provided. * * Note: We require setting the [JavaAudioDeviceModule.Builder.setSamplesReadyCallback] to provide * support for [LocalAudioTrack.addSink]. If you wish to grab the audio samples * from the local microphone track, use [LocalAudioTrack.addSink] instead of setting your own * callback. */ val javaAudioDeviceModuleCustomizer: ((builder: JavaAudioDeviceModule.Builder) -> Unit)? = null, /** * On Android 11+, the audio mode will reset itself from [AudioManager.MODE_IN_COMMUNICATION] if * there is no audio playback or capture for 6 seconds (for example when joining a room with * no speakers and the local mic is muted.) This mode reset will cause unexpected * behavior when trying to change the volume, causing it to not properly change the volume. * * We use a workaround by playing a silent audio track to keep the communication mode from * resetting. * * Setting this flag to true will disable the workaround. * * This flag is a no-op when the audio mode is set to anything other than * [AudioManager.MODE_IN_COMMUNICATION]. */ val disableCommunicationModeWorkaround: Boolean = false, /** * Options for processing the mic and incoming audio. */ val audioProcessorOptions: AudioProcessorOptions? = null, /** * Devices may take some time initializing the audio stack for recording. * Prewarming allows starting up the underlying audio recording prior to publish, letting * the audio device be ready immediately when the track is fully published. * * If set to true, disables audio recording prewarming (and the related * [LocalAudioTrack.prewarm] function), and audio resources are only used while the * track is connected and published. Defaults to false. */ val disableAudioPrewarming: Boolean = false, ) @@ -63,7 +63,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 incomingDataTrackManager: IncomingDataTrackManager, 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>, ) @@ -73,7 +73,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, encoded: ByteArray) + NestedBlockDepth:SignalClient.kt$SignalClient$private fun handleSignalResponse(connection: SignalConnection, response: LivekitRtc.SignalResponse, encoded: ByteArray) SwallowedException:FlowExt.kt$e: CancellationException SwallowedException:LocalVideoTrack.kt$LocalVideoTrack$e: Exception SwallowedException:TextureViewRenderer.kt$TextureViewRenderer$e: NotFoundException @@ -88,7 +88,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 bc4c8235..386303b4 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 @@ -101,6 +101,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 @@ -125,7 +126,8 @@ internal constructor( private val dataPacketCryptorFactory: DataPacketCryptorManager.Factory, private val outgoingDataTrackManager: OutgoingDataTrackManager, private val incomingDataTrackManager: IncomingDataTrackManager, -) : SignalClient.Listener { +) : SignalClient.Listener, + SignalClient.SignalSessionListener { internal var listener: Listener? = null /** @@ -181,9 +183,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 @@ -275,6 +288,7 @@ internal constructor( if (connectionState == ConnectionState.DISCONNECTED) { connectionState = ConnectionState.CONNECTING } + client.prepareSignalConnection(fullReconnectEpoch.get()) val joinResponse = client.join(url, token, options, roomOptions) ensureActive() @@ -422,13 +436,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!") @@ -672,6 +686,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 @@ -1298,6 +1313,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 @@ -1316,7 +1339,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) { @@ -1766,6 +1789,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 788ca4da..3eccbd8e 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 @@ -95,7 +95,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 @@ -143,6 +148,10 @@ constructor( var connectionState: ConnectionState = ConnectionState.DISCONNECTED + internal fun prepareSignalConnection(fullReconnectEpoch: Long) { + nextFullReconnectEpoch = fullReconnectEpoch + } + /** * @throws Exception if fails to connect. */ @@ -216,9 +225,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) } } } @@ -279,7 +289,7 @@ constructor( responseFlowJob = coroutineScope.launch { responseFlow.collect { incoming -> responseFlow.resetReplayCache() - handleSignalResponseImpl(incoming.ws, incoming.response, incoming.encoded) + handleSignalResponseImpl(incoming.connection, incoming.response, incoming.encoded) } } } @@ -313,7 +323,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 } @@ -322,7 +332,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 } @@ -331,11 +342,11 @@ constructor( .mergeFrom(byteArray) val response = signalResponseBuilder.build() - handleSignalResponse(webSocket, response, byteArray) + handleSignalResponse(connection, response, byteArray) } override fun onClosed(webSocket: WebSocket, code: Int, reason: String) { - if (webSocket != currentWs) { + if (webSocket !== currentSignalConnection?.webSocket) { return } handleWebSocketClose(reason, code) @@ -346,7 +357,7 @@ constructor( } override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) { - if (webSocket != currentWs) { + if (webSocket !== currentSignalConnection?.webSocket) { return } var reason: String? = null @@ -758,20 +769,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, encoded: ByteArray) { - if (ws != currentWs) { + private fun handleSignalResponse(connection: SignalConnection, response: LivekitRtc.SignalResponse, encoded: ByteArray) { + if (connection !== currentSignalConnection) { return } @@ -801,7 +813,7 @@ constructor( joinContinuation = null } else if (response.hasLeave()) { // Some reconnects may immediately send leave back without a join response first. - handleSignalResponseImpl(ws, response, encoded) + handleSignalResponseImpl(connection, response, encoded) val cont = joinContinuation joinContinuation = null cont?.resumeWithException( @@ -843,11 +855,11 @@ constructor( return } } - responseFlow.tryEmit(IncomingSignal(ws, response, encoded)) + responseFlow.tryEmit(IncomingSignal(connection, response, encoded)) } - private fun handleSignalResponseImpl(ws: WebSocket, response: LivekitRtc.SignalResponse, encoded: ByteArray) { - if (ws != currentWs) { + private fun handleSignalResponseImpl(connection: SignalConnection, response: LivekitRtc.SignalResponse, encoded: ByteArray) { + if (connection !== currentSignalConnection) { LKLog.v { "received message from old websocket, discarding." } return } @@ -885,7 +897,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 -> { @@ -1032,7 +1049,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") } } @@ -1073,8 +1090,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 @@ -1090,6 +1107,15 @@ constructor( serverInfo = null } + private class SignalConnection( + val webSocket: WebSocket, + val fullReconnectEpoch: Long, + ) + + internal interface SignalSessionListener { + fun onLocalTrackPublishedInSession(response: LivekitRtc.TrackPublishedResponse, fullReconnectEpoch: Long) + } + internal interface Listener { fun onServerAnswer(sessionDescription: SessionDescription, offerId: Int) fun onServerOffer(sessionDescription: SessionDescription, offerId: Int) @@ -1121,7 +1147,7 @@ constructor( * Data-track managers parse the encoded form themselves. */ private class IncomingSignal( - val ws: WebSocket, + val connection: SignalConnection, val response: LivekitRtc.SignalResponse, val encoded: ByteArray, ) 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 816ed9b6..765a508d 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 @@ -70,6 +71,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 @@ -83,7 +85,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 @@ -143,6 +144,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. @@ -498,9 +506,9 @@ internal constructor( } }, ) - var publication: LocalTrackPublication? = null + var result: PublishResult? = null try { - publication = publishTrackImpl( + result = publishTrackImpl( track = track, options = options, requestConfig = { @@ -516,16 +524,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 } /** @@ -624,7 +651,7 @@ internal constructor( }, encodings = encodings, publishListener = publishListener, - ) + )?.publication } catch (e: TrackException.PublishException) { LKLog.e(e) { "Error thrown when publishing track:" } } @@ -653,7 +680,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( @@ -662,7 +689,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 @@ -784,7 +811,7 @@ internal constructor( // so no need to call negotiate manually. } - suspend fun requestAddTrack(): TrackInfo? { + suspend fun requestAddTrack(): PublishAcceptance? { return try { engine.addTrack( cid = cid, @@ -801,12 +828,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() } @@ -815,12 +843,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() @@ -841,13 +869,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" } @@ -883,9 +912,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, @@ -1030,15 +1069,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) @@ -1464,7 +1501,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() @@ -1494,6 +1531,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 6225fb1e..c3b59048 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 1b1742bd..4f6135b7 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() + } } From d9064e9aa4601f4f044b056aaacd45da216590c0 Mon Sep 17 00:00:00 2001 From: Adrian Niculescu <15037449+adrian-niculescu@users.noreply.github.com> Date: Wed, 16 Sep 2026 12:15:48 +0300 Subject: [PATCH 2/2] Fixed mock stream tracks breaking Mockito inline spies The public `enabled` constructor property generated a `void setEnabled(boolean)` accessor next to the `boolean setEnabled(boolean)` override. Mockito's inline mock maker resolves the intercepted method by name and parameter types, and the JVM picks either overload arbitrarily, so a spied track could run the property setter, return null, and fail unboxing in `Track.stop()`. Making the property private removes the accessor. --- .../java/io/livekit/android/test/mock/MockAudioStreamTrack.kt | 2 +- .../java/io/livekit/android/test/mock/MockVideoStreamTrack.kt | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockAudioStreamTrack.kt b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockAudioStreamTrack.kt index 61f4cad9..99b3755a 100644 --- a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockAudioStreamTrack.kt +++ b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockAudioStreamTrack.kt @@ -23,7 +23,7 @@ import livekit.org.webrtc.audio.AudioProcessingOptionsResult class MockAudioStreamTrack( val id: String = "id", val kind: String = AUDIO_TRACK_KIND, - var enabled: Boolean = true, + private var enabled: Boolean = true, var state: State = State.LIVE, ) : AudioTrack(1L) { diff --git a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockVideoStreamTrack.kt b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockVideoStreamTrack.kt index 93b9da4e..202debb4 100644 --- a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockVideoStreamTrack.kt +++ b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockVideoStreamTrack.kt @@ -1,5 +1,5 @@ /* - * Copyright 2023-2024 LiveKit, Inc. + * Copyright 2023-2026 LiveKit, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,7 +23,7 @@ import java.util.UUID class MockVideoStreamTrack( val id: String = UUID.randomUUID().toString(), val kind: String = VIDEO_TRACK_KIND, - var enabled: Boolean = true, + private var enabled: Boolean = true, var state: State = State.LIVE, ) : VideoTrack(1L) { val sinks = mutableSetOf()