fix: prevent keepalive write-after-close and add regression test

Author: SERDUN

packages/webtrit_signaling/lib/src/exceptions.dart

@@ -52,6 +52,15 @@ class WebtritSignalingTransactionTimeoutException extends WebtritSignalingTransa
   const WebtritSignalingTransactionTimeoutException(super.id, super.transactionId);
 }
 
+class WebtritSignalingBadStateException extends WebtritSignalingException {
+  const WebtritSignalingBadStateException(super.id, this.error);
+
+  final StateError error;
+
+  @override
+  String toString() => '${super.toString()}, stateError: $error';
+}
+
 class WebtritSignalingKeepaliveTransactionTimeoutException extends WebtritSignalingTransactionTimeoutException {
   const WebtritSignalingKeepaliveTransactionTimeoutException(super.id, super.transactionId);
 }

packages/webtrit_signaling/lib/src/webtrit_signaling_client.dart

@@ -271,7 +271,15 @@ class WebtritSignalingClient {
   void _addData(dynamic data) {
     _logger.fine(() => '_addData add: $data');
 
-    _wsc.sink.add(data);
+    try {
+      if (_wsc.closeCode != null) {
+        throw StateError('Cannot add event after closing (detected via closeCode)');
+      }
+
+      _wsc.sink.add(data);
+    } on StateError catch (e) {
+      throw WebtritSignalingBadStateException(_id, e);
+    }
   }
 
   void _cleanupTransactions(int? code, String? reason) {
@@ -301,9 +309,23 @@ class WebtritSignalingClient {
       final elapsed = await _executeKeepaliveTransaction(defaultExecuteTransactionTimeoutDuration);
       _logger.finest('handshake keepalive latency: $elapsed');
 
-      _startKeepaliveTimer();
+      if (_wsc.closeCode == null) {
+        _startKeepaliveTimer();
+      }
     } on WebtritSignalingTransactionTimeoutException catch (e, stackTrace) {
       _onError(WebtritSignalingKeepaliveTransactionTimeoutException(e.id, e.transactionId), stackTrace);
+    } on WebtritSignalingBadStateException {
+      _logger.fine('Keepalive stopped gracefully due to closed socket.');
+      // Catches the specific exception thrown when attempting to write to a closed socket.
+      // This indicates a race condition where the Keepalive timer triggered shortly after
+      // the socket was closed but before the timer could be cancelled.
+      //
+      // Since the socket is already in a terminal state, the standard disconnection
+      // handlers (onDone/onDisconnect) are responsible for the lifecycle management.
+      // Reporting this as an error would be redundant and potentially misleading.
+      //
+      // Gracefully terminate the recursive Keepalive loop here.
+      return;
     } catch (error, stackTrace) {
       _onError(error, stackTrace);
     }

packages/webtrit_signaling/pubspec.yaml

@@ -19,3 +19,5 @@ dependencies:
 dev_dependencies:
   lints: ^6.0.0
   test: ^1.26.3
+  mocktail: ^1.0.4
+  fake_async: ^1.3.1

packages/webtrit_signaling/test/webtrit_signaling_integration_test.dart

@@ -0,0 +1,93 @@
+import 'dart:async';
+import 'dart:convert';
+
+import 'package:fake_async/fake_async.dart';
+import 'package:mocktail/mocktail.dart';
+import 'package:test/test.dart';
+import 'package:web_socket_channel/web_socket_channel.dart';
+
+import 'package:webtrit_signaling/webtrit_signaling.dart';
+
+class MockWebSocketChannel extends Mock implements WebSocketChannel {}
+
+class MockWebSocketSink extends Mock implements WebSocketSink {}
+
+void main() {
+  group('WebtritSignalingClient Keepalive Race Condition', () {
+    late MockWebSocketChannel mockChannel;
+    late MockWebSocketSink mockSink;
+    late StreamController<dynamic> streamController;
+    late WebtritSignalingClient client;
+
+    setUp(() {
+      mockChannel = MockWebSocketChannel();
+      mockSink = MockWebSocketSink();
+      streamController = StreamController<dynamic>();
+
+      when(() => mockChannel.sink).thenReturn(mockSink);
+      when(() => mockChannel.stream).thenAnswer((_) => streamController.stream);
+      when(() => mockChannel.closeCode).thenReturn(null);
+      when(() => mockSink.close(any(), any())).thenAnswer((_) => Future.value());
+    });
+
+    tearDown(() {
+      streamController.close();
+    });
+
+    test('should gracefully terminate keepalive loop without error when timer fires on a closed socket', () {
+      fakeAsync((async) {
+        client = WebtritSignalingClient.inner(mockChannel);
+
+        var errorReported = false;
+        Object? reportedError;
+
+        client.listen(
+          onStateHandshake: (_) {},
+          onEvent: (_) {},
+          onError: (e, s) {
+            errorReported = true;
+            reportedError = e;
+          },
+          onDisconnect: (_, _) {},
+        );
+
+        var isPhysicalSocketClosed = false;
+
+        // Throws StateError to simulate writing to a closed sink.
+        when(() => mockSink.add(any())).thenAnswer((invocation) {
+          if (isPhysicalSocketClosed) {
+            throw StateError('Bad state: Cannot add event after closing.');
+          }
+        });
+
+        // Initiates the Keepalive timer (100ms interval).
+        final handshakeData = jsonEncode({
+          'handshake': 'state',
+          'timestamp': 1705322000000,
+          'keepalive_interval': 100,
+          'registration': {'status': 'registered'},
+          'lines': [],
+          'user_active_calls': [],
+          'presence_contacts_info': {},
+          'guest_line': null,
+        });
+
+        streamController.add(handshakeData);
+        async.flushMicrotasks();
+
+        // Simulates a physical connection drop without explicit client disconnection.
+        // The client remains unaware of the transport failure, allowing the timer to persist.
+        isPhysicalSocketClosed = true;
+
+        // Advances time to trigger the Keepalive timer (100ms + buffer).
+        async.elapse(const Duration(milliseconds: 200));
+
+        // Confirms the write was attempted (validating the race condition logic executed).
+        verify(() => mockSink.add(any())).called(greaterThan(0));
+
+        // Ensures the internal exception was handled silently and not reported.
+        expect(errorReported, isFalse, reason: 'Race condition caused an exception to leak to onError: $reportedError');
+      });
+    });
+  });
+}