Reconnect with other tab on close

Author: simolus3

demos/django-todolist/README.md

@@ -19,13 +19,6 @@ username: testuser
 password: testpassword
 ```
 
-## WEB NOTes
-
-```
-flutter run -d web-server
-google-chrome --disable-web-security --user-data-dir='.dart_tool/.chrome'
-```
-
 # Service Configuration
 
 This demo can be used with cloud or local services.

demos/supabase-anonymous-auth/pubspec.yaml

@@ -25,5 +25,9 @@ dev_dependencies:
 
   flutter_lints: ^3.0.1
 
+dependency_overrides:
+  sqlite_async:
+    path: /home/simon/src/sqlite_async.dart/packages/sqlite_async
+
 flutter:
   uses-material-design: true

demos/supabase-edge-function-auth/pubspec.yaml

@@ -25,5 +25,9 @@ dev_dependencies:
 
   flutter_lints: ^3.0.1
 
+dependency_overrides:
+  sqlite_async:
+    path: /home/simon/src/sqlite_async.dart/packages/sqlite_async
+
 flutter:
   uses-material-design: true

demos/supabase-simple-chat/pubspec.yaml

@@ -54,6 +54,10 @@ dev_dependencies:
   # rules and activating additional ones.
   flutter_lints: ^3.0.1
 
+dependency_overrides:
+  sqlite_async:
+    path: /home/simon/src/sqlite_async.dart/packages/sqlite_async
+
 # For information on the generic Dart part of this file, see the
 # following page: https://dart.dev/tools/pub/pubspec
 

demos/supabase-todolist/pubspec.yaml

@@ -27,5 +27,9 @@ dev_dependencies:
 
   flutter_lints: ^3.0.1
 
+dependency_overrides:
+  sqlite_async:
+    path: /home/simon/src/sqlite_async.dart/packages/sqlite_async
+
 flutter:
   uses-material-design: true

packages/powersync/lib/src/database/web/web_powersync_database.dart

@@ -142,7 +142,7 @@ class PowerSyncDatabaseImpl
     final storage = BucketStorage(database);
     StreamingSync sync;
     // Try using a shared worker for the synchronization implementation to avoid
-    // duplicate work across tabs.
+    // duplicating work across tabs.
     try {
       sync = await SyncWorkerHandle.start(
         this,

packages/powersync/lib/src/web/sync_worker.dart

@@ -54,6 +54,7 @@ class _ConnectedClient {
   final _SyncWorker _worker;
 
   _SyncRunner? _runner;
+  StreamSubscription? _logSubscription;
 
   _ConnectedClient(MessagePort port, this._worker) {
     channel = WorkerCommunicationChannel(
@@ -73,6 +74,30 @@ class _ConnectedClient {
         }
       },
     );
+
+    _logSubscription = _logger.onRecord.listen((record) {
+      final msg = StringBuffer(
+          '[${record.loggerName}] ${record.level.name}: ${record.time}: ${record.message}');
+
+      if (record.error != null) {
+        msg
+          ..writeln()
+          ..write(record.error);
+      }
+      if (record.stackTrace != null) {
+        msg
+          ..writeln()
+          ..write(record.stackTrace);
+      }
+
+      channel.notify(SyncWorkerMessageType.logEvent, msg.toString().toJS);
+    });
+  }
+
+  void markClosed() {
+    _logSubscription?.cancel();
+    _runner?.unregisterClient(this);
+    _runner = null;
   }
 }
 
@@ -82,62 +107,39 @@ class _SyncRunner {
   final StreamGroup<_RunnerEvent> _group = StreamGroup();
   final StreamController<_RunnerEvent> _mainEvents = StreamController();
 
+  StreamingSync? sync;
+  _ConnectedClient? databaseHost;
+  final connections = <_ConnectedClient>[];
+
   _SyncRunner(this.identifier) {
     _group.add(_mainEvents.stream);
 
     Future(() async {
-      final connections = <_ConnectedClient>[];
-      StreamingSync? sync;
-
       await for (final event in _group.stream) {
         try {
           switch (event) {
             case _AddConnection(:final client):
               connections.add(client);
               if (sync == null) {
-                _logger.info('Sync setup: Requesting database');
-
-                // This is the first client, ask for a database connection
-                final connection = await client.channel.requestDatabase();
-                _logger.info('Sync setup: Connecting to endpoint');
-                final database = await WebSqliteConnection.connectToEndpoint((
-                  connectPort: connection.databasePort,
-                  connectName: connection.databaseName,
-                  lockName: connection.lockName,
-                ));
-                _logger.info('Sync setup: Has database, starting sync!');
-
-                // todo: Detect client going down (sqlite_web exposes this), fall
-                // back to other connection in that case.
-
-                sync = StreamingSyncImplementation(
-                  adapter: BucketStorage(database),
-                  credentialsCallback: client.channel.credentialsCallback,
-                  invalidCredentialsCallback:
-                      client.channel.invalidCredentialsCallback,
-                  uploadCrud: client.channel.uploadCrud,
-                  updateStream: powerSyncUpdateNotifications(
-                      database.updates ?? const Stream.empty()),
-                  retryDelay: Duration(seconds: 3),
-                  client: FetchClient(mode: RequestMode.cors),
-                  identifier: identifier,
-                );
-                sync.statusStream.listen((event) {
-                  _logger.fine('Broadcasting sync event: $event');
-                  for (final client in connections) {
-                    client.channel.notify(
-                        SyncWorkerMessageType.notifySyncStatus,
-                        SerializedSyncStatus.from(event));
-                  }
-                });
-                sync.streamingSync();
+                await _requestDatabase(client);
               }
             case _RemoveConnection(:final client):
               connections.remove(client);
               if (connections.isEmpty) {
                 await sync?.abort();
                 sync = null;
               }
+            case _ActiveDatabaseClosed():
+              _logger.info('Remote database closed, finding a new client');
+              sync?.abort();
+              sync = null;
+
+              final newHost = await _collectActiveClients();
+              if (newHost == null) {
+                _logger.info('No client remains');
+              } else {
+                await _requestDatabase(newHost);
+              }
           }
         } catch (e, s) {
           _logger.warning('Error handling $event', e, s);
@@ -146,6 +148,84 @@ class _SyncRunner {
     });
   }
 
+  /// Pings all current [connections], removing those that don't answer in 5s
+  /// (as they are likely closed tabs as well).
+  ///
+  /// Returns the first client that responds (without waiting for others).
+  Future<_ConnectedClient?> _collectActiveClients() async {
+    final candidates = connections.toList();
+    if (candidates.isEmpty) {
+      return null;
+    }
+
+    final firstResponder = Completer<_ConnectedClient?>();
+    var pendingRequests = candidates.length;
+
+    for (final candidate in candidates) {
+      candidate.channel.ping().then((_) {
+        pendingRequests--;
+        if (!firstResponder.isCompleted) {
+          firstResponder.complete(candidate);
+        }
+      }).timeout(const Duration(seconds: 5), onTimeout: () {
+        pendingRequests--;
+        candidate.markClosed();
+        if (pendingRequests == 0 && !firstResponder.isCompleted) {
+          // All requests have timed out, no connection remains
+          firstResponder.complete(null);
+        }
+      });
+    }
+
+    return firstResponder.future;
+  }
+
+  Future<void> _requestDatabase(_ConnectedClient client) async {
+    _logger.info('Sync setup: Requesting database');
+
+    // This is the first client, ask for a database connection
+    final connection = await client.channel.requestDatabase();
+    _logger.info('Sync setup: Connecting to endpoint');
+    final database = await WebSqliteConnection.connectToEndpoint((
+      connectPort: connection.databasePort,
+      connectName: connection.databaseName,
+      lockName: connection.lockName,
+    ));
+    _logger.info('Sync setup: Has database, starting sync!');
+    databaseHost = client;
+
+    database.closedFuture.then((_) {
+      _logger.fine('Detected closed client');
+      client.markClosed();
+
+      if (client == databaseHost) {
+        _logger
+            .info('Tab providing sync database has gone down, reconnecting...');
+        _mainEvents.add(const _ActiveDatabaseClosed());
+      }
+    });
+
+    sync = StreamingSyncImplementation(
+      adapter: BucketStorage(database),
+      credentialsCallback: client.channel.credentialsCallback,
+      invalidCredentialsCallback: client.channel.invalidCredentialsCallback,
+      uploadCrud: client.channel.uploadCrud,
+      updateStream: powerSyncUpdateNotifications(
+          database.updates ?? const Stream.empty()),
+      retryDelay: Duration(seconds: 3),
+      client: FetchClient(mode: RequestMode.cors),
+      identifier: identifier,
+    );
+    sync!.statusStream.listen((event) {
+      _logger.fine('Broadcasting sync event: $event');
+      for (final client in connections) {
+        client.channel.notify(SyncWorkerMessageType.notifySyncStatus,
+            SerializedSyncStatus.from(event));
+      }
+    });
+    sync!.streamingSync();
+  }
+
   void registerClient(_ConnectedClient client) {
     _mainEvents.add(_AddConnection(client));
   }
@@ -168,3 +248,7 @@ final class _RemoveConnection implements _RunnerEvent {
 
   _RemoveConnection(this.client);
 }
+
+final class _ActiveDatabaseClosed implements _RunnerEvent {
+  const _ActiveDatabaseClosed();
+}

packages/powersync/lib/src/web/sync_worker_protocol.dart

@@ -40,6 +40,11 @@ enum SyncWorkerMessageType {
   /// of [SerializedSyncStatus].
   notifySyncStatus,
 
+  /// Notifies clients about a log event emitted by the worker (typically only
+  /// used when workers were compiled in debug mode).
+  /// The payload is a [JSString].
+  logEvent,
+
   okResponse,
   errorResponse,
 }
@@ -254,6 +259,10 @@ final class WorkerCommunicationChannel {
         case SyncWorkerMessageType.notifySyncStatus:
           _events.add((type, message.payload));
           return;
+        case SyncWorkerMessageType.logEvent:
+          final msg = (message.payload as JSString).toDart;
+          _logger.info('[Sync Worker]: $msg');
+          return;
       }
 
       try {

packages/powersync/pubspec.yaml

@@ -40,7 +40,6 @@ dev_dependencies:
   shelf_static: ^1.1.2
   stream_channel: ^2.1.2
   path: ^1.8.3
-  js: ">=0.6.7 <0.8.0"
 
 platforms:
   android: