diff --git a/.changes/buffer-wait-typed-error b/.changes/buffer-wait-typed-error new file mode 100644 index 000000000..a83e7896e --- /dev/null +++ b/.changes/buffer-wait-typed-error @@ -0,0 +1 @@ +patch type="fixed" "Data stream writes on a closed engine fail with UnexpectedConnectionState instead of a plain string error" \ No newline at end of file diff --git a/.changes/engine-terminal-closure b/.changes/engine-terminal-closure new file mode 100644 index 000000000..0429b4483 --- /dev/null +++ b/.changes/engine-terminal-closure @@ -0,0 +1 @@ +patch type="fixed" "Announce engine closure before room disposal callbacks and when reconnection gives up, and reject data channel buffer waits started after closure" diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 576c96ed4..118cb247b 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -242,7 +242,6 @@ class Engine extends Disposable with EventsEmittable { _setUpSignalListeners(); onDispose(() async { - _isClosed = true; await cleanUp(); await events.dispose(); await _signalListener.dispose(); @@ -250,6 +249,13 @@ class Engine extends Disposable with EventsEmittable { }); } + @override + Future dispose() { + // Disposal callbacks run in reverse registration order and may await engine work. + markClosed(); + return super.dispose(); + } + Future connect( String url, String token, { @@ -417,6 +423,11 @@ class Engine extends Disposable with EventsEmittable { } Future waitForBufferStatusLow(Reliability kind) async { + // A wait started after closure cannot observe the closing event. + if (_isClosed || isDisposed) { + throw UnexpectedConnectionState('Engine disconnected'); + } + final Completer completer = Completer(); if (isBufferStatusLow(kind) == true) { @@ -424,7 +435,7 @@ class Engine extends Disposable with EventsEmittable { } else { onClosing() { if (!completer.isCompleted) { - completer.completeError('Engine disconnected'); + completer.completeError(UnexpectedConnectionState('Engine disconnected')); } } @@ -1151,7 +1162,7 @@ class Engine extends Disposable with EventsEmittable { if (_reconnectAttempts >= _reconnectCount) { logger.fine('reconnectAttempts exceeded, disconnecting...'); - _isClosed = true; + markClosed(); await cleanUp(); events.emit( @@ -1263,6 +1274,7 @@ class Engine extends Disposable with EventsEmittable { unawaited(handleReconnect(ClientDisconnectReason.reconnectRetry)); } else { logger.fine('attemptReconnect: disconnecting...'); + markClosed(); // clean up before emitting, room's EngineDisconnectedEvent handler // drops the event while fullReconnectOnNext is still true and // cleanUp() is what resets it @@ -1677,11 +1689,19 @@ class Engine extends Disposable with EventsEmittable { events.emit(EngineRoomMovedEvent(response: event.response)); }); + // Announce terminal closure before tearing down transports or event listeners. + // A full reconnect also disposes transports, but must keep the engine open. + @internal + void markClosed() { + if (_isClosed) return; + _isClosed = true; + events.emit(EngineClosingEvent()); + } + Future disconnect({ DisconnectReason reason = DisconnectReason.clientInitiated, }) async { - _isClosed = true; - events.emit(EngineClosingEvent()); + markClosed(); if (connectionState == ConnectionState.connected) { await signalClient.sendLeave(); } else { diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index 9b1c93e2d..9a963e356 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -241,6 +241,13 @@ class Room extends DisposableChangeNotifier with EventsEmittable { }); } + @override + Future dispose() { + // Room cleanup tears down transports before Engine.dispose() is reached. + engine.markClosed(); + return super.dispose(); + } + /// prepareConnection should be called as soon as the page is loaded, in order /// to speed up the connection attempt. This function will /// - perform DNS resolution and pre-warm the DNS cache diff --git a/test/core/disconnect_during_reconnect_test.dart b/test/core/disconnect_during_reconnect_test.dart index e044f646d..63fa71f99 100644 --- a/test/core/disconnect_during_reconnect_test.dart +++ b/test/core/disconnect_during_reconnect_test.dart @@ -12,9 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. -// disconnect() closes the engine for good, even when a full reconnect is -// already tearing down the old connection and is about to connect again, and -// the room can connect again afterwards. +// disconnect() and Room.dispose() close the engine for good, even when a full +// reconnect is already tearing down the old connection and is about to connect +// again, and the room can connect again after disconnect(). @Timeout(Duration(seconds: 10)) library; @@ -93,4 +93,35 @@ void main() { await container.connectRoom(); expect(room.connectionState, ConnectionState.connected); }); + + test('disposing the room during a full reconnect does not connect again', () async { + final restartPaused = Completer(); + final resumeRestart = Completer(); + container.engine.publisher!.onDispose(() async { + restartPaused.complete(); + await resumeRestart.future; + }); + final disposePaused = Completer(); + final resumeDispose = Completer(); + container.engine.subscriber!.onDispose(() async { + disposePaused.complete(); + await resumeDispose.future; + }); + + container.engine.fullReconnectOnNext = true; + final droppedHandlers = ws.handlers; + ws.onDispose(); + await restartPaused.future; + + // The room's teardown reaches the subscriber while the full reconnect still holds the publisher. + final disposing = room.dispose(); + await disposePaused.future; + resumeRestart.complete(); + // Give a reconnect that was not stopped time to open its socket. + await Future.delayed(const Duration(milliseconds: 500)); + resumeDispose.complete(); + await disposing; + + expect(identical(ws.handlers, droppedHandlers), isTrue, reason: 'the full reconnect opened a new signal socket'); + }); } diff --git a/test/core/engine_closing_test.dart b/test/core/engine_closing_test.dart new file mode 100644 index 000000000..7c5de99f3 --- /dev/null +++ b/test/core/engine_closing_test.dart @@ -0,0 +1,179 @@ +// 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. + +@Timeout(Duration(seconds: 5)) +library; + +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; + +import 'package:livekit_client/livekit_client.dart'; +import 'package:livekit_client/src/core/engine.dart'; +import 'package:livekit_client/src/internal/events.dart'; +import 'package:livekit_client/src/types/internal.dart'; +import '../mock/e2e_container.dart'; +import '../mock/peerconnection_mock.dart'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + late E2EContainer container; + late List events; + + setUp(() async { + resetMockDataChannels(); + container = E2EContainer(); + await container.connectRoom(); + events = []; + container.engine.events.listen(events.add); + }); + + tearDown(() async { + await container.dispose(); + }); + + test('disposing the room announces engine closure', () async { + await container.room.dispose(); + + expect(events.whereType(), hasLength(1)); + expect(container.engine.isClosed, isTrue); + }); + + test('room closure is announced while the first transport teardown is blocked', () async { + final teardownStarted = Completer(); + final releaseTeardown = Completer(); + container.engine.publisher!.onDispose(() async { + teardownStarted.complete(); + await releaseTeardown.future; + }); + + final disposing = container.room.dispose(); + await teardownStarted.future; + final closedDuringTeardown = container.engine.isClosed; + final closingEventsDuringTeardown = events.whereType().length; + final emitterDisposedDuringTeardown = container.engine.events.isDisposed; + releaseTeardown.complete(); + await disposing; + + expect(closedDuringTeardown, isTrue); + expect(closingEventsDuringTeardown, 1); + expect(emitterDisposedDuringTeardown, isFalse); + }); + + for (final owner in ['room', 'engine']) { + test('$owner is marked closed before later disposal callbacks run', () async { + final disposable = owner == 'room' ? container.room : container.engine; + bool? closedInCallback; + disposable.onDispose(() async { + closedInCallback = container.engine.isClosed; + }); + + if (owner == 'room') { + await container.room.dispose(); + } else { + await container.engine.dispose(); + } + + expect(closedInCallback, isTrue); + expect(events.whereType(), hasLength(1)); + }); + } + + test('engine closure is delivered while disposing with its emitter still alive', () async { + bool? engineDisposedWhenDelivered; + bool? emitterDisposedWhenDelivered; + container.engine.events.once((_) { + engineDisposedWhenDelivered = container.engine.isDisposed; + emitterDisposedWhenDelivered = container.engine.events.isDisposed; + }); + + await container.engine.dispose(); + + expect(engineDisposedWhenDelivered, isTrue); + expect(emitterDisposedWhenDelivered, isFalse); + expect(container.engine.events.isDisposed, isTrue); + }); + + test('an unrecoverable reconnect announces closure before disconnection', () async { + container.wsConnector.connectError = CertificatePinningException( + 'Certificate pin mismatch', + host: 'www.example.com', + ); + final disconnected = container.engine.events.waitFor( + duration: const Duration(seconds: 1), + ); + + await container.engine.attemptReconnect(ClientDisconnectReason.reconnectRetry); + expect((await disconnected).reason, DisconnectReason.signalingConnectionFailure); + + expect(container.engine.isClosed, isTrue); + expect(events.whereType(), hasLength(1)); + expect( + events.indexWhere((event) => event is EngineClosingEvent), + lessThan(events.indexWhere((event) => event is EngineDisconnectedEvent)), + ); + }); + + test('exhausting reconnect attempts announces engine closure', () async { + // A generic resume failure stays recoverable without escalating to a full reconnect. + container.wsConnector.connectError = Exception('temporary connection failure'); + final disconnected = container.engine.events.waitFor( + duration: const Duration(seconds: 1), + ); + + // Drive the attempts directly rather than waiting through the retry delays. + for (var i = 0; i < defaultRetryDelaysInMs.length; i++) { + await container.engine.attemptReconnect(ClientDisconnectReason.reconnectRetry); + } + expect((await disconnected).reason, DisconnectReason.reconnectAttemptsExceeded); + + expect(container.engine.isClosed, isTrue); + expect(events.whereType(), hasLength(1)); + expect( + events.indexWhere((event) => event is EngineClosingEvent), + lessThan(events.indexWhere((event) => event is EngineDisconnectedEvent)), + ); + }); + + test('a recoverable reconnect does not announce engine closure', () async { + container.wsConnector.connectError = Exception('temporary connection failure'); + + await container.engine.attemptReconnect(ClientDisconnectReason.reconnectRetry); + + expect(container.engine.isClosed, isFalse); + expect(events.whereType(), isEmpty); + expect(events.whereType(), isEmpty); + }); + + test('a buffer wait started after disposal rejects even when the channel reports low', () async { + await container.room.dispose(); + + await expectLater( + container.engine.waitForBufferStatusLow(Reliability.reliable), + throwsA(isA()), + ); + }); + + test('disconnect followed by disposal announces closure only once', () async { + await container.engine.disconnect(); + await expectLater( + container.engine.waitForBufferStatusLow(Reliability.reliable), + throwsA(isA()), + ); + await container.room.dispose(); + + expect(events.whereType(), hasLength(1)); + }); +}