Prepare dev update (#136)

Author: simolus3

packages/drift_sqlite_async/CHANGELOG.md

@@ -1,4 +1,4 @@
-## 0.3.0-wip
+## 0.3.0-wip.0
 
 - Support versions 3.x of the `sqlite3` package.
 

packages/drift_sqlite_async/pubspec.yaml

@@ -1,5 +1,5 @@
 name: drift_sqlite_async
-version: 0.3.0-wip
+version: 0.3.0-wip.0
 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.

packages/sqlite_async/CHANGELOG.md

@@ -1,4 +1,8 @@
-## 0.14.0-wip
+## 0.14.0-wip.0
+
+__Note__: This version of `sqlite_async` is still in development and there might be additional
+API changes between this release and the final `0.14.0` version. This release is mostly meant for
+internal testing.
 
 - Support versions 3.x of the `sqlite3` package and 0.6.0 of `sqlite3_web`.
 - Remove the `sqlite3_open.dart` library, SQLite libraries are no longer loaded through Dart.

packages/sqlite_async/lib/src/common/connection/sync_sqlite_connection.dart

@@ -8,6 +8,7 @@ import 'package:sqlite_async/src/update_notification.dart';
 import 'package:sqlite_async/src/utils/profiler.dart';
 
 import '../../impl/context.dart';
+import '../../utils/shared_utils.dart';
 
 /// A simple "synchronous" connection which provides the async SqliteConnection
 /// implementation using a synchronous SQLite connection
@@ -35,6 +36,16 @@ final class SyncSqliteConnection extends SqliteConnection {
     );
   }
 
