fix(realtime): cancel pending reconnect on disconnect after a dropped socket (#1477)

## What

Fixes three related problems in `RealtimeClient` around a socket that
dropped unexpectedly:

1. `disconnect()` now always cancels a pending reconnect, so
disconnecting after the socket has already dropped no longer silently
reopens the connection.
2. Automatic reconnects keep their exponential backoff instead of
resetting to the shortest delay on every attempt.
3. A manual `connect()` after an unexpected drop now actually reconnects
instead of being a no-op.

## Why

When the socket drops unexpectedly, `_onConnClose` sets `connState =
closed` and schedules a reconnect through
`reconnectTimer.scheduleTimeout()`. `conn` is not nulled on a drop, so a
later `disconnect()` still enters the `conn != null` branch, but
`oldState == closed` makes `shouldCloseSink` false, so the close block
that cancelled the timer was skipped. `conn` was nulled and the method
returned, yet the armed backoff timer still fired, ran `_reconnect()`
and reopened a connection the user explicitly closed.

Cancelling the timer on every `disconnect()` fixes that, but it cannot
use `reconnectTimer.reset()`: the reconnect path is `_reconnect ->
disconnect -> connect`, so `reset()` would zero the retry counter on
every attempt and defeat exponential backoff. During an outage the
client would hammer an unreachable server at the shortest interval
instead of backing off. A new `RetryTimer.cancel()` cancels the
scheduled timer without touching the retry counter, so a user disconnect
still stops a pending reconnect while automatic reconnects keep backing
off. The counter is still zeroed on a successful open through `reset()`
in `_onConnOpen`.

Finally, because `conn` is intentionally kept non-null after a drop so
`conn.closeCode` and `conn.closeReason` stay readable, the `if (conn !=
null) return` guard in `connect()` treated the dead channel as a live
connection and turned a manual `connect()` into a no-op. `connect()` now
only bails out when the connection is live or in progress. When `conn`
is set but `connState == closed`, it tears the stale connection down
through `disconnect()` first, which cancels the old stream subscription,
and then opens a fresh one.

## Not a breaking change

The public `conn` field is still left non-null after an unexpected drop,
so consumers reading `conn.closeCode` or `conn.closeReason` are
unaffected. `connState` transitions are unchanged: a socket that was
already `closed` still ends up `closed`, and an open socket still ends
up `disconnected` after `disconnect()`. `RetryTimer.reset()` keeps its
existing behavior and now delegates its cancel step to the new
`cancel()` method.

## Tests

Added three tests to the `disconnect` group:

- `cancels a pending reconnect after an unexpected drop`: drop the
socket so a reconnect is scheduled, then `disconnect()` and assert the
transport is never invoked again.
- `reconnects on a manual connect() after an unexpected drop`: drop the
socket, call `connect()` manually, and assert a fresh connection is
opened and `connState` becomes `open`.
- `grows the reconnect backoff across failed attempts`: fail the first
few connection attempts and assert the retry counter passed to
`reconnectAfterMs` grows as `1, 2, 3` instead of being reset to `1` on
every reconnect.

Each test fails without its corresponding fix.

---------

Co-authored-by: Lukas Klingsbo <lukas.klingsbo@gmail.com>

Author: AndroidPoet

packages/realtime_client/lib/src/realtime_client.dart

@@ -213,7 +213,10 @@ class RealtimeClient {
   @internal
   Future<void> connect() async {
     if (conn != null) {
-      return;
+      if (connState != SocketStates.closed) {
+        return;
+      }
+      await disconnect();
     }
 
     try {
@@ -300,9 +303,15 @@ class RealtimeClient {
           await conn.sink.close();
         }
         connState = SocketStates.disconnected;
-        reconnectTimer.reset();
         log('transport', 'disconnected', null, Level.FINE);
       }
+
+      // Cancel any reconnect scheduled by `_onConnClose`. When the socket has
+      // already dropped (`connState == closed`) the block above is skipped, so
+      // without this an armed backoff timer would fire after the user
+      // explicitly disconnected and silently reopen the connection.
+      reconnectTimer.cancel();
+
       this.conn = null;
       await _connectionSubscription?.cancel();
       _connectionSubscription = null;

packages/realtime_client/lib/src/retry_timer.dart

@@ -1,5 +1,7 @@
 import 'dart:async';
 
+import 'package:meta/meta.dart';
+
 typedef TimerCallback = void Function();
 typedef TimerCalculation = int Function(int tries);
 
@@ -34,7 +36,14 @@ class RetryTimer {
   /// Cancels any previous timer and reset tries
   void reset() {
     _tries = 0;
-    if (_timer != null) _timer!.cancel();
+    cancel();
+  }
+
+  /// Cancels any scheduled timer without resetting tries, so exponential
+  /// backoff is preserved across reconnect attempts.
+  @internal
+  void cancel() {
+    _timer?.cancel();
   }
 
   /// Cancels any previous scheduleTimeout and schedules callback

packages/realtime_client/test/socket_test.dart

@@ -322,6 +322,155 @@ void main() {
       expect(socket.conn, isNull);
     });
 
+    test('cancels a pending reconnect after an unexpected drop', () async {
+      final streamController = StreamController<dynamic>();
+      final mockedSocketChannel = MockIOWebSocketChannel();
+      final mockedSink = MockWebSocketSink();
+      var connectCount = 0;
+
+      when(() => mockedSocketChannel.ready).thenAnswer((_) => Future.value());
+      when(() => mockedSocketChannel.sink).thenReturn(mockedSink);
+      when(() => mockedSocketChannel.stream)
+          .thenAnswer((_) => streamController.stream);
+      when(() => mockedSink.close(any(), any()))
+          .thenAnswer((_) => Future.value());
+      when(() => mockedSink.close()).thenAnswer((_) => Future.value());
+
+      final mockedSocket = RealtimeClient(
+        socketEndpoint,
+        // Reconnect almost immediately so the test doesn't wait for the
+        // default backoff.
+        reconnectAfterMs: (tries) => 20,
+        transport: (url, headers) {
+          connectCount++;
+          return mockedSocketChannel;
+        },
+      );
+
+      await mockedSocket.connect();
+      expect(connectCount, 1);
+
+      // Simulate the server dropping the connection: `onDone` fires, the socket
+      // is marked closed and a reconnect is scheduled.
+      await streamController.close();
+      await Future.delayed(const Duration(milliseconds: 5));
+      expect(mockedSocket.connState, SocketStates.closed);
+
+      // The user disconnects explicitly while the socket is already closed.
+      await mockedSocket.disconnect();
+
+      // Wait past the reconnect delay; the scheduled reconnect must be canceled.
+      await Future.delayed(const Duration(milliseconds: 60));
+      expect(connectCount, 1,
+          reason: 'must not reopen after a user disconnect');
+    });
+
+    test('reconnects on a manual connect() after an unexpected drop', () async {
+      final firstController = StreamController<dynamic>();
+      final firstChannel = MockIOWebSocketChannel();
+      final firstSink = MockWebSocketSink();
+      when(() => firstChannel.ready).thenAnswer((_) => Future.value());
+      when(() => firstChannel.sink).thenReturn(firstSink);
+      when(() => firstChannel.stream).thenAnswer((_) => firstController.stream);
+      when(() => firstSink.close(any(), any()))
+          .thenAnswer((_) => Future.value());
+      when(() => firstSink.close()).thenAnswer((_) => Future.value());
+
+      final secondController = StreamController<dynamic>();
+      addTearDown(secondController.close);
+      final secondChannel = MockIOWebSocketChannel();
+      final secondSink = MockWebSocketSink();
+      when(() => secondChannel.ready).thenAnswer((_) => Future.value());
+      when(() => secondChannel.sink).thenReturn(secondSink);
+      when(() => secondChannel.stream)
+          .thenAnswer((_) => secondController.stream);
+      when(() => secondSink.close(any(), any()))
+          .thenAnswer((_) => Future.value());
+      when(() => secondSink.close()).thenAnswer((_) => Future.value());
+
+      var connectCount = 0;
+      final mockedSocket = RealtimeClient(
+        socketEndpoint,
+        // Large delay so the automatic reconnect stays dormant during the
+        // test and the manual reconnect below is what reopens the socket.
+        reconnectAfterMs: (tries) => 100000,
+        transport: (url, headers) {
+          connectCount++;
+          return connectCount == 1 ? firstChannel : secondChannel;
+        },
+      );
+
+      await mockedSocket.connect();
+      expect(connectCount, 1);
+      expect(mockedSocket.connState, SocketStates.open);
+
+      // Simulate the server dropping the connection.
+      await firstController.close();
+      await Future.delayed(const Duration(milliseconds: 5));
+      expect(mockedSocket.connState, SocketStates.closed);
+
+      // A manual reconnect must open a fresh connection instead of being a
+      // no-op because `conn` still references the dropped socket.
+      await mockedSocket.connect();
+      expect(connectCount, 2,
+          reason: 'manual connect() must reconnect after a drop');
+      expect(mockedSocket.connState, SocketStates.open);
+
+      await mockedSocket.disconnect();
+    });
+
+    test('grows the reconnect backoff across failed attempts', () async {
+      final triesSeen = <int>[];
+      var attempt = 0;
+
+      final failingChannel = MockIOWebSocketChannel();
+      final failingSink = MockWebSocketSink();
+      when(() => failingChannel.ready)
+          .thenAnswer((_) => Future.error(Exception('unavailable')));
+      when(() => failingChannel.sink).thenReturn(failingSink);
+      when(() => failingSink.close(any(), any()))
+          .thenAnswer((_) => Future.value());
+      when(() => failingSink.close()).thenAnswer((_) => Future.value());
+
+      final successController = StreamController<dynamic>();
+      addTearDown(successController.close);
+      final successChannel = MockIOWebSocketChannel();
+      final successSink = MockWebSocketSink();
+      when(() => successChannel.ready).thenAnswer((_) => Future.value());
+      when(() => successChannel.sink).thenReturn(successSink);
+      when(() => successChannel.stream)
+          .thenAnswer((_) => successController.stream);
+      when(() => successSink.close(any(), any()))
+          .thenAnswer((_) => Future.value());
+      when(() => successSink.close()).thenAnswer((_) => Future.value());
+
+      final mockedSocket = RealtimeClient(
+        socketEndpoint,
+        reconnectAfterMs: (tries) {
+          triesSeen.add(tries);
+          return 10;
+        },
+        transport: (url, headers) {
+          attempt++;
+          // Fail the first attempts so the client keeps retrying, then let it
+          // connect so the reconnect loop stops.
+          return attempt <= 3 ? failingChannel : successChannel;
+        },
+      );
+
+      await mockedSocket.connect();
+
+      // Wait for the failing attempts to cycle and the fourth to connect.
+      await Future.delayed(const Duration(milliseconds: 100));
+
+      // The retry counter must grow (1, 2, 3, ...) across reconnect attempts
+      // instead of being reset to 1 on every `disconnect()` in `_reconnect`.
+      expect(triesSeen.take(3), [1, 2, 3]);
+      expect(mockedSocket.connState, SocketStates.open);
+
+      await mockedSocket.disconnect();
+    });
+
     test('disconnecting an open connection', () async {
       await socket.connect();
       expect(socket.connState, SocketStates.open);