implement sync locks

Author: stevensJourney

demos/supabase-anonymous-auth/pubspec.lock

@@ -350,7 +350,7 @@ packages:
       path: "../../packages/powersync"
       relative: true
     source: path
-    version: "1.3.0-alpha.7"
+    version: "1.3.0-alpha.8"
   realtime_client:
     dependency: transitive
     description:
@@ -479,10 +479,11 @@ packages:
   sqlite_async:
     dependency: "direct main"
     description:
-      name: sqlite_async
-      sha256: "7c5a9bec86b6f5b7511b9ba30974fa7ea470aee2dc0d5b7021f6321a439a8d63"
-      url: "https://pub.dev"
-    source: hosted
+      path: "packages/sqlite_async"
+      ref: "feat/navigator-locks"
+      resolved-ref: "588d92f3dc98ab2629d2b420ad607639b835214b"
+      url: "https://github.com/powersync-ja/sqlite_async.dart.git"
+    source: git
     version: "0.8.0"
   stack_trace:
     dependency: transitive

demos/supabase-anonymous-auth/pubspec.yaml

@@ -16,7 +16,11 @@ dependencies:
   supabase_flutter: ^2.0.2
   path: ^1.8.3
   logging: ^1.2.0
-  sqlite_async: ^0.8.0
+  sqlite_async:
+    git:
+      url: https://github.com/powersync-ja/sqlite_async.dart.git
+      ref: feat/navigator-locks
+      path: packages/sqlite_async
   universal_io: ^2.2.2
 
 dev_dependencies:

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

@@ -350,7 +350,7 @@ packages:
       path: "../../packages/powersync"
       relative: true
     source: path
-    version: "1.3.0-alpha.7"
+    version: "1.3.0-alpha.8"
   realtime_client:
     dependency: transitive
     description:
@@ -479,10 +479,11 @@ packages:
   sqlite_async:
     dependency: "direct main"
     description:
-      name: sqlite_async
-      sha256: "7c5a9bec86b6f5b7511b9ba30974fa7ea470aee2dc0d5b7021f6321a439a8d63"
-      url: "https://pub.dev"
-    source: hosted
+      path: "packages/sqlite_async"
+      ref: "feat/navigator-locks"
+      resolved-ref: "588d92f3dc98ab2629d2b420ad607639b835214b"
+      url: "https://github.com/powersync-ja/sqlite_async.dart.git"
+    source: git
     version: "0.8.0"
   stack_trace:
     dependency: transitive

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

@@ -16,7 +16,11 @@ dependencies:
   supabase_flutter: ^2.0.2
   path: ^1.8.3
   logging: ^1.2.0
-  sqlite_async: ^0.8.0
+  sqlite_async:
+    git:
+      url: https://github.com/powersync-ja/sqlite_async.dart.git
+      ref: feat/navigator-locks
+      path: packages/sqlite_async
   universal_io: ^2.2.2
 
 dev_dependencies:

demos/supabase-simple-chat/pubspec.lock

@@ -382,7 +382,7 @@ packages:
       path: "../../packages/powersync"
       relative: true
     source: path
-    version: "1.3.0-alpha.7"
+    version: "1.3.0-alpha.8"
   realtime_client:
     dependency: transitive
     description:
@@ -535,10 +535,11 @@ packages:
   sqlite_async:
     dependency: transitive
     description:
-      name: sqlite_async
-      sha256: "7c5a9bec86b6f5b7511b9ba30974fa7ea470aee2dc0d5b7021f6321a439a8d63"
-      url: "https://pub.dev"
-    source: hosted
+      path: "packages/sqlite_async"
+      ref: "feat/navigator-locks"
+      resolved-ref: "588d92f3dc98ab2629d2b420ad607639b835214b"
+      url: "https://github.com/powersync-ja/sqlite_async.dart.git"
+    source: git
     version: "0.8.0"
   stack_trace:
     dependency: transitive

demos/supabase-todolist/pubspec.lock

@@ -454,14 +454,14 @@ packages:
       path: "../../packages/powersync"
       relative: true
     source: path
-    version: "1.3.0-alpha.7"
+    version: "1.3.0-alpha.8"
   powersync_attachments_helper:
     dependency: "direct main"
     description:
       path: "../../packages/powersync_attachments_helper"
       relative: true
     source: path
-    version: "0.3.0-alpha.2"
+    version: "0.3.0-alpha.3"
   realtime_client:
     dependency: transitive
     description:
@@ -590,10 +590,11 @@ packages:
   sqlite_async:
     dependency: "direct main"
     description:
-      name: sqlite_async
-      sha256: "7c5a9bec86b6f5b7511b9ba30974fa7ea470aee2dc0d5b7021f6321a439a8d63"
-      url: "https://pub.dev"
-    source: hosted
+      path: "packages/sqlite_async"
+      ref: "feat/navigator-locks"
+      resolved-ref: "588d92f3dc98ab2629d2b420ad607639b835214b"
+      url: "https://github.com/powersync-ja/sqlite_async.dart.git"
+    source: git
     version: "0.8.0"
   stack_trace:
     dependency: transitive

