More worker debugging

Author: simolus3

demos/benchmarks/pubspec.lock

@@ -111,10 +111,10 @@ packages:
     dependency: "direct main"
     description:
       name: http
-      sha256: "2c11f3f94c687ee9bad77c171151672986360b2b001d109814ee7140b2cf261b"
+      sha256: bb2ce4590bc2667c96f318d68cac1b5a7987ec819351d32b1c987239a815e007
       url: "https://pub.dev"
     source: hosted
-    version: "1.4.0"
+    version: "1.5.0"
   http_parser:
     dependency: transitive
     description:
@@ -135,26 +135,26 @@ packages:
     dependency: transitive
     description:
       name: leak_tracker
-      sha256: "6bb818ecbdffe216e81182c2f0714a2e62b593f4a4f13098713ff1685dfb6ab0"
+      sha256: "8dcda04c3fc16c14f48a7bb586d4be1f0d1572731b6d81d51772ef47c02081e0"
       url: "https://pub.dev"
     source: hosted
-    version: "10.0.9"
+    version: "11.0.1"
   leak_tracker_flutter_testing:
     dependency: transitive
     description:
       name: leak_tracker_flutter_testing
-      sha256: f8b613e7e6a13ec79cfdc0e97638fddb3ab848452eff057653abd3edba760573
+      sha256: "1dbc140bb5a23c75ea9c4811222756104fbcd1a27173f0c34ca01e16bea473c1"
       url: "https://pub.dev"
     source: hosted
-    version: "3.0.9"
+    version: "3.0.10"
   leak_tracker_testing:
     dependency: transitive
     description:
       name: leak_tracker_testing
-      sha256: "6ba465d5d76e67ddf503e1161d1f4a6bc42306f9d66ca1e8f079a47290fb06d3"
+      sha256: "8d5a2d49f4a66b49744b23b018848400d23e54caf9463f4eb20df3eb8acb2eb1"
       url: "https://pub.dev"
     source: hosted
-    version: "3.0.1"
+    version: "3.0.2"
   lints:
     dependency: transitive
     description:
@@ -281,21 +281,21 @@ packages:
       path: "../../packages/powersync"
       relative: true
     source: path
-    version: "1.15.0"
+    version: "1.15.2"
   powersync_core:
     dependency: "direct overridden"
     description:
       path: "../../packages/powersync_core"
       relative: true
     source: path
-    version: "1.5.0"
+    version: "1.5.2"
   powersync_flutter_libs:
     dependency: "direct overridden"
     description:
       path: "../../packages/powersync_flutter_libs"
       relative: true
     source: path
-    version: "0.4.10"
+    version: "0.4.11"
   pub_semver:
     dependency: transitive
     description:
@@ -337,10 +337,10 @@ packages:
     dependency: transitive
     description:
       name: sqlite3
-      sha256: "310af39c40dd0bb2058538333c9d9840a2725ae0b9f77e4fd09ad6696aa8f66e"
+      sha256: f393d92c71bdcc118d6203d07c991b9be0f84b1a6f89dd4f7eed348131329924
       url: "https://pub.dev"
     source: hosted
-    version: "2.7.5"
+    version: "2.9.0"
   sqlite3_flutter_libs:
     dependency: transitive
     description:
@@ -353,18 +353,18 @@ packages:
     dependency: transitive
     description:
       name: sqlite3_web
-      sha256: "967e076442f7e1233bd7241ca61f3efe4c7fc168dac0f38411bdb3bdf471eb3c"
+      sha256: "0f6ebcb4992d1892ac5c8b5ecd22a458ab9c5eb6428b11ae5ecb5d63545844da"
       url: "https://pub.dev"
     source: hosted
-    version: "0.3.1"
+    version: "0.3.2"
   sqlite_async:
     dependency: "direct main"
     description:
       name: sqlite_async
-      sha256: a60e8d5c8df8e694933bd5a312c38393e79ad77d784bb91c6f38ba627bfb7aec
+      sha256: "6116bfc6aef6ce77730b478385ba4a58873df45721f6a9bc6ffabf39b6576e36"
       url: "https://pub.dev"
     source: hosted
-    version: "0.11.4"
+    version: "0.12.1"
   stack_trace:
     dependency: transitive
     description:
@@ -401,10 +401,10 @@ packages:
     dependency: transitive
     description:
       name: test_api
