diff --git a/lib/src/realtime_mixin.dart b/lib/src/realtime_mixin.dart index 72edf5ee..97bec956 100644 --- a/lib/src/realtime_mixin.dart +++ b/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); } } diff --git a/test/src/realtime_mixin_test.dart b/test/src/realtime_mixin_test.dart new file mode 100644 index 00000000..81cd3a21 --- /dev/null +++ b/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 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 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 _controller = + StreamController.broadcast(); + + @override + final FakeWebSocketSink sink = FakeWebSocketSink(); + + int? _closeCode; + + @override + Stream get stream => _controller.stream; + + @override + int? get closeCode => _closeCode; + + @override + String? get closeReason => null; + + @override + String? get protocol => null; + + @override + Future get ready => Future.value(); + + /// Simulate a frame sent by the server. + void emit(Map 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 channels) => + subscribeTo(channels); +} + +void main() { + group('RealtimeMixin policy violation (1008)', () { + late List 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 = []; + 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()); + 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))); + }); +}