+  @override
+  Future<T> readTransaction<T>(
+      Future<T> Function(SqliteReadContext tx) callback,
+      {Duration? lockTimeout}) {
+    return readLock((ctx) async {
+      return await internalReadTransaction(ctx, callback,
+          isDedicatedReadConnection: false);
+    }, lockTimeout: lockTimeout, debugContext: 'readTransaction()');
+  }
+
   @override
   Future<T> abortableReadLock<T>(
       Future<T> Function(SqliteReadContext tx) callback,

packages/sqlite_async/lib/src/common/sqlite_database.dart

@@ -7,6 +7,7 @@ import 'package:sqlite_async/src/sqlite_options.dart';
 import 'package:sqlite_async/src/sqlite_connection.dart';
 
 import '../impl/platform.dart' as platform;
+import '../utils/shared_utils.dart';
 
 /// A SQLite database instance.
 ///
@@ -96,4 +97,14 @@ abstract base class SqliteDatabase extends SqliteConnection {
 @internal
 abstract base class SqliteDatabaseImpl extends SqliteDatabase {
   SqliteDatabaseImpl() : super._();
+
+  @override
+  Future<T> readTransaction<T>(
+      Future<T> Function(SqliteReadContext tx) callback,
+      {Duration? lockTimeout}) {
+    return readLock((ctx) async {
+      return await internalReadTransaction(ctx, callback,
+          isDedicatedReadConnection: maxReaders > 0);
+    }, lockTimeout: lockTimeout, debugContext: 'readTransaction()');
+  }
 }

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

@@ -30,7 +30,16 @@ import 'worker.dart';
 final class NativeSqliteDatabaseImpl extends SqliteDatabaseImpl {
   @override
   final NativeSqliteOpenFactory openFactory;
-  late final Future<SqliteConnectionPool> _pool = _openNativePool(openFactory);
+  late final Future<SqliteConnectionPool> _pool =
+      _openNativePool(openFactory).then((pool) {
+    // Pipe updates into a stream controller as soon as the pool is opened to
+    // avoid missing updates if the stream is only listened to later.
+    _updates.addStream(pool.updatedTables.map((e) {
+      return UpdateNotification(e.toSet());
+    }));
+
+    return pool;
+  });
   bool _isClosed = false;
   final _lockGuard = Object();
 
@@ -43,10 +52,11 @@ final class NativeSqliteDatabaseImpl extends SqliteDatabaseImpl {
 
   final Queue<IsolateWorker> _workers;
 
+  final StreamController<UpdateNotification> _updates =
+      StreamController.broadcast(sync: true);
+
   @override
-  late final Stream<UpdateNotification> updates = Stream.fromFuture(_pool)
-      .asyncExpand((pool) => pool.updatedTables
-          .map((changedTables) => UpdateNotification(changedTables.toSet())));
+  Stream<UpdateNotification> get updates => _updates.stream;
 
   NativeSqliteDatabaseImpl(this.openFactory)
       :
@@ -84,58 +94,6 @@ final class NativeSqliteDatabaseImpl extends SqliteDatabaseImpl {
     }
   }
 
-  /// Open a read-only transaction.
-  ///
-  /// Up to [maxReaders] read transactions can run concurrently.
-  /// After that, read transactions are queued.
-  ///
-  /// Read transactions can run concurrently to a write transaction.
-  ///
-  /// Changes from any write transaction are not visible to read transactions
-  /// started before it.
-  @override
-  Future<T> readTransaction<T>(
-      Future<T> Function(SqliteReadContext tx) callback,
-      {Duration? lockTimeout}) async {
-    return _useConnection(
-      writer: false,
-      abortTrigger: lockTimeout?.asTimeout,
-      debugContext: 'readTransaction',
-      (context) {
-        return _transactionInLease(context, callback);
-      },
-    );
-  }
-
-  /// Open a read-write transaction.
-  ///
-  /// Only a single write transaction can run at a time - any concurrent
-  /// transactions are queued.
-  ///
-  /// The write transaction is automatically committed when the callback finishes,
-  /// or rolled back on any error.
-  @override
-  Future<T> writeTransaction<T>(
-      Future<T> Function(SqliteWriteContext tx) callback,
-      {Duration? lockTimeout}) {
-    return _useConnection(
-      writer: true,
-      abortTrigger: lockTimeout?.asTimeout,
-      debugContext: 'writeTransaction',
-      (context) {
-        return _transactionInLease(context, callback);
-      },
-    );
-  }
-
-  Future<T> _transactionInLease<T>(
-    _LeasedContext context,
-    Future<T> Function(SqliteWriteContext tx) callback,
-  ) {
-    final ctx = ScopedWriteContext(context);
-    return ctx.writeTransaction(callback).whenComplete(ctx.invalidate);
-  }
-
   @override
   Future<T> abortableReadLock<T>(
       Future<T> Function(SqliteReadContext tx) callback,

packages/sqlite_async/lib/src/sqlite_connection.dart

@@ -136,11 +136,7 @@ abstract class SqliteConnection implements SqliteWriteContext {
   /// instance will error.
   Future<T> readTransaction<T>(
       Future<T> Function(SqliteReadContext tx) callback,
-      {Duration? lockTimeout}) {
-    return readLock((ctx) async {
-      return await internalReadTransaction(ctx, callback);
-    }, lockTimeout: lockTimeout, debugContext: 'readTransaction()');
-  }
+      {Duration? lockTimeout});
 
   /// Open a read-write transaction.
   ///

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

@@ -5,10 +5,35 @@ import 'package:sqlite3/common.dart';
 
 import '../sqlite_connection.dart';
 
-Future<T> internalReadTransaction<T>(SqliteReadContext ctx,
-    Future<T> Function(SqliteReadContext tx) callback) async {
+Future<T> internalReadTransaction<T>(
+  SqliteReadContext ctx,
+  Future<T> Function(SqliteReadContext tx) callback, {
+  required bool isDedicatedReadConnection,
+}) async {
   try {
-    await ctx.getAll('BEGIN');
+    // We want read transactions to observe the state of the database at the
+    // time they've been opened. By default however, SQLite only starts the
+    // transaction on the first statement (BEGIN just sets a bit to disable
+    // autocommit). This is a problem for snippets like:
+    //
+    // await db.readTransaction((tx) async {
+    //   // point in time 1.
+    //   await longDelay();
+    //   await readFromDb(tx); // should read state from point in time 1!
+    // });
+    //
+    // With a write concurrent to `longDelay()`, the actual transaction would
+    // start too late and observe state from after it was opened in Dart. This
+    // is why we use BEGIN IMMEDIATE instead of BEGIN. However, we only need
+    // this for read connections: If the database "pool" is backed by a single
+    // connection (e.g. on the web), using BEGIN IMMEDIATE would be fairly
+    // expensive for a read. A concurrent write wouldn't be a concern there
+    // because the read context blocks the single connection.
+    // Either way, this is not a consistency issue. Transactions observe a
+    // single consistent snapshot either way, we just want them to observe an
+    // earlier snapshot in some cases.
+
+    await ctx.getAll(isDedicatedReadConnection ? 'BEGIN IMMEDIATE' : 'BEGIN');
     final result = await callback(ctx);
     await ctx.getAll('END TRANSACTION');
     return result;

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

@@ -68,7 +68,7 @@ final class WebDatabase extends SqliteDatabaseImpl
   @override
 
   /// Not supported on web. There is only 1 connection.
-  int get maxReaders => throw UnimplementedError();
+  int get maxReaders => 0;
 
   @override
 

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

@@ -29,7 +29,7 @@ final class AsyncWebDatabaseImpl extends SqliteDatabaseImpl
   late Stream<UpdateNotification> updates;
 
   @override
-  int get maxReaders => openFactory.sqliteOptions.maxReaders;
+  int get maxReaders => 0;
 
   @override
   @protected