demos/supabase-todolist/pubspec.yaml

@@ -20,7 +20,11 @@ dependencies:
   camera: ^0.10.5+7
   image: ^4.1.3
   universal_io: ^2.2.2
-  sqlite_async: ^0.8.0
+  sqlite_async:
+    git:
+      url: https://github.com/powersync-ja/sqlite_async.dart.git
+      ref: feat/navigator-locks
+      path: packages/sqlite_async
 
 dev_dependencies:
   flutter_test:

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

@@ -134,17 +134,19 @@ class PowerSyncDatabaseImpl
 
     await isInitialized;
 
-    // TODO multitab support
+    // TODO better multitab support
     final storage = BucketStorage(database);
-
     final sync = StreamingSyncImplementation(
         adapter: storage,
         credentialsCallback: connector.getCredentialsCached,
         invalidCredentialsCallback: connector.fetchCredentials,
         uploadCrud: () => connector.uploadData(this),
         updateStream: updates,
         retryDelay: Duration(seconds: 3),
-        client: FetchClient(mode: RequestMode.cors));
+        client: FetchClient(mode: RequestMode.cors),
+        // Only allows 1 sync implementation to run at a time per database
+        // This should be global (across tabs) when using Navigator locks.
+        identifier: database.openFactory.path);
     sync.statusStream.listen((event) {
       setStatus(event);
     });

packages/powersync/lib/src/streaming_sync.dart

@@ -4,6 +4,7 @@ import 'package:http/http.dart' as http;
 import 'package:powersync/src/abort_controller.dart';
 import 'package:powersync/src/exceptions.dart';
 import 'package:powersync/src/log_internal.dart';
+import 'package:sqlite_async/mutex.dart';
 
 import 'bucket_storage.dart';
 import 'connector.dart';
@@ -37,14 +38,22 @@ class StreamingSyncImplementation {
 
   SyncStatus lastStatus = const SyncStatus();
 
+  final Mutex syncMutex, crudMutex;
+
   StreamingSyncImplementation(
       {required this.adapter,
       required this.credentialsCallback,
       this.invalidCredentialsCallback,
       required this.uploadCrud,
       required this.updateStream,
       required this.retryDelay,
-      required http.Client client}) {
+      required http.Client client,
+
+      /// A unique identifier for this streaming sync implementation
+      /// A good value is typically the DB file path which it will mutate when syncing.
+      String? identifier = "unknown"})
+      : syncMutex = Mutex(identifier: "sync-${identifier}"),
+        crudMutex = Mutex(identifier: "crud-${identifier}") {
     _client = client;
     statusStream = _statusStreamController.stream;
   }
@@ -65,8 +74,10 @@ class StreamingSyncImplementation {
           await invalidCredentialsCallback!();
           invalidCredentials = false;
         }
-        await streamingSyncIteration(abortController: abortController);
-        // Continue immediately
+        // Protect sync iterations with exclusivity (if a valid Mutex is provided)
+        await syncMutex.lock(
+            () => streamingSyncIteration(abortController: abortController),
+            timeout: retryDelay);
       } catch (e, stacktrace) {
         final message = _syncErrorMessage(e);
         isolateLogger.warning('Sync error: $message', e, stacktrace);
@@ -110,21 +121,23 @@ class StreamingSyncImplementation {
   }
 
   Future<bool> uploadCrudBatch() async {
-    if ((await adapter.hasCrud())) {
-      _updateStatus(uploading: true);
-      await uploadCrud();
-      return false;
-    } else {
-      // This isolate is the only one triggering
-      final updated = await adapter.updateLocalTarget(() async {
-        return getWriteCheckpoint();
-      });
-      if (updated) {
-        _localPingController.add(null);
-      }
+    return crudMutex.lock(() async {
+      if ((await adapter.hasCrud())) {
+        _updateStatus(uploading: true);
+        await uploadCrud();
+        return false;
+      } else {
+        // This isolate is the only one triggering
+        final updated = await adapter.updateLocalTarget(() async {
+          return getWriteCheckpoint();
+        });
+        if (updated) {
+          _localPingController.add(null);
+        }
 
-      return true;
-    }
+        return true;
+      }
+    }, timeout: retryDelay);
   }
 
   Future<String> getWriteCheckpoint() async {

packages/powersync/pubspec.yaml

@@ -10,7 +10,12 @@ dependencies:
   flutter:
     sdk: flutter
 
-  sqlite_async: ^0.8.0
+  sqlite_async:
+    git:
+      url: https://github.com/powersync-ja/sqlite_async.dart.git
+      ref: feat/navigator-locks
+      path: packages/sqlite_async
+
   universal_io: ^2.0.0
   sqlite3_flutter_libs: ^0.5.15
   meta: ^1.0.0