From af2ec748c8a4c7f737df6dd39f2aac4e75560b51 Mon Sep 17 00:00:00 2001 From: Claudear <262350598+claudear@users.noreply.github.com> Date: Tue, 11 Aug 2026 15:09:07 +0000 Subject: [PATCH 1/2] fix(realtime): surface 1008 policy violations instead of throwing into the zone MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A server-sent realtime `error` frame with code 1008 was rethrown from inside the WebSocket stream listener. A throw there is forwarded to the zone's uncaught-error handler, so it never reached the `try/catch` in `_createSocket`, never cancelled the subscription, never closed the socket and was never delivered to `subscription.stream.listen(onError:)` — app code had no way to catch it and re-authenticate. The only thing that stopped the reconnect loop afterwards was the incidental `_websok?.closeCode == status.policyViolation` check in `_retry`, which is unreliable: `closeCode` is null when the failure surfaces through the channel's onError path, and 1006/1000 when a proxy such as a Cloudflare tunnel terminates the connection without forwarding the server's 1008 close frame. In those cases the client reconnected into the same rejection once a second, forever. - `handleError` now records the fatal state, stops the heartbeat, cancels the subscription, closes the socket and delivers the `AppwriteException` to every subscriber. - `_retry` is gated on that recorded state rather than the racy `closeCode` check, and no longer resets `_reconnect = true` on its early return. - `_retries` is reset on the application-level `connected` event instead of right after the transport connects, so the backoff escalates when the server keeps rejecting the connection. - onError/onDone no longer schedule two reconnects for one failure, and the stale stream subscription is cancelled before being replaced. Co-Authored-By: Claude Opus 5 --- lib/src/realtime_mixin.dart | 63 ++++++++++-- test/src/realtime_mixin_test.dart | 155 ++++++++++++++++++++++++++++++ 2 files changed, 212 insertions(+), 6 deletions(-) create mode 100644 test/src/realtime_mixin_test.dart diff --git a/lib/src/realtime_mixin.dart b/lib/src/realtime_mixin.dart index 72edf5ee..147afbb8 100644 --- a/lib/src/realtime_mixin.dart +++ b/lib/src/realtime_mixin.dart @@ -39,6 +39,7 @@ mixin RealtimeMixin { bool _pendingSocketRebuild = false; int? get closeCode => _websok?.closeCode; bool _reconnect = true; + AppwriteException? _fatalError; int _retries = 0; StreamSubscription? _websocketSubscription; bool _creatingSocket = false; @@ -99,7 +100,22 @@ 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; + // onError and onDone both fire for a single failure on some platforms + // (the browser channel emits error-then-done), which would otherwise + // schedule two reconnects and double-count the retries. + var retryScheduled = false; + void scheduleRetry() { + if (retryScheduled) return; + retryScheduled = true; + _retry(); + } + + await _websocketSubscription?.cancel(); _websocketSubscription = _websok?.stream.listen((response) { final data = RealtimeResponse.fromJson(response); switch (data.type) { @@ -128,6 +144,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 +180,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) { @@ -184,8 +205,14 @@ mixin RealtimeMixin { } 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,12 +409,36 @@ 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(); } } + /// 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); + } + } + /// Fire-and-forget presence upsert. Records the latest payload in state so /// that — if the WebSocket isn't open yet, or later reconnects — the most /// recent presence is automatically (re)sent on the next `connected` event. diff --git a/test/src/realtime_mixin_test.dart b/test/src/realtime_mixin_test.dart new file mode 100644 index 00000000..1cc85f44 --- /dev/null +++ b/test/src/realtime_mixin_test.dart @@ -0,0 +1,155 @@ +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)); + }); + + 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))); + }); +} From ddc4568e0152ab373cfa8a0aa455a81055e40284 Mon Sep 17 00:00:00 2001 From: Claudear <262350598+claudear@users.noreply.github.com> Date: Tue, 11 Aug 2026 15:14:38 +0000 Subject: [PATCH 2/2] fix(realtime): don't reuse a rejected socket, and dedupe retries per failure Addresses two gaps in the previous commit found in review: - `_createSocket` reused the existing socket when the URL matched and `closeCode` was still null. After a policy violation that close is still in flight, so re-subscribing after re-authenticating pushed the pending subscribes into a dying connection and left `_fatalError` set, leaving the client permanently disconnected. The reuse branch now also requires that no fatal error is recorded. - `handleError`'s recoverable branch called `_retry()` directly, so an error frame followed by the stream terminating scheduled two reconnects and counted one failure twice. Retry deduplication moved from a closure local to `_createSocket` into `_scheduleRetry`, which all three paths now use. Co-Authored-By: Claude Opus 5 --- lib/src/realtime_mixin.dart | 36 ++++++++++++++++++++----------- test/src/realtime_mixin_test.dart | 7 ++++++ 2 files changed, 30 insertions(+), 13 deletions(-) diff --git a/lib/src/realtime_mixin.dart b/lib/src/realtime_mixin.dart index 147afbb8..97bec956 100644 --- a/lib/src/realtime_mixin.dart +++ b/lib/src/realtime_mixin.dart @@ -40,6 +40,7 @@ mixin RealtimeMixin { int? get closeCode => _websok?.closeCode; bool _reconnect = true; AppwriteException? _fatalError; + bool _retryScheduled = false; int _retries = 0; StreamSubscription? _websocketSubscription; bool _creatingSocket = false; @@ -91,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; @@ -105,15 +112,7 @@ mixin RealtimeMixin { // client back. _reconnect = true; _fatalError = null; - // onError and onDone both fire for a single failure on some platforms - // (the browser channel emits error-then-done), which would otherwise - // schedule two reconnects and double-count the retries. - var retryScheduled = false; - void scheduleRetry() { - if (retryScheduled) return; - retryScheduled = true; - _retry(); - } + _retryScheduled = false; await _websocketSubscription?.cancel(); _websocketSubscription = _websok?.stream.listen((response) { @@ -180,14 +179,14 @@ mixin RealtimeMixin { }, onDone: () { _appConnected = false; _stopHeartbeat(); - scheduleRetry(); + _scheduleRetry(); }, onError: (err, stack) { _appConnected = false; _stopHeartbeat(); for (var subscription in _subscriptions.values) { subscription.controller.addError(err, stack); } - scheduleRetry(); + _scheduleRetry(); }); } catch (e) { if (e is AppwriteException) { @@ -204,6 +203,17 @@ 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 { // `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 @@ -412,7 +422,7 @@ mixin RealtimeMixin { _handleFatalError( AppwriteException(response.data["message"], response.data["code"])); } else { - _retry(); + _scheduleRetry(); } } diff --git a/test/src/realtime_mixin_test.dart b/test/src/realtime_mixin_test.dart index 1cc85f44..81cd3a21 100644 --- a/test/src/realtime_mixin_test.dart +++ b/test/src/realtime_mixin_test.dart @@ -135,6 +135,13 @@ void main() { // 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 {