fix(signaling): guard Transaction against double-complete race (WT-1046) (#1110)

* fix(signaling): guard Transaction against double-complete race (WT-1046)

Add _isDone flag and _finish() helper to Transaction so that all three
terminal paths (handleResponse, terminateByDisconnect, _onTimeout) are
symmetric. Previously only _onTimeout had the isCompleted guard; a late
server response arriving after a timeout could call _completer.complete()
on an already-completed Completer, throwing StateError and cascading into
WebtritSignalingTransactionTimeoutException for all ICE transactions.

Add unit tests covering all terminal paths and race scenarios using
fake_async for deterministic timer control.

* fix(signaling): log warning on duplicate Transaction completion attempt

* fix(signaling): address Copilot review — update docs and wrap catchError with unawaited

Author: SERDUN

packages/webtrit_signaling/lib/src/transaction.dart

@@ -1,7 +1,25 @@
 import 'dart:async';
 
+import 'package:logging/logging.dart';
+
 import 'exceptions.dart';
 
+final _logger = Logger('Transaction');
+
+/// Represents a single in-flight signaling request and its expected response.
+///
+/// A [Transaction] is created for every request sent to the signaling server.
+/// It holds a [Completer] that resolves when one of three terminal events occurs:
+///
+/// - [handleResponse] — the server replied within the timeout window.
+/// - [terminateByDisconnect] — the WebSocket closed before a reply arrived.
+/// - The internal timeout timer fires (after [timeoutDuration]).
+///
+/// Only the **first** terminal event takes effect. Subsequent calls to any of
+/// the three completion paths are logged at [Level.WARNING] and ignored via
+/// the [_isDone] guard, preventing a `StateError: Future already completed`
+/// if, for example, a late server response arrives after the timeout has
+/// already fired.
 class Transaction {
   static int _createCounter = 0;
 
@@ -14,6 +32,10 @@ class Transaction {
   final _completer = Completer<Map<String, dynamic>>();
   late final Timer _timer;
 
+  /// `true` once any terminal path ([handleResponse], [terminateByDisconnect],
+  /// or timeout) has run. Guards against double-completion of [_completer].
+  var _isDone = false;
+
   Transaction({required this.signalingClientId, String? id, required Duration timeoutDuration}) {
     if (id != null) {
       this.id = id;
@@ -28,22 +50,55 @@ class Transaction {
 
   Future<Map<String, dynamic>> get future => _completer.future;
 
+  /// Called when the server sends a response matching this transaction's [id].
+  ///
+  /// Completes [future] with [responseMessage]. No-op if the transaction has
+  /// already been resolved by a timeout or disconnect.
   void handleResponse(Map<String, dynamic> responseMessage) {
-    _timer.cancel();
+    if (_isDone) {
+      _logger.warning(
+        '$signalingClientId handleResponse called on already-completed transaction $id — ignoring late response',
+      );
+      return;
+    }
+    _finish();
     _completer.complete(responseMessage);
   }
 
+  /// Called when the WebSocket disconnects before a response is received.
+  ///
+  /// Completes [future] with a
+  /// [WebtritSignalingTransactionTerminateByDisconnectException]. No-op if the
+  /// transaction has already been resolved.
   void terminateByDisconnect([int? closeCode, String? closeReason]) {
-    _timer.cancel();
+    if (_isDone) {
+      _logger.warning(
+        '$signalingClientId terminateByDisconnect called on already-completed transaction $id — ignoring (code: $closeCode reason: $closeReason)',
+      );
+      return;
+    }
+    _finish();
     _completer.completeError(
       WebtritSignalingTransactionTerminateByDisconnectException(signalingClientId, id, closeCode, closeReason),
     );
   }
 
   void _onTimeout() {
-    if (_completer.isCompleted) {
+    if (_isDone) {
+      _logger.warning(
+        '$signalingClientId _onTimeout fired on already-completed transaction $id — ignoring late timeout',
+      );
       return;
     }
+    _finish();
     _completer.completeError(WebtritSignalingTransactionTimeoutException(signalingClientId, id), StackTrace.current);
   }
+
+  /// Marks the transaction as done and cancels the timeout timer.
+  ///
+  /// Must be called before completing [_completer] in every terminal path.
+  void _finish() {
+    _isDone = true;
+    _timer.cancel();
+  }
 }

packages/webtrit_signaling/test/src/transaction_test.dart

@@ -0,0 +1,135 @@
+import 'dart:async';
+
+import 'package:fake_async/fake_async.dart';
+import 'package:test/test.dart';
+
+import 'package:webtrit_signaling/src/exceptions.dart';
+import 'package:webtrit_signaling/src/transaction.dart';
+
+void main() {
+  const clientId = 0;
+  const timeout = Duration(seconds: 5);
+
+  // Helper to suppress unhandled future errors inside fakeAsync blocks.
+  // catchError must return the same type as the future.
+  void suppressError(Transaction tx) {
+    unawaited(tx.future.catchError((_) => <String, dynamic>{}));
+  }
+
+  group('Transaction.handleResponse', () {
+    test('completes future with response value', () async {
+      final tx = Transaction(signalingClientId: clientId, timeoutDuration: timeout);
+      tx.handleResponse({'response': 'ack'});
+      expect(await tx.future, {'response': 'ack'});
+    });
+
+    test('cancels timeout — timer does not fire after response', () {
+      fakeAsync((async) {
+        final tx = Transaction(signalingClientId: clientId, timeoutDuration: timeout);
+        tx.handleResponse({'response': 'ack'});
+
+        Object? error;
+        unawaited(
+          tx.future.catchError((e) {
+            error = e;
+            return <String, dynamic>{};
+          }),
+        );
+        async.elapse(timeout * 2);
+
+        expect(error, isNull);
+      });
+    });
+
+    test('second call is silently ignored — no StateError', () async {
+      final tx = Transaction(signalingClientId: clientId, timeoutDuration: timeout);
+      tx.handleResponse({'first': true});
+      expect(() => tx.handleResponse({'second': true}), returnsNormally);
+      expect(await tx.future, {'first': true});
+    });
+
+    test('late call after timeout does not throw StateError', () {
+      fakeAsync((async) {
+        final tx = Transaction(signalingClientId: clientId, timeoutDuration: timeout);
+        suppressError(tx);
+        async.elapse(timeout);
+        expect(() => tx.handleResponse({'late': 'response'}), returnsNormally);
+      });
+    });
+  });
+
+  group('Transaction.terminateByDisconnect', () {
+    test('completes future with disconnect error', () {
+      final tx = Transaction(signalingClientId: clientId, timeoutDuration: timeout);
+      tx.terminateByDisconnect(1000, 'normal');
+      expect(tx.future, throwsA(isA<WebtritSignalingTransactionTerminateByDisconnectException>()));
+    });
+
+    test('cancels timeout — timer does not fire after disconnect', () {
+      fakeAsync((async) {
+        final tx = Transaction(signalingClientId: clientId, timeoutDuration: timeout);
+
+        Object? capturedError;
+        unawaited(
+          tx.future.catchError((e) {
+            capturedError = e;
+            return <String, dynamic>{};
+          }),
+        );
+
+        tx.terminateByDisconnect();
+        async.flushMicrotasks();
+        async.elapse(timeout * 2);
+
+        expect(capturedError, isA<WebtritSignalingTransactionTerminateByDisconnectException>());
+      });
+    });
+
+    test('late call after timeout does not throw StateError', () {
+      fakeAsync((async) {
+        final tx = Transaction(signalingClientId: clientId, timeoutDuration: timeout);
+        suppressError(tx);
+        async.elapse(timeout);
+        expect(() => tx.terminateByDisconnect(1001, 'going away'), returnsNormally);
+      });
+    });
+  });
+
+  group('Transaction timeout', () {
+    test('completes future with timeout error after duration elapses', () {
+      fakeAsync((async) {
+        final tx = Transaction(signalingClientId: clientId, timeoutDuration: timeout);
+
+        Object? capturedError;
+        unawaited(
+          tx.future.catchError((e) {
+            capturedError = e;
+            return <String, dynamic>{};
+          }),
+        );
+
+        expect(capturedError, isNull);
+        async.elapse(timeout);
+        expect(capturedError, isA<WebtritSignalingTransactionTimeoutException>());
+      });
+    });
+
+    test('does not fire if handleResponse called first', () {
+      fakeAsync((async) {
+        final tx = Transaction(signalingClientId: clientId, timeoutDuration: timeout);
+        tx.handleResponse({'ok': true});
+
+        Object? error;
+        unawaited(
+          tx.future.catchError((e) {
+            error = e;
+            return <String, dynamic>{};
+          }),
+        );
+        async.elapse(timeout * 2);
+
+        expect(error, isNull);
+      });
+    });
+  });
+}