From f7b3006727480e263606d2bbfe58947038edc750 Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:13:23 +0800 Subject: [PATCH 1/4] Announce engine closure on terminal lifecycle paths --- .changes/engine-terminal-closure | 1 + lib/src/core/engine.dart | 21 +++++- test/core/engine_closing_test.dart | 116 +++++++++++++++++++++++++++++ 3 files changed, 134 insertions(+), 4 deletions(-) create mode 100644 .changes/engine-terminal-closure create mode 100644 test/core/engine_closing_test.dart diff --git a/.changes/engine-terminal-closure b/.changes/engine-terminal-closure new file mode 100644 index 000000000..1a8d6e5ce --- /dev/null +++ b/.changes/engine-terminal-closure @@ -0,0 +1 @@ +patch type="fixed" "Announce engine closure when a room is disposed or 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..7d87c37d0 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -242,7 +242,7 @@ class Engine extends Disposable with EventsEmittable { _setUpSignalListeners(); onDispose(() async { - _isClosed = true; + _markClosed(); await cleanUp(); await events.dispose(); await _signalListener.dispose(); @@ -417,6 +417,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 'Engine disconnected'; + } + final Completer completer = Completer(); if (isBufferStatusLow(kind) == true) { @@ -1151,7 +1156,7 @@ class Engine extends Disposable with EventsEmittable { if (_reconnectAttempts >= _reconnectCount) { logger.fine('reconnectAttempts exceeded, disconnecting...'); - _isClosed = true; + _markClosed(); await cleanUp(); events.emit( @@ -1263,6 +1268,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 +1683,18 @@ 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. + 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/test/core/engine_closing_test.dart b/test/core/engine_closing_test.dart new file mode 100644 index 000000000..fbde44407 --- /dev/null +++ b/test/core/engine_closing_test.dart @@ -0,0 +1,116 @@ +// 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 '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('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('Engine disconnected')); + }); + + test('disconnect followed by disposal announces closure only once', () async { + await container.engine.disconnect(); + await expectLater(container.engine.waitForBufferStatusLow(Reliability.reliable), throwsA('Engine disconnected')); + await container.room.dispose(); + + expect(events.whereType(), hasLength(1)); + }); +} From 21803ac11a12b265a8a23f9a44dbd4ad3953bb30 Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Fri, 30 Oct 2026 03:52:55 +0800 Subject: [PATCH 2/4] Signal engine closure before disposal callbacks --- .changes/engine-terminal-closure | 2 +- lib/src/core/engine.dart | 17 ++++++--- lib/src/core/room.dart | 7 ++++ test/core/engine_closing_test.dart | 57 ++++++++++++++++++++++++++++++ 4 files changed, 77 insertions(+), 6 deletions(-) diff --git a/.changes/engine-terminal-closure b/.changes/engine-terminal-closure index 1a8d6e5ce..0429b4483 100644 --- a/.changes/engine-terminal-closure +++ b/.changes/engine-terminal-closure @@ -1 +1 @@ -patch type="fixed" "Announce engine closure when a room is disposed or reconnection gives up, and reject data channel buffer waits started after closure" +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 7d87c37d0..757cf6f4b 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 { - _markClosed(); 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, { @@ -1156,7 +1162,7 @@ class Engine extends Disposable with EventsEmittable { if (_reconnectAttempts >= _reconnectCount) { logger.fine('reconnectAttempts exceeded, disconnecting...'); - _markClosed(); + markClosed(); await cleanUp(); events.emit( @@ -1268,7 +1274,7 @@ class Engine extends Disposable with EventsEmittable { unawaited(handleReconnect(ClientDisconnectReason.reconnectRetry)); } else { logger.fine('attemptReconnect: disconnecting...'); - _markClosed(); + markClosed(); // clean up before emitting, room's EngineDisconnectedEvent handler // drops the event while fullReconnectOnNext is still true and // cleanUp() is what resets it @@ -1685,7 +1691,8 @@ class Engine extends Disposable with EventsEmittable { // Announce terminal closure before tearing down transports or event listeners. // A full reconnect also disposes transports, but must keep the engine open. - void _markClosed() { + @internal + void markClosed() { if (_isClosed) return; _isClosed = true; events.emit(EngineClosingEvent()); @@ -1694,7 +1701,7 @@ class Engine extends Disposable with EventsEmittable { Future disconnect({ DisconnectReason reason = DisconnectReason.clientInitiated, }) async { - _markClosed(); + 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/engine_closing_test.dart b/test/core/engine_closing_test.dart index fbde44407..dbcbf3934 100644 --- a/test/core/engine_closing_test.dart +++ b/test/core/engine_closing_test.dart @@ -15,6 +15,8 @@ @Timeout(Duration(seconds: 5)) library; +import 'dart:async'; + import 'package:flutter_test/flutter_test.dart'; import 'package:livekit_client/livekit_client.dart'; @@ -49,6 +51,61 @@ void main() { 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', From 82ca96422004484f6204986ac1f3e6a599d79d53 Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Tue, 3 Nov 2026 01:07:14 +0800 Subject: [PATCH 3/4] Reject buffer waits with a LiveKit exception instead of a string waitForBufferStatusLow() threw, or completed its wait with, the plain string 'Engine disconnected' once the engine closed. Data stream writes surface that error, so an app catching LiveKitException missed it. Both paths now use UnexpectedConnectionState, which client-sdk-js also throws when its engine is closed. --- .changes/buffer-wait-typed-error | 1 + lib/src/core/engine.dart | 4 ++-- test/core/engine_closing_test.dart | 10 ++++++++-- 3 files changed, 11 insertions(+), 4 deletions(-) create mode 100644 .changes/buffer-wait-typed-error 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/lib/src/core/engine.dart b/lib/src/core/engine.dart index 757cf6f4b..118cb247b 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -425,7 +425,7 @@ 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 'Engine disconnected'; + throw UnexpectedConnectionState('Engine disconnected'); } final Completer completer = Completer(); @@ -435,7 +435,7 @@ class Engine extends Disposable with EventsEmittable { } else { onClosing() { if (!completer.isCompleted) { - completer.completeError('Engine disconnected'); + completer.completeError(UnexpectedConnectionState('Engine disconnected')); } } diff --git a/test/core/engine_closing_test.dart b/test/core/engine_closing_test.dart index dbcbf3934..7c5de99f3 100644 --- a/test/core/engine_closing_test.dart +++ b/test/core/engine_closing_test.dart @@ -160,12 +160,18 @@ void main() { 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('Engine disconnected')); + 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('Engine disconnected')); + await expectLater( + container.engine.waitForBufferStatusLow(Reliability.reliable), + throwsA(isA()), + ); await container.room.dispose(); expect(events.whereType(), hasLength(1)); From 8d4c725feac94eb039dd6461c368f7bc81bb6b24 Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Tue, 3 Nov 2026 19:21:42 +0800 Subject: [PATCH 4/4] Test disposing the room during a full reconnect The room tears down its transports before it closes the engine, so a full reconnect that resumes during that teardown used to open a new signal socket. The test pauses the reconnect and the dispose against each other and fails on main without the disposal ordering in this PR. --- .../disconnect_during_reconnect_test.dart | 37 +++++++++++++++++-- 1 file changed, 34 insertions(+), 3 deletions(-) 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'); + }); }