Merge remote-tracking branch 'origin/main' into get_all_connections

Author: davidmartos96

CHANGELOG.md

@@ -3,6 +3,63 @@
 All notable changes to this project will be documented in this file.
 See [Conventional Commits](https://conventionalcommits.org) for commit guidelines.
 
+## 2025-08-08
+
+### Changes
+
+---
+
+Packages with breaking changes:
+
+ - There are no breaking changes in this release.
+
+Packages with other changes:
+
+ - [`sqlite_async` - `v0.12.1`](#sqlite_async---v0121)
+ - [`sqlite_async` - `v0.12.0`](#sqlite_async---v0120)
+ - [`drift_sqlite_async` - `v0.2.3+1`](#drift_sqlite_async---v0231)
+
+Packages with dependency updates only:
+
+> Packages listed below depend on other packages in this workspace that have had changes. Their versions have been incremented to bump the minimum dependency versions of the packages they depend upon in this project.
+
+ - `drift_sqlite_async` - `v0.2.3+1`
+
+---
+
+#### `sqlite_async` - `v0.12.1`
+
+- Fix distributing updates from shared worker.
+
+#### `sqlite_async` - `v0.12.0`
+
+ - Avoid large transactions creating a large internal update queue.
+
+
+## 2025-07-29
+
+---
+
+Packages with breaking changes:
+
+ - There are no breaking changes in this release.
+
+Packages with other changes:
+
+ - [`sqlite_async` - `v0.11.8`](#sqlite_async---v0118)
+ - [`drift_sqlite_async` - `v0.2.3`](#drift_sqlite_async---v023)
+
+---
+
+#### `sqlite_async` - `v0.11.8`
+
+- Support nested transactions (emulated with `SAVEPOINT` statements).
+- Fix web compilation issues with version `2.8.0` of `package:sqlite3`.
+
+#### `drift_sqlite_async` - `v0.2.3`
+
+- Support nested transactions.
+
 ## 2025-06-03
 
 ---

packages/drift_sqlite_async/CHANGELOG.md

@@ -1,3 +1,11 @@
+## 0.2.3+1
+
+ - Update a dependency to the latest release.
+
+## 0.2.3
+
+- Support nested transactions.
+
 ## 0.2.2
 
 - Fix write detection when using UPDATE/INSERT/DELETE with RETURNING in raw queries.

packages/drift_sqlite_async/lib/src/executor.dart

@@ -129,13 +129,28 @@ class _SqliteAsyncTransactionDelegate extends SupportedTransactionDelegate {
 
   _SqliteAsyncTransactionDelegate(this._db);
 
+  @override
+  FutureOr<void> Function(QueryDelegate, Future<void> Function(QueryDelegate))?
+      get startNested => _startNested;
+
   @override
   Future<void> startTransaction(Future Function(QueryDelegate p1) run) async {
-    await _db.writeTransaction((context) async {
+    await _startTransaction(_db, run);
+  }
+
+  Future<void> _startTransaction(
+      SqliteWriteContext context, Future Function(QueryDelegate p1) run) async {
+    await context.writeTransaction((context) async {
       final delegate = _SqliteAsyncQueryDelegate(context, null);
       return run(delegate);
     });
   }
+
+  Future<void> _startNested(
+      QueryDelegate outer, Future<void> Function(QueryDelegate) block) async {
+    await _startTransaction(
+        (outer as _SqliteAsyncQueryDelegate)._context, block);
+  }
 }
 
 class _SqliteAsyncVersionDelegate extends DynamicVersionDelegate {

packages/drift_sqlite_async/pubspec.yaml

@@ -1,5 +1,5 @@
 name: drift_sqlite_async
-version: 0.2.2
+version: 0.2.3+1
 homepage: https://github.com/powersync-ja/sqlite_async.dart
 repository: https://github.com/powersync-ja/sqlite_async.dart
 description: Use Drift with a sqlite_async database, allowing both to be used in the same application.
@@ -14,12 +14,12 @@ topics:
 environment:
   sdk: ">=3.0.0 <4.0.0"
 dependencies:
-  drift: ">=2.19.0 <3.0.0"
-  sqlite_async: ^0.11.0
+  drift: ">=2.28.0 <3.0.0"
+  sqlite_async: ^0.12.0
 
 dev_dependencies:
   build_runner: ^2.4.8
-  drift_dev: ">=2.19.0 <3.0.0"
+  drift_dev: ">=2.28.0 <3.0.0"
   glob: ^2.1.2
   lints: ^5.0.0
   sqlite3: ^2.4.0

packages/drift_sqlite_async/test/db_test.dart

@@ -117,5 +117,35 @@ void main() {
       final deleted = await dbu.delete(dbu.todoItems).go();
       expect(deleted, 10);
     });
+
+    test('nested transactions', () async {
+      await dbu
+          .into(dbu.todoItems)
+          .insert(TodoItemsCompanion.insert(description: 'root'));
+
+      await dbu.transaction(() async {
+        await dbu
+            .into(dbu.todoItems)
+            .insert(TodoItemsCompanion.insert(description: 'tx0'));
+
+        await dbu.transaction(() async {
+          await dbu
+              .into(dbu.todoItems)
+              .insert(TodoItemsCompanion.insert(description: 'tx1'));
+
+          await expectLater(() {
+            return dbu.transaction(() async {
+              await dbu
+                  .into(dbu.todoItems)
+                  .insert(TodoItemsCompanion.insert(description: 'tx2'));
+              throw 'rollback';
+            });
+          }, throwsA(anything));
+        });
+      });
+
+      final items = await dbu.todoItems.all().get();
+      expect(items.map((e) => e.description).toSet(), {'root', 'tx0', 'tx1'});
+    });
   });
 }

packages/sqlite_async/CHANGELOG.md

@@ -1,3 +1,16 @@
+## 0.12.1
+
+- Fix distributing updates from shared worker.
+
+## 0.12.0
+
+ - Avoid large transactions creating a large internal update queue.
+
+## 0.11.8
+
+- Support nested transactions (emulated with `SAVEPOINT` statements).
+- Fix web compilation issues with version `2.8.0` of `package:sqlite3`.
+
 ## 0.11.7
 
 - Shared worker: Release locks owned by connected client tab when it closes.

packages/sqlite_async/lib/src/native/database/native_sqlite_connection_impl.dart

@@ -303,40 +303,13 @@ Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
   final server = params.portServer;
   final commandPort = ReceivePort();
 
-  Timer? updateDebouncer;
-  Set<String> updatedTables = {};
+  db.updatedTables.listen((changedTables) {
+    client.fire(UpdateNotification(changedTables));
+  });
+
   int? txId;
   Object? txError;
 
-  void maybeFireUpdates() {
-    // We keep buffering the set of updated tables until we are not
-    // in a transaction. Firing transactions inside a transaction
-    // has multiple issues:
-    // 1. Watched queries would detect changes to the underlying tables,
-    //    but the data would not be visible to queries yet.
-    // 2. It would trigger many more notifications than required.
-    //
-    // This still includes updates for transactions that are rolled back.
-    // We could handle those better at a later stage.
-
-    if (updatedTables.isNotEmpty && db.autocommit) {
-      client.fire(UpdateNotification(updatedTables));
-      updatedTables.clear();
-    }
-    updateDebouncer?.cancel();
-    updateDebouncer = null;
-  }
-
-  db.updates.listen((event) {
-    updatedTables.add(event.tableName);
-
-    // This handles two cases:
-    // 1. Update arrived after _SqliteIsolateClose (not sure if this could happen).
-    // 2. Long-running _SqliteIsolateClosure that should fire updates while running.
-    updateDebouncer ??=
-        Timer(const Duration(milliseconds: 1), maybeFireUpdates);
-  });
-
   ResultSet runStatement(_SqliteIsolateStatement data) {
     if (data.sql == 'BEGIN' || data.sql == 'BEGIN IMMEDIATE') {
       if (txId != null) {
@@ -388,8 +361,6 @@ Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
           throw sqlite.SqliteException(
               0, 'Transaction must be closed within the read or write lock');
         }
-        // We would likely have received updates by this point - fire now.
-        maybeFireUpdates();
         return null;
       case _SqliteIsolateStatement():
         return task.timeSync(
@@ -399,11 +370,7 @@ Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
           parameters: data.args,
         );
       case _SqliteIsolateClosure():
-        try {
-          return await data.cb(db);
-        } finally {
-          maybeFireUpdates();
-        }
+        return await data.cb(db);
       case _SqliteIsolateConnectionClose():
         db.dispose();
         return null;

packages/sqlite_async/lib/src/utils/shared_utils.dart

@@ -1,6 +1,8 @@
 import 'dart:async';
 import 'dart:convert';
 
+import 'package:sqlite3/common.dart';
+
 import '../sqlite_connection.dart';
 
 Future<T> internalReadTransaction<T>(SqliteReadContext ctx,
@@ -75,3 +77,87 @@ Object? mapParameter(Object? parameter) {
 List<Object?> mapParameters(List<Object?> parameters) {
   return [for (var p in parameters) mapParameter(p)];
 }
+
+extension ThrottledUpdates on CommonDatabase {
+  /// An unthrottled stream of updated tables that emits on every commit.
+  ///
+  /// A paused subscription on this stream will buffer changed tables into a
+  /// growing set instead of losing events, so this stream is simple to throttle
+  /// downstream.
+  Stream<Set<String>> get updatedTables {
+    final listeners = <_UpdateListener>[];
+    var uncommitedUpdates = <String>{};
+    var underlyingSubscriptions = <StreamSubscription<void>>[];
+
+    void handleUpdate(SqliteUpdate update) {
+      uncommitedUpdates.add(update.tableName);
+    }
+
+    void afterCommit() {
+      for (final listener in listeners) {
+        listener.notify(uncommitedUpdates);
+      }
+
+      uncommitedUpdates.clear();
+    }
+
+    void afterRollback() {
+      uncommitedUpdates.clear();
+    }
+
+    void addListener(_UpdateListener listener) {
+      listeners.add(listener);
+
+      if (listeners.length == 1) {
+        // First listener, start listening for raw updates on underlying
+        // database.
+        underlyingSubscriptions = [
+          updatesSync.listen(handleUpdate),
+          commits.listen((_) => afterCommit()),
+          commits.listen((_) => afterRollback())
+        ];
+      }
+    }
+
+    void removeListener(_UpdateListener listener) {
+      listeners.remove(listener);
+      if (listeners.isEmpty) {
+        for (final sub in underlyingSubscriptions) {
+          sub.cancel();
+        }
+      }
+    }
+
+    return Stream.multi(
+      (listener) {
+        final wrapped = _UpdateListener(listener);
+        addListener(wrapped);
+
+        listener.onResume = wrapped.addPending;
+        listener.onCancel = () => removeListener(wrapped);
+      },
+      isBroadcast: true,
+    );
+  }
+}
+
+class _UpdateListener {
+  final MultiStreamController<Set<String>> downstream;
+  Set<String> buffered = {};
+
+  _UpdateListener(this.downstream);
+
+  void notify(Set<String> pendingUpdates) {
+    buffered.addAll(pendingUpdates);
+    if (!downstream.isPaused) {
+      addPending();
+    }
+  }
+
+  void addPending() {
+    if (buffered.isNotEmpty) {
+      downstream.add(buffered);
+      buffered = {};
+    }
+  }
+}

packages/sqlite_async/lib/src/web/database.dart

@@ -21,6 +21,9 @@ class WebDatabase
   final Mutex? _mutex;
   final bool profileQueries;
 
+  @override
+  final Stream<UpdateNotification> updates;
+
   /// For persistent databases that aren't backed by a shared worker, we use
   /// web broadcast channels to forward local update events to other tabs.
   final BroadcastUpdates? broadcastUpdates;
@@ -32,6 +35,7 @@ class WebDatabase
     this._database,
     this._mutex, {
     required this.profileQueries,
+    required this.updates,
     this.broadcastUpdates,
   });
 
@@ -113,10 +117,6 @@ class WebDatabase
     }
   }
 
-  @override
-  Stream<UpdateNotification> get updates =>
-      _database.updates.map((event) => UpdateNotification({event.tableName}));
-
   @override
   Future<T> writeTransaction<T>(
       Future<T> Function(SqliteWriteContext tx) callback,

packages/sqlite_async/lib/src/web/protocol.dart

@@ -13,6 +13,8 @@ enum CustomDatabaseMessageKind {
   getAutoCommit,
   executeInTransaction,
   executeBatchInTransaction,
+  updateSubscriptionManagement,
+  notifyUpdates,
 }
 
 extension type CustomDatabaseMessage._raw(JSObject _) implements JSObject {