-      sha256: fb31f383e2ee25fbbfe06b40fe21e1e458d14080e3c67e7ba0acfde4df4e0bbd
+      sha256: "522f00f556e73044315fa4585ec3270f1808a4b186c936e612cab0b565ff1e00"
       url: "https://pub.dev"
     source: hosted
-    version: "0.7.4"
+    version: "0.7.6"
   typed_data:
     dependency: transitive
     description:
@@ -433,10 +433,10 @@ packages:
     dependency: transitive
     description:
       name: vector_math
-      sha256: "80b3257d1492ce4d091729e3a67a60407d227c27241d6927be0130c98e741803"
+      sha256: d530bd74fea330e6e364cda7a85019c434070188383e1cd8d9777ee586914c5b
       url: "https://pub.dev"
     source: hosted
-    version: "2.1.4"
+    version: "2.2.0"
   vm_service:
     dependency: transitive
     description:
@@ -470,5 +470,5 @@ packages:
     source: hosted
     version: "3.1.3"
 sdks:
-  dart: ">=3.7.0 <4.0.0"
+  dart: ">=3.8.0-0 <4.0.0"
   flutter: ">=3.27.0"

demos/supabase-todolist/lib/app_config_template.dart

@@ -8,5 +8,5 @@ class AppConfig {
       ''; // Optional. Only required when syncing attachments and using Supabase Storage. See packages/powersync_attachments_helper.
   // Whether the PowerSync instance uses sync streams to make fetching todo
   // items optional.
-  static const bool hasSyncStreams = true;
+  static const bool hasSyncStreams = false;
 }

demos/supabase-todolist/lib/widgets/todo_list_page.dart

