fix(realtime): prevent null check crash in connect() during rapid lifecycle transitions (#1321)

* fix(realtime): prevent null check crash in connect() during rapid lifecycle transitions

Capture the WebSocketChannel in a local variable before awaiting ready,
then add a staleness guard (conn != localConn) after the await to bail
out if disconnect() ran concurrently. This prevents the TypeError when
conn is nullified between the await and the stream.listen() call.

Also replace the conn!.sink.done access in onResumed() with a stored
_disconnectFuture to eliminate a second NPE risk during lifecycle
flapping.

Fixes #1320

* style(realtime): use explicit type for localConn variable

* fix(realtime): add staleness guard in catch block and check connState in post-await guard

Address review feedback:
- Add conn != localConn guard inside the catch block to avoid firing
  error handlers for a stale connection attempt.
- Strengthen the post-await guard to also check connState == connecting,
  preventing connect() from proceeding if disconnect() has already set
  connState to disconnecting but hasn't yet nulled conn.

* style(realtime): fix formatting in channel_test.dart for 80-char line limit

* fix(supabase_flutter): clear _disconnectFuture after completion instead of eagerly

Address review feedback from icnahom: if a second resumed event fires
while disconnect is still in progress (e.g. paused → resumed → inactive
→ resumed), the eagerly-nulled _disconnectFuture causes the second
resumed to fall into the else branch where connect() no-ops because
conn is still non-null.

Move _disconnectFuture = null into the .then() callback with a guard to
avoid overwriting a newer future set by a subsequent paused event.

* test(supabase_flutter): add lifecycle reconnection tests

Add transport field to RealtimeClientOptions (non-breaking, additive)
so tests can inject a mock WebSocket transport. Add two lifecycle tests:
- paused → resumed reconnects after disconnect completes
- paused → resumed → inactive → resumed still reconnects (validates
  _disconnectFuture is not eagerly cleared)

Author: brunovsiqueira

packages/realtime_client/lib/src/realtime_client.dart

@@ -178,11 +178,16 @@ class RealtimeClient {
       log('transport', 'connecting to $endPointURL', null);
       log('transport', 'connecting', null, Level.FINE);
       connState = SocketStates.connecting;
-      conn = transport(endPointURL, headers);
+      final WebSocketChannel localConn = transport(endPointURL, headers);
+      conn = localConn;
 
       try {
-        await conn!.ready;
+        await localConn.ready;
       } catch (error) {
+        // Bail out if disconnect() ran or a new connect() started during await
+        if (conn != localConn) {
+          return;
+        }
         // Don't schedule a reconnect and emit error if connection has been
         // closed by the user or [disconnect] waits for the connection to be
         // ready before closing it.
@@ -195,10 +200,15 @@ class RealtimeClient {
         return;
       }
 
+      // Guard: bail out if disconnect() ran during the await
+      if (conn != localConn || connState != SocketStates.connecting) {
+        return;
+      }
+
       connState = SocketStates.open;
 
       _onConnOpen();
-      conn!.stream.listen(
+      localConn.stream.listen(
         // incoming messages
         (message) => onConnMessage(message as String),
         onError: _onConnError,

packages/realtime_client/test/channel_test.dart

@@ -221,7 +221,8 @@ void main() {
       expect(callbackOtherCalled, 1);
     });
 
-    test('maintains type safety after off() - '
+    test(
+        'maintains type safety after off() - '
         'reproduces web hot restart issue', () {
       // This test reproduces the issue where .where().toList() returns
       // List<dynamic> on Flutter web during hot restart, causing a

packages/realtime_client/test/socket_test.dart

@@ -1,3 +1,4 @@
+import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 
@@ -643,4 +644,131 @@ void main() {
       verifyNever(() => mockedSink.add(any()));
     });
   });
+
+  group('connect/disconnect race condition', () {
+    test(
+        'connect does not crash if disconnect nullifies conn during await ready',
+        () async {
+      final readyCompleter = Completer<void>();
+      final mockedSocketChannel = MockIOWebSocketChannel();
+      final mockedSink = MockWebSocketSink();
+
+      when(() => mockedSocketChannel.ready)
+          .thenAnswer((_) => readyCompleter.future);
+      when(() => mockedSocketChannel.sink).thenReturn(mockedSink);
+      when(() => mockedSink.close(any(), any()))
+          .thenAnswer((_) => Future.value());
+      when(() => mockedSink.close()).thenAnswer((_) => Future.value());
+
+      final socket = RealtimeClient(
+        socketEndpoint,
+        transport: (url, headers) => mockedSocketChannel,
+      );
+
+      // Start connect — it will suspend at await ready
+      final connectFuture = socket.connect();
+
+      // Start disconnect (also suspends on ready since state is connecting)
+      final disconnectFuture = socket.disconnect();
+
+      // Now complete the ready future — both connect and disconnect can proceed
+      readyCompleter.complete();
+      await disconnectFuture;
+      await connectFuture;
+
+      // Should NOT have transitioned to open because disconnect nullified conn
+      expect(socket.connState, isNot(SocketStates.open));
+      expect(socket.conn, isNull);
+    });
+
+    test('connect bails out when connState changes during await ready',
+        () async {
+      final readyCompleter = Completer<void>();
+      final mockedSocketChannel = MockIOWebSocketChannel();
+      final mockedSink = MockWebSocketSink();
+
+      when(() => mockedSocketChannel.ready)
+          .thenAnswer((_) => readyCompleter.future);
+      when(() => mockedSocketChannel.sink).thenReturn(mockedSink);
+      when(() => mockedSink.close(any(), any()))
+          .thenAnswer((_) => Future.value());
+      when(() => mockedSink.close()).thenAnswer((_) => Future.value());
+
+      final socket = RealtimeClient(
+        socketEndpoint,
+        transport: (url, headers) => mockedSocketChannel,
+      );
+
+      // Start connect
+      final connectFuture = socket.connect();
+
+      // Start disconnect — also awaits ready
+      final disconnectFuture = socket.disconnect();
+
+      // Complete ready — both proceed
+      readyCompleter.complete();
+      await disconnectFuture;
+      await connectFuture;
+
+      expect(socket.connState, isNot(SocketStates.open));
+    });
+
+    test('rapid connect-disconnect-connect cycle does not crash', () async {
+      final readyCompleter1 = Completer<void>();
+      final mockedSocketChannel1 = MockIOWebSocketChannel();
+      final mockedSink1 = MockWebSocketSink();
+
+      when(() => mockedSocketChannel1.ready)
+          .thenAnswer((_) => readyCompleter1.future);
+      when(() => mockedSocketChannel1.sink).thenReturn(mockedSink1);
+      when(() => mockedSink1.close(any(), any()))
+          .thenAnswer((_) => Future.value());
+      when(() => mockedSink1.close()).thenAnswer((_) => Future.value());
+
+      final readyCompleter2 = Completer<void>();
+      final mockedSocketChannel2 = MockIOWebSocketChannel();
+      final mockedSink2 = MockWebSocketSink();
+      final streamController2 = StreamController<dynamic>.broadcast();
+
+      when(() => mockedSocketChannel2.ready)
+          .thenAnswer((_) => readyCompleter2.future);
+      when(() => mockedSocketChannel2.sink).thenReturn(mockedSink2);
+      when(() => mockedSocketChannel2.stream)
+          .thenAnswer((_) => streamController2.stream);
+      when(() => mockedSink2.close(any(), any()))
+          .thenAnswer((_) => Future.value());
+      when(() => mockedSink2.close()).thenAnswer((_) => Future.value());
+
+      var callCount = 0;
+      final socket = RealtimeClient(
+        socketEndpoint,
+        transport: (url, headers) {
+          callCount++;
+          if (callCount == 1) return mockedSocketChannel1;
+          return mockedSocketChannel2;
+        },
+      );
+
+      // First connect — suspends at await ready
+      final connectFuture1 = socket.connect();
+
+      // Start disconnect (also suspends on ready)
+      final disconnectFuture = socket.disconnect();
+
+      // Complete the first ready — both proceed, connect bails out
+      readyCompleter1.complete();
+      await disconnectFuture;
+      await connectFuture1;
+
+      // Second connect with a fresh mock
+      readyCompleter2.complete();
+      await socket.connect();
+
+      expect(socket.connState, SocketStates.open);
+      expect(socket.conn, mockedSocketChannel2);
+
+      await socket.disconnect();
+      await streamController2.close();
+    });
+  });
 }

packages/supabase/lib/src/realtime_client_options.dart

@@ -17,10 +17,14 @@ class RealtimeClientOptions {
   /// the timeout to trigger push timeouts
   final Duration? timeout;
 
+  /// Custom WebSocket transport factory for the RealtimeClient.
+  final WebSocketTransport? transport;
+
   /// {@macro realtime_client_options}
   const RealtimeClientOptions({
     this.eventsPerSecond,
     this.logLevel,
     this.timeout,
+    this.transport,
   });
 }

packages/supabase/lib/src/supabase_client.dart

@@ -341,6 +341,7 @@ class SupabaseClient {
       httpClient: _authHttpClient,
       timeout: options.timeout ?? RealtimeConstants.defaultTimeout,
       customAccessToken: accessToken,
+      transport: options.transport,
     );
   }
 

packages/supabase_flutter/lib/src/supabase.dart

@@ -98,7 +98,8 @@ class Supabase with WidgetsBindingObserver {
       _instance._logSubscription = Logger('supabase').onRecord.listen((record) {
         if (record.level >= Level.INFO) {
           debugPrint(
-              '${record.loggerName}: ${record.level.name}: ${record.message} ${record.error ?? ""}');
+            '${record.loggerName}: ${record.level.name}: ${record.message} ${record.error ?? ""}',
+          );
         }
       });
     }
@@ -138,9 +139,7 @@ class Supabase with WidgetsBindingObserver {
       // Wrap `recoverSession()` in a `CancelableOperation` so that it can be canceled in dispose
       // if still in progress
       _instance._restoreSessionCancellableOperation =
-          CancelableOperation.fromFuture(
-        supabaseAuth.recoverSession(),
-      );
+          CancelableOperation.fromFuture(supabaseAuth.recoverSession());
     }
 
     _log.info('***** Supabase init completed *****');
@@ -172,6 +171,8 @@ class Supabase with WidgetsBindingObserver {
 
   CancelableOperation<void>? _realtimeReconnectOperation;
 
+  Future<void>? _disconnectFuture;
+
   StreamSubscription? _logSubscription;
 
   /// Dispose the instance to free up resources.
@@ -197,7 +198,7 @@ class Supabase with WidgetsBindingObserver {
   }) {
     final headers = {
       ...Constants.defaultHeaders,
-      if (customHeaders != null) ...customHeaders
+      if (customHeaders != null) ...customHeaders,
     };
     client = SupabaseClient(
       supabaseUrl,
@@ -229,41 +230,52 @@ class Supabase with WidgetsBindingObserver {
       case AppLifecycleState.detached:
       case AppLifecycleState.paused:
         _realtimeReconnectOperation?.cancel();
-        Supabase.instance.client.realtime.disconnect();
+        _disconnectFuture = Supabase.instance.client.realtime.disconnect();
       default:
     }
   }
 
   Future<void> onResumed() async {
     final realtime = Supabase.instance.client.realtime;
     if (realtime.channels.isNotEmpty) {
-      if (realtime.connState == SocketStates.disconnecting &&
-          realtime.conn != null) {
-        // If the socket is still disconnecting from e.g.
+      final disconnectFuture = _disconnectFuture;
+      if (disconnectFuture != null) {
+        // If a disconnect is still in progress from e.g.
         // [AppLifecycleState.paused] we should wait for it to finish before
-        // reconnecting.
+        // reconnecting. This avoids accessing conn! which may be nullified.
+        //
+        // We clear _disconnectFuture only after the future completes (not
+        // eagerly) so that a second resumed event (e.g. inactive → resumed)
+        // still sees the in-progress disconnect and waits for it instead of
+        // falling through to the else branch where connect() would no-op.
 
         bool cancel = false;
-        final connectFuture = realtime.conn!.sink.done.then(
-          (_) async {
-            // Make this connect cancelable so that it does not connect if the
-            // disconnect took so long that the app is already in background
-            // again.
+        final connectFuture = disconnectFuture.then((_) async {
+          // Only clear if not replaced by a newer disconnect from a
+          // subsequent paused event.
+          if (_disconnectFuture == disconnectFuture) {
+            _disconnectFuture = null;
+          }
 
-            if (!cancel) {
+          // Make this connect cancelable so that it does not connect if the
+          // disconnect took so long that the app is already in background
+          // again.
+          if (!cancel) {
+            // ignore: invalid_use_of_internal_member
+            await realtime.connect();
+            for (final channel in realtime.channels) {
               // ignore: invalid_use_of_internal_member
-              await realtime.connect();
-              for (final channel in realtime.channels) {
+              if (channel.isJoined) {
                 // ignore: invalid_use_of_internal_member
-                if (channel.isJoined) {
-                  // ignore: invalid_use_of_internal_member
-                  channel.forceRejoin();
-                }
+                channel.forceRejoin();
               }
             }
-          },
-          onError: (error) {},
-        );
+          }
+        }, onError: (error) {
+          if (_disconnectFuture == disconnectFuture) {
+            _disconnectFuture = null;
+          }
+        });
         _realtimeReconnectOperation = CancelableOperation.fromFuture(
           connectFuture,
           onCancel: () => cancel = true,

packages/supabase_flutter/pubspec.yaml

@@ -30,6 +30,7 @@ dev_dependencies:
     sdk: flutter
   flutter_lints: ^3.0.1
   path: ^1.8.3
+  web_socket_channel: '>=2.3.0 <4.0.0'
 
 platforms:
   android: