HS-1258249: fix uncatchable realtime 1008 error and endless 1-second reconnect loop

lib/src/realtime_mixin.dart

@@ -39,6 +39,8 @@ mixin RealtimeMixin {
   bool _pendingSocketRebuild = false;
   int? get closeCode => _websok?.closeCode;
   bool _reconnect = true;
+  AppwriteException? _fatalError;
+  bool _retryScheduled = false;
   int _retries = 0;
   StreamSubscription? _websocketSubscription;
   bool _creatingSocket = false;
@@ -90,7 +92,13 @@ mixin RealtimeMixin {
         _websok = await getWebSocket(uri);
         _lastUrl = uri.toString();
       } else {
-        if (_lastUrl == uri.toString() && _websok?.closeCode == null) {
+        // A rejected socket is unusable even while its `closeCode` is still
+        // null, because the close it was sent has not completed yet. Reusing
+        // it here would push the pending subscribes into a dying connection
+        // and leave `_fatalError` set, so the client would never recover.
+        if (_lastUrl == uri.toString() &&
+            _websok?.closeCode == null &&
+            _fatalError == null) {
           _sendPendingSubscribes();
           _creatingSocket = false;
           return;
@@ -99,7 +107,14 @@ mixin RealtimeMixin {
         _lastUrl = uri.toString();
         _websok = await getWebSocket(uri);
       }
-      _retries = 0;
+      // A freshly requested connection clears any recorded fatal state, so a
+      // caller that re-authenticates and subscribes again gets a working
+      // client back.
+      _reconnect = true;
+      _fatalError = null;
+      _retryScheduled = false;
+
+      await _websocketSubscription?.cancel();
       _websocketSubscription = _websok?.stream.listen((response) {
         final data = RealtimeResponse.fromJson(response);
         switch (data.type) {
@@ -128,6 +143,11 @@ mixin RealtimeMixin {
                 'queries': entry.value.queries,
               };
             }
+            // Reset the backoff only once the application-level handshake
+            // succeeded. Resetting it as soon as the transport connects makes
+            // every attempt look like the first one, so a server that keeps
+            // rejecting the connection is retried once a second forever.
+            _retries = 0;
             _appConnected = true;
             _sendPendingSubscribes();
             _flushPendingPresence();
@@ -159,14 +179,14 @@ mixin RealtimeMixin {
       }, onDone: () {
         _appConnected = false;
         _stopHeartbeat();
-        _retry();
+        _scheduleRetry();
       }, onError: (err, stack) {
         _appConnected = false;
         _stopHeartbeat();
         for (var subscription in _subscriptions.values) {
           subscription.controller.addError(err, stack);
         }
-        _retry();
+        _scheduleRetry();
       });
     } catch (e) {
       if (e is AppwriteException) {
@@ -183,9 +203,26 @@ mixin RealtimeMixin {
     }
   }
 
+  /// A single connection failure can surface more than once — the browser
+  /// channel emits error-then-done, and a server error frame is usually
+  /// followed by the stream terminating. Without this guard each of those
+  /// paths schedules its own reconnect and bumps `_retries`, so one failure
+  /// looks like several and rebuilds the socket twice.
+  void _scheduleRetry() {
+    if (_retryScheduled) return;
+    _retryScheduled = true;
+    _retry();
+  }
+
   void _retry() async {
-    if (!_reconnect || _websok?.closeCode == status.policyViolation) {
-      _reconnect = true;
+    // `closeCode` is an unreliable signal for a policy violation: it is null
+    // when the failure surfaces through the channel's onError path, and it is
+    // 1006/1000 when a proxy or tunnel tears the connection down without
+    // forwarding the server's 1008 close frame. `_fatalError` records the
+    // rejection itself, so it holds in all of those cases.
+    if (!_reconnect ||
+        _fatalError != null ||
+        _websok?.closeCode == status.policyViolation) {
       return;
     }
     _retries++;
@@ -382,9 +419,33 @@ mixin RealtimeMixin {
 
   void handleError(RealtimeResponse response) {
     if (response.data['code'] == status.policyViolation) {
-      throw AppwriteException(response.data["message"], response.data["code"]);
+      _handleFatalError(
+          AppwriteException(response.data["message"], response.data["code"]));
     } else {
-      _retry();
+      _scheduleRetry();
+    }
+  }
+
+  /// A policy violation (1008) means the server rejected this connection at the
+  /// application level, so reconnecting only gets rejected the same way.
+  ///
+  /// Throwing from here would escape into the zone's uncaught error handler —
+  /// this runs inside the WebSocket stream listener — leaving the socket open,
+  /// the retry loop running and the exception unreachable for app code.
+  /// Instead the connection is torn down and the exception is delivered to
+  /// every subscriber, so callers can react (e.g. re-authenticate and
+  /// subscribe again).
+  void _handleFatalError(AppwriteException error) {
+    _fatalError = error;
+    _reconnect = false;
+    _appConnected = false;
+    _stopHeartbeat();
+    final subscription = _websocketSubscription;
+    _websocketSubscription = null;
+    subscription?.cancel();
+    _websok?.sink.close(status.policyViolation, error.message);
+    for (var subscription in _subscriptions.values) {
+      subscription.controller.addError(error);
     }
   }
 

test/src/realtime_mixin_test.dart

@@ -0,0 +1,162 @@
+import 'dart:async';
+import 'dart:convert';
+
+import 'package:appwrite/src/client.dart';
+import 'package:appwrite/src/exception.dart';
+import 'package:appwrite/src/realtime_mixin.dart';
+import 'package:appwrite/src/realtime_subscription.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:web_socket_channel/web_socket_channel.dart';
+
+class FakeClient implements Client {
+  @override
+  Map<String, String> config = {'project': 'testProject'};
+
+  @override
+  String? get endPointRealtime => 'wss://demo.appwrite.io/v1';
+
+  @override
+  dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
+}
+
+class FakeWebSocketSink implements WebSocketSink {
+  final List<dynamic> sent = [];
+  bool closed = false;
+  int? closeCode;
+
+  @override
+  void add(dynamic data) => sent.add(data);
+
+  @override
+  Future close([int? closeCode, String? closeReason]) async {
+    closed = true;
+    this.closeCode = closeCode;
+  }
+
+  @override
+  dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
+}
+
+class FakeWebSocketChannel implements WebSocketChannel {
+  final StreamController<dynamic> _controller =
+      StreamController<dynamic>.broadcast();
+
+  @override
+  final FakeWebSocketSink sink = FakeWebSocketSink();
+
+  int? _closeCode;
+
+  @override
+  Stream<dynamic> get stream => _controller.stream;
+
+  @override
+  int? get closeCode => _closeCode;
+
+  @override
+  String? get closeReason => null;
+
+  @override
+  String? get protocol => null;
+
+  @override
+  Future<void> get ready => Future.value();
+
+  /// Simulate a frame sent by the server.
+  void emit(Map<String, dynamic> message) =>
+      _controller.add(jsonEncode(message));
+
+  /// Simulate the connection dropping. [code] is null when the socket dies
+  /// without a close frame reaching the client (e.g. a proxy/tunnel timeout).
+  void dropConnection({int? code}) {
+    _closeCode = code;
+    _controller.close();
+  }
+
+  @override
+  dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
+}
+
+class TestRealtime with RealtimeMixin {
+  TestRealtime(Client client, WebSocketFactory factory) {
+    this.client = client;
+    getWebSocket = factory;
+  }
+
+  RealtimeSubscription subscribe(List<Object> channels) =>
+      subscribeTo(channels);
+}
+
+void main() {
+  group('RealtimeMixin policy violation (1008)', () {
+    late List<FakeWebSocketChannel> channels;
+    late TestRealtime realtime;
+
+    setUp(() {
+      channels = [];
+      realtime = TestRealtime(FakeClient(), (uri) async {
+        final channel = FakeWebSocketChannel();
+        channels.add(channel);
+        return channel;
+      });
+    });
+
+    test(
+        'delivers the exception to subscribers and stops reconnecting when the '
+        'server rejects the connection', () async {
+      final errors = <Object>[];
+      final subscription = realtime.subscribe(['tables']);
+      subscription.stream.listen((_) {}, onError: errors.add);
+
+      // Let the socket open and register its listener.
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(channels, hasLength(1));
+
+      // The server rejects the connection at the application level, then the
+      // socket dies without the 1008 close code reaching the client — this is
+      // what happens behind a tunnel/proxy that drops the connection itself.
+      channels.first.emit({
+        'type': 'error',
+        'data': {'code': 1008, 'message': 'Server Error'},
+      });
+      await Future.delayed(Duration(milliseconds: 50));
+
+      // The exception must reach application code instead of being thrown as
+      // an uncatchable async error inside the WebSocket stream listener.
+      expect(errors, hasLength(1));
+      expect(errors.single, isA<AppwriteException>());
+      expect((errors.single as AppwriteException).code, 1008);
+
+      // The dead socket must be torn down.
+      expect(channels.first.sink.closed, isTrue);
+
+      channels.first.dropConnection();
+
+      // Retrying would be rejected the same way, so no reconnect must be
+      // scheduled — previously the client hammered the server once a second.
+      await Future.delayed(Duration(milliseconds: 1500));
+      expect(channels, hasLength(1));
+
+      // Once the application has reacted to the error (e.g. re-authenticated),
+      // subscribing again must open a fresh socket rather than reusing the
+      // rejected one, whose close has not completed yet.
+      realtime.subscribe(['tables']);
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(channels, hasLength(2));
+    });
+
+    test('still reconnects after a recoverable error', () async {
+      realtime.subscribe(['tables']);
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(channels, hasLength(1));
+
+      channels.first.emit({
+        'type': 'error',
+        'data': {'code': 1011, 'message': 'Server Error'},
+      });
+      channels.first.dropConnection(code: 1011);
+
+      await Future.delayed(Duration(milliseconds: 1500));
+      expect(channels, hasLength(2));
+    }, timeout: Timeout(Duration(seconds: 30)));
+  });
+}