@@ -132,7 +132,7 @@ class _SyncStreamTodosState extends State<_SyncStreamTodoListWidget> {
             false;
 
         if (!hasSynced) {
-          return const CircularProgressIndicator();
+          return const Center(child: CircularProgressIndicator());
         } else {
           return StreamBuilder(
             stream: widget.list.watchItems(),

packages/powersync_core/lib/src/sync/connection_manager.dart

@@ -255,7 +255,7 @@ final class ConnectionManager {
       },
     });
 
-    await _activeGroup.syncMutex.lock(() async {
+    await _activeGroup.syncConnectMutex.lock(() async {
       if (_abortActiveSync == null) {
         // Since we're not connected, update the offline sync status to reflect
         // the new subscription.

packages/powersync_core/lib/src/sync/instruction.dart

@@ -14,7 +14,8 @@ sealed class Instruction {
         EstablishSyncStream.fromJson(establish as Map<String, Object?>),
       {'FetchCredentials': final creds} =>
         FetchCredentials.fromJson(creds as Map<String, Object?>),
-      {'CloseSyncStream': _} => const CloseSyncStream(),
+      {'CloseSyncStream': final closeOptions as Map<String, Object?>} =>
+        CloseSyncStream(closeOptions['hide_disconnect'] as bool),
       {'FlushFileSystem': _} => const FlushFileSystem(),
       {'DidCompleteSync': _} => const DidCompleteSync(),
       _ => UnknownSyncInstruction(json)
@@ -142,7 +143,9 @@ final class FetchCredentials implements Instruction {
 }
 
 final class CloseSyncStream implements Instruction {
-  const CloseSyncStream();
+  final bool hideDisconnect;
+
+  const CloseSyncStream(this.hideDisconnect);
 }
 
 final class FlushFileSystem implements Instruction {

packages/powersync_core/lib/src/sync/streaming_sync.dart

@@ -309,7 +309,12 @@ class StreamingSyncImplementation implements StreamingSync {
   }
 
   Future<void> _rustStreamingSyncIteration() async {
-    await _ActiveRustStreamingIteration(this).syncIteration();
+    logger.info('Starting Rust sync iteration');
+    final response = await _ActiveRustStreamingIteration(this).syncIteration();
+    logger.info(
+        'Ending Rust sync iteration. Immediate restart: ${response.immediateRestart}');
+    // Note: With the current loop in streamingSync(), any return value that
+    // isn't an exception triggers an immediate restart.
   }
 
   Future<(List<BucketRequest>, Map<String, BucketDescription?>)>
@@ -610,7 +615,7 @@ final class _ActiveRustStreamingIteration {
   var _hadSyncLine = false;
 
   StreamSubscription<void>? _completedUploads;
-  final Completer<void> _completedStream = Completer();
+  final Completer<RustSyncIterationResult> _completedStream = Completer();
 
   _ActiveRustStreamingIteration(this.sync);
 
@@ -620,7 +625,7 @@ final class _ActiveRustStreamingIteration {
         .toList();
   }
 
-  Future<void> syncIteration() async {
+  Future<RustSyncIterationResult> syncIteration() async {
     try {
       await _control(
         'start',
@@ -632,7 +637,7 @@ final class _ActiveRustStreamingIteration {
         }),
       );
       assert(_completedStream.isCompleted, 'Should have started streaming');
-      await _completedStream.future;
+      return await _completedStream.future;
     } finally {
       _isActive = false;
       _completedUploads?.cancel();
@@ -655,10 +660,12 @@ final class _ActiveRustStreamingIteration {
     }).map(ReceivedLine.new);
   }
 
-  Future<void> _handleLines(EstablishSyncStream request) async {
+  Future<RustSyncIterationResult> _handleLines(
+      EstablishSyncStream request) async {
     final events = addBroadcast(
         _receiveLines(request.request), sync._nonLineSyncEvents.stream);
 
+    var needsImmediateRestart = false;
     loop:
     await for (final event in events) {
       if (!_isActive || sync.aborted) {
@@ -674,7 +681,8 @@ final class _ActiveRustStreamingIteration {
           await _control('line_text', line);
         case UploadCompleted():
           await _control('completed_upload');
-        case AbortCurrentIteration():
+        case AbortCurrentIteration(:final hideDisconnectState):
+          needsImmediateRestart = hideDisconnectState;
           break loop;
         case TokenRefreshComplete():
           await _control('refreshed_token');
@@ -683,6 +691,8 @@ final class _ActiveRustStreamingIteration {
               convert.json.encode(_encodeSubscriptions(currentSubscriptions)));
       }
     }
+
+    return (immediateRestart: needsImmediateRestart);
   }
 
   /// Triggers a local CRUD upload when the first sync line has been received.
@@ -736,10 +746,11 @@ final class _ActiveRustStreamingIteration {
             sync.logger.warning('Could not prefetch credentials', e, s);
           });
         }
-      case CloseSyncStream():
+      case CloseSyncStream(:final hideDisconnect):
         if (!sync.aborted) {
           _isActive = false;
-          sync._nonLineSyncEvents.add(const AbortCurrentIteration());
+          sync._nonLineSyncEvents
+              .add(AbortCurrentIteration(hideDisconnectState: hideDisconnect));
         }
       case FlushFileSystem():
         await sync.adapter.flushFileSystem();
@@ -751,6 +762,8 @@ final class _ActiveRustStreamingIteration {
   }
 }
 
+typedef RustSyncIterationResult = ({bool immediateRestart});
+
 sealed class SyncEvent {}
 
 final class ReceivedLine implements SyncEvent {
@@ -768,7 +781,14 @@ final class TokenRefreshComplete implements SyncEvent {
 }
 
 final class AbortCurrentIteration implements SyncEvent {
-  const AbortCurrentIteration();
+  /// Whether we should immediately disconnect and hide the `disconnected`
+  /// state.
+  ///
+  /// This is used when we're changing subscription, to hide the brief downtime
+  /// we have while reconnecting.
+  final bool hideDisconnectState;
+
+  const AbortCurrentIteration({this.hideDisconnectState = false});
 }
 
 final class HandleChangedSubscriptions implements SyncEvent {

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

@@ -229,6 +229,8 @@ class _SyncRunner {
     final before = currentStreams.toSet();
     final after = connections.values.flattenedToSet;
     if (!const SetEquality<SubscribedStream>().equals(before, after)) {
+      _logger.info(
+          'Subscriptions across tabs have changed, checking whether a reconnect is necessary');
       currentStreams = after.toList();
       sync?.updateSubscriptions(currentStreams);
     }
@@ -320,6 +322,7 @@ class _SyncRunner {
       client: BrowserClient(),
       identifier: identifier,
       activeSubscriptions: currentStreams,
+      logger: _logger,
     );
     sync!.statusStream.listen((event) {
       _logger.fine('Broadcasting sync event: $event');