diff --git a/.changes/reconnect-request-dispatch b/.changes/reconnect-request-dispatch new file mode 100644 index 000000000..bc1a1f7e8 --- /dev/null +++ b/.changes/reconnect-request-dispatch @@ -0,0 +1 @@ +patch type="fixed" "Reconnect requests are no longer dropped when one attempt is already running, and a reason that requires a full reconnect is no longer lost when a later request replaces it" \ No newline at end of file diff --git a/.changes/resume-severed-signal-retry b/.changes/resume-severed-signal-retry new file mode 100644 index 000000000..f32c46c3f --- /dev/null +++ b/.changes/resume-severed-signal-retry @@ -0,0 +1 @@ +patch type="fixed" "A resume whose signal connection drops before it completes is retried instead of being reported as reconnected" diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index b9d0240f3..ba82dfefb 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -114,8 +114,18 @@ class Engine extends Disposable with EventsEmittable { String? _connectedServerAddress; String? get connectedServerAddress => _connectedServerAddress; + /// A *pending* full-reconnect request. Consumed at the start of each + /// reconnect attempt, so it is false while an attempt runs unless a new + /// request arrived mid-attempt — use [isFullReconnectInProgress] to ask + /// what the running attempt is doing. bool fullReconnectOnNext = false; + bool _attemptIsFullReconnect = false; + + /// Whether the reconnect attempt currently running is a full reconnect + /// (as opposed to a resume). False when no attempt is in flight. + bool get isFullReconnectInProgress => _attemptIsFullReconnect; + // server-provided ice servers List _serverProvidedIceServers = []; @@ -1148,6 +1158,15 @@ class Engine extends Disposable with EventsEmittable { fullReconnectOnNext = true; } + // Consume the flag up front: this attempt's mode is now fixed, and from + // here a `true` value unambiguously means a *new* full-reconnect request + // arrived while we were running (e.g. a server RECONNECT leave during a + // resume), which the finally block dispatches. Mirrors client-sdk-js. + final fullReconnect = fullReconnectOnNext; + fullReconnectOnNext = false; + _attemptIsFullReconnect = fullReconnect; + + var succeeded = false; try { _attemptingReconnect = true; @@ -1163,7 +1182,7 @@ class Engine extends Disposable with EventsEmittable { ); } - if (fullReconnectOnNext) { + if (fullReconnect) { await restartConnection(); } else { await resumeConnection( @@ -1174,11 +1193,14 @@ class Engine extends Disposable with EventsEmittable { _clearPendingReconnect(); _attemptingReconnect = false; _isReconnecting = false; + succeeded = true; } catch (e) { _reconnectAttempts = _reconnectAttempts + 1; + logger.fine('attemptReconnect: ${fullReconnect ? 'full reconnect' : 'resume'} failed: $e'); bool recoverable = true; - if (e is WebSocketException || e is MediaConnectException) { - // cannot resume connection, need to do full reconnect + if (fullReconnect || e is WebSocketException || e is MediaConnectException) { + // a failed full reconnect stays a full reconnect; a resume that failed + // at the transport or media layer cannot be resumed again fullReconnectOnNext = true; } @@ -1206,6 +1228,15 @@ class Engine extends Disposable with EventsEmittable { } } finally { _attemptingReconnect = false; + _attemptIsFullReconnect = false; + + // A full reconnect requested while this attempt was running that a + // successful attempt didn't act on — dispatch it now. The failure path + // already retries, so only the success path needs this. + if (succeeded && fullReconnectOnNext && !_isClosed) { + logger.fine('attemptReconnect: full reconnect requested mid-attempt, dispatching'); + unawaited(handleReconnect(ClientDisconnectReason.reconnectRetry)); + } } } @@ -1264,6 +1295,18 @@ class Engine extends Disposable with EventsEmittable { logger.fine('resumeConnection: primary connected'); } + // The socket can drop while the peer connections were being restored. A + // resume that ends with a dead signal connection is a failure, not a + // success: throwing here lets the retry path run another resume instead of + // reporting the room as reconnected and cancelling the pending request. + // Mirrors the re-check in client-sdk-js and rust-sdks. + if (signalClient.connectionState != ConnectionState.connected) { + throw ConnectException( + 'resumeConnection: signal connection severed during resume', + reason: ConnectionErrorReason.InternalError, + ); + } + _isReconnecting = false; events.emit(const EngineResumedEvent()); } @@ -1311,7 +1354,10 @@ class Engine extends Disposable with EventsEmittable { await ensurePublisherConnected(); } - fullReconnectOnNext = false; + // fullReconnectOnNext is not cleared here. attemptReconnect consumed the + // request that started this restart, so a true value at this point is a + // new request (e.g. a RECONNECT leave from the node we just joined) that + // the finally block in attemptReconnect dispatches once we return. _regionUrlProvider?.resetAttempts(); events.emit(const EngineRestartedEvent()); } catch (error) { @@ -1449,7 +1495,11 @@ class Engine extends Disposable with EventsEmittable { }) ..on((event) async { logger.fine('Signal connected'); - _reconnectAttempts = 0; + // The attempt counter is not reset here. A resume opens its socket before + // the peer connections are restored, so a reset on socket connect would + // let an attempt that fails afterwards start again from zero and never + // reach the retry limit. _clearPendingReconnect resets it once an attempt + // has fully succeeded, and cleanUp on disconnect. events.emit(const EngineConnectedEvent()); }) ..on((event) async { diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index 16201d97b..5c3b2a077 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -507,7 +507,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { info: event.response.participant, ); - if (engine.fullReconnectOnNext) { + if (engine.isFullReconnectInProgress) { await _localParticipant!.updateFromInfo(event.response.participant); } @@ -522,7 +522,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { if (connectOptions.protocolVersion.index >= ProtocolVersion.v8.index && engine.fastConnectOptions != null && - !engine.fullReconnectOnNext) { + !engine.isFullReconnectInProgress) { final options = engine.fastConnectOptions!; final audio = options.microphone; @@ -651,7 +651,12 @@ class Room extends DisposableChangeNotifier with EventsEmittable { notifyListeners(); }) ..on((event) async { - if (!engine.fullReconnectOnNext || event.reason == DisconnectReason.clientInitiated) { + // Suppress while a full reconnect is either pending or running — the + // engine is going to re-establish the session, this is not a real + // disconnect. Both flags are needed since the attempt consumes the + // pending one when it starts. + if ((!engine.fullReconnectOnNext && !engine.isFullReconnectInProgress) || + event.reason == DisconnectReason.clientInitiated) { await _cleanUp(disposeLocalParticipant: false); events.emit(RoomDisconnectedEvent(reason: event.reason)); notifyListeners(); diff --git a/test/core/reconnect_request_dispatch_test.dart b/test/core/reconnect_request_dispatch_test.dart new file mode 100644 index 000000000..8e9e770a0 --- /dev/null +++ b/test/core/reconnect_request_dispatch_test.dart @@ -0,0 +1,261 @@ +// Copyright 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. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Reconnect-request bookkeeping: a request must not be dropped because another +// attempt was running, and the escalation implied by a request's reason must +// not be lost when a later request replaces it. +// +// The first test ports rust-sdks' `test_resume_escalation_sticks_across_cycles` +// (livekit/tests/peer_connection_signaling_test.rs). That test needs a live SFU, +// two participants and a published sine track, and observes the escalation via +// `LocalTrackRepublished` because only the full-reconnect path republishes. +// Here the mock transport lets us inject the concurrent request directly and +// observe the escalation as `RoomReconnectingEvent`, which only the full path +// emits. + +@Timeout(Duration(seconds: 10)) +library; + +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; + +import 'package:livekit_client/livekit_client.dart'; +import 'package:livekit_client/src/proto/livekit_models.pb.dart' as lk_models; +import 'package:livekit_client/src/proto/livekit_rtc.pb.dart' as lk_rtc; +import 'package:livekit_client/src/types/internal.dart'; +import '../mock/e2e_container.dart'; +import '../mock/peerconnection_mock.dart'; +import '../mock/websocket_mock.dart'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + late E2EContainer container; + late Room room; + late MockWebSocketConnector ws; + + setUp(() async { + resetMockDataChannels(); + container = E2EContainer(); + room = container.room; + ws = container.wsConnector; + await container.connectRoom(); + }); + + tearDown(() async { + await container.dispose(); + }); + + /// Spin until the SDK opens a new signal socket, returning its URI. + Future awaitNewSocket(Object? previousHandlers, {int maxWaitMs = 2000}) async { + for (var i = 0; i < maxWaitMs ~/ 10 && identical(ws.handlers, previousHandlers); i++) { + await Future.delayed(const Duration(milliseconds: 10)); + } + expect(identical(ws.handlers, previousHandlers), isFalse, reason: 'SDK never re-opened the signal connection'); + return ws.uri!; + } + + test('a full-reconnect request arriving mid-resume is dispatched after it succeeds', () async { + final roomEvents = []; + final cancel = room.events.listen(roomEvents.add); + + // Cycle 1: a resume, with a full-reconnect request injected while it runs. + final firstHandlers = ws.handlers; + ws.onDispose(); + final resumeUri = await awaitNewSocket(firstHandlers); + expect(resumeUri.queryParameters['reconnect'], '1', reason: 'cycle 1 must be a resume'); + + // What a server `Leave{RECONNECT}` mid-resume does. + container.engine.fullReconnectOnNext = true; + + final resumeHandlers = ws.handlers; + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + await room.events.waitFor(duration: const Duration(seconds: 5)); + + // Cycle 2 must be a full reconnect, dispatched from the request that + // arrived while cycle 1 was in flight — not silently dropped. + final restartUri = await awaitNewSocket(resumeHandlers); + expect(restartUri.queryParameters['reconnect'], isNull, reason: 'cycle 2 must re-join, not resume'); + await cancel(); + + expect( + roomEvents.whereType(), + isNotEmpty, + reason: 'cycle 1 must have been a resume, otherwise the cross-cycle behavior is not under test', + ); + expect( + roomEvents.whereType(), + isNotEmpty, + reason: 'the mid-attempt full-reconnect request must still be honored', + ); + }); + + test('an escalating reason is not lost when a later request replaces it', () async { + final roomEvents = []; + final cancel = room.events.listen(roomEvents.add); + final previousHandlers = ws.handlers; + + // A PeerConnection failure demands a full reconnect. The socket close that + // follows lands a second request whose reason (`signal`) implies only a + // resume, and it replaces the first request's pending timer. + unawaited(container.engine.handleReconnect(ClientDisconnectReason.peerConnectionFailed)); + await container.engine.handleReconnect(ClientDisconnectReason.signal); + + final uri = await awaitNewSocket(previousHandlers); + await cancel(); + + expect(uri.queryParameters['reconnect'], isNull, reason: 'the peer-connection failure must still force a re-join'); + expect(roomEvents.whereType(), isNotEmpty); + expect(roomEvents.whereType(), isEmpty); + }); + + test('a peer failure reported mid-resume is dispatched as a full reconnect afterwards', () async { + final roomEvents = []; + final cancel = room.events.listen(roomEvents.add); + + final firstHandlers = ws.handlers; + ws.onDispose(); + final resumeUri = await awaitNewSocket(firstHandlers); + expect(resumeUri.queryParameters['reconnect'], '1', reason: 'cycle 1 must be a resume'); + + // The real request path, not a direct flag write: a PeerConnection reports + // failed while the resume is in flight. handleReconnect records the + // escalation and schedules a timer that the running attempt swallows. + unawaited(container.engine.handleReconnect(ClientDisconnectReason.peerConnectionFailed)); + + final resumeHandlers = ws.handlers; + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + await room.events.waitFor(duration: const Duration(seconds: 5)); + + final restartUri = await awaitNewSocket(resumeHandlers); + await cancel(); + + expect(restartUri.queryParameters['reconnect'], isNull, reason: 'the peer failure must still force a re-join'); + expect(roomEvents.whereType(), isNotEmpty); + expect(container.engine.fullReconnectOnNext, isFalse, reason: 'the request must be consumed, not left stale'); + }); + + test('a RECONNECT leave arriving mid-restart is not lost', () async { + final roomEvents = []; + final cancel = room.events.listen(roomEvents.add); + + // Cycle 1: a full reconnect, answered with a JoinResponse. + final firstHandlers = ws.handlers; + container.engine.fullReconnectOnNext = true; + ws.onDispose(); + final joinUri = await awaitNewSocket(firstHandlers); + expect(joinUri.queryParameters['reconnect'], isNull, reason: 'cycle 1 must be a re-join'); + + // The node we are joining asks for another full reconnect before the join + // completes. restartConnection used to reset the flag after joining, which + // erased this request. + ws.onData( + lk_rtc.SignalResponse( + leave: lk_rtc.LeaveRequest( + action: lk_rtc.LeaveRequest_Action.RECONNECT, + reason: lk_models.DisconnectReason.STATE_MISMATCH, + ), + ).writeToBuffer(), + ); + + final joinHandlers = ws.handlers; + await container.answerJoin(); + await room.events.waitFor(duration: const Duration(seconds: 5)); + + // Cycle 2: the leave-driven full reconnect must still run. + final secondJoinUri = await awaitNewSocket(joinHandlers); + await cancel(); + + expect(secondJoinUri.queryParameters['reconnect'], isNull, reason: 'cycle 2 must be a re-join too'); + expect(roomEvents.whereType(), hasLength(2)); + }); + + test('a signal drop during the resume is retried instead of reported as success', () async { + final roomEvents = []; + final cancel = room.events.listen(roomEvents.add); + + final firstHandlers = ws.handlers; + ws.onDispose(); + await awaitNewSocket(firstHandlers); + + // The server answers the resume and the socket dies right behind it, before + // the peer connection work finishes. The attempt must not end in + // RoomReconnectedEvent with a dead signal connection. + final resumeHandlers = ws.handlers; + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + ws.onDispose(); + + final retryUri = await awaitNewSocket(resumeHandlers); + expect(retryUri.queryParameters['reconnect'], '1', reason: 'a severed signal is retried as a resume'); + expect( + roomEvents.whereType(), + isEmpty, + reason: 'the attempt with the dead socket must not be reported as a success', + ); + + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + await room.events.waitFor(duration: const Duration(seconds: 5)); + await cancel(); + + expect(roomEvents.whereType(), hasLength(1)); + }); + + test('repeated severed resumes count towards the retry limit', () async { + final attempts = []; + final cancel = room.events.listen((event) { + if (event is RoomAttemptReconnectEvent) attempts.add(event.attempt); + }); + + // Every resume opens its socket, gets its ReconnectResponse and then loses + // the socket before the attempt completes. The socket connect used to reset + // the attempt counter, so each failure scheduled "attempt 2" again and the + // retry limit was never reached. + var handlers = ws.handlers; + ws.onDispose(); + for (var i = 0; i < 3; i++) { + await awaitNewSocket(handlers, maxWaitMs: 6000); + handlers = ws.handlers; + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + ws.onDispose(); + // let the attempt reach its final check, fail and schedule the retry + await Future.delayed(const Duration(milliseconds: 50)); + } + + // Leave the engine in a clean state: answer the fourth attempt properly. + await awaitNewSocket(handlers, maxWaitMs: 6000); + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + await room.events.waitFor(duration: const Duration(seconds: 5)); + await cancel(); + + // Each failure schedules twice (the socket close and the retry), so look at + // the distinct attempt numbers: they must climb, not repeat. + expect(attempts.where((a) => a > 1).toSet().toList(), [ + 2, + 3, + 4, + ], reason: 'every failed attempt must advance the counter'); + }, timeout: const Timeout(Duration(seconds: 30))); + + test('a successful resume leaves no full-reconnect state behind', () async { + final previousHandlers = ws.handlers; + ws.onDispose(); + await awaitNewSocket(previousHandlers); + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + await room.events.waitFor(duration: const Duration(seconds: 5)); + + expect(container.engine.fullReconnectOnNext, isFalse); + expect(container.engine.isFullReconnectInProgress, isFalse); + }); +}