Merge pull request #23 from powersync-ja/resume-transactions

Support resuming transactions / v0.6.0

Author: rkistner

.github/workflows/test.yaml

@@ -30,15 +30,21 @@ jobs:
     strategy:
       matrix:
         include:
+          - sqlite_version: "3440200"
+            sqlite_url: "https://www.sqlite.org/2023/sqlite-autoconf-3440200.tar.gz"
+            dart_sdk: 3.2.4
+          - sqlite_version: "3430200"
+            sqlite_url: "https://www.sqlite.org/2023/sqlite-autoconf-3430200.tar.gz"
+            dart_sdk: 3.2.4
           - sqlite_version: "3420000"
             sqlite_url: "https://www.sqlite.org/2023/sqlite-autoconf-3420000.tar.gz"
-            dart_sdk: 3.0.6
+            dart_sdk: 3.2.4
           - sqlite_version: "3410100"
             sqlite_url: "https://www.sqlite.org/2023/sqlite-autoconf-3410100.tar.gz"
-            dart_sdk: 2.19.1
+            dart_sdk: 3.2.4
           - sqlite_version: "3380000"
             sqlite_url: "https://www.sqlite.org/2022/sqlite-autoconf-3380000.tar.gz"
-            dart_sdk: 2.19.1
+            dart_sdk: 3.2.0
     steps:
       - uses: actions/checkout@v3
       - uses: dart-lang/setup-dart@v1

CHANGELOG.md

@@ -1,3 +1,9 @@
+## 0.6.0
+
+- Allow catching errors and continuing the transaction. This is technically a breaking change, although it should not be an issue in most cases.
+- Add `tx.closed` and `db/tx.getAutoCommit()` to check whether transactions are active.
+- Requires sqlite3 ^2.3.0 and Dart ^3.2.0.
+
 ## 0.5.2
 
 - Fix releasing of locks when closing `SharedMutex``.

lib/src/connection_pool.dart

@@ -49,6 +49,15 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
       : _writeConnection = writeConnection,
         _upstreamPort = upstreamPort;
 
+  /// Returns true if the _write_ connection is currently in autocommit mode.
+  @override
+  Future<bool> getAutoCommit() async {
+    if (_writeConnection == null) {
+      throw AssertionError('Closed');
+    }
+    return await _writeConnection!.getAutoCommit();
+  }
+
   @override
   Future<T> readLock<T>(Future<T> Function(SqliteReadContext tx) callback,
       {Duration? lockTimeout, String? debugContext}) async {

lib/src/port_channel.dart

@@ -286,7 +286,7 @@ class _PortChannelResult<T> {
       return _result as T;
     } else {
       if (_error != null && stackTrace != null) {
-        Error.throwWithStackTrace(_error!, stackTrace!);
+        Error.throwWithStackTrace(_error, stackTrace!);
       } else {
         throw _error!;
       }

lib/src/sqlite_connection.dart

@@ -13,6 +13,18 @@ abstract class SqliteReadContext {
   Future<sqlite.Row?> getOptional(String sql,
       [List<Object?> parameters = const []]);
 
+  /// For transactions, returns true if the lock is held (even if it has been
+  /// rolled back).
+  ///
+  /// For database connections, returns true if the connection hasn't been closed
+  /// yet.
+  bool get closed;
+
+  /// Returns true if auto-commit is enabled. This means the database is not
+  /// currently in a transaction. This may be true even if a transaction lock
+  /// is still held, when the transaction has been committed or rolled back.
+  Future<bool> getAutoCommit();
+
   /// Run a function within a database isolate, with direct synchronous access
   /// to the underlying database.
   ///
@@ -105,5 +117,6 @@ abstract class SqliteConnection extends SqliteWriteContext {
   Future<void> close();
 
   /// Returns true if the connection is closed
+  @override
   bool get closed;
 }

lib/src/sqlite_connection_impl.dart

@@ -49,6 +49,21 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
     return _isolateClient.closed;
   }
 
+  @override
+  Future<bool> getAutoCommit() async {
+    if (closed) {
+      throw AssertionError('Closed');
+    }
+    // We use a _TransactionContext without a lock here.
+    // It is safe to call this in the middle of another transaction.
+    final ctx = _TransactionContext(_isolateClient);
+    try {
+      return await ctx.getAutoCommit();
+    } finally {
+      await ctx.close();
+    }
+  }
+
   Future<void> _open(SqliteOpenFactory openFactory,
       {required bool primary,
       required SerializedPortClient upstreamPort}) async {
@@ -145,6 +160,11 @@ class _TransactionContext implements SqliteWriteContext {
 
   _TransactionContext(this._sendPort);
 
+  @override
+  bool get closed {
+    return _closed;
+  }
+
   @override
   Future<sqlite.ResultSet> execute(String sql,
       [List<Object?> parameters = const []]) async {
@@ -176,6 +196,15 @@ class _TransactionContext implements SqliteWriteContext {
     }
   }
 
+  @override
+  Future<bool> getAutoCommit() async {
+    return await computeWithDatabase(
+      (db) async {
+        return db.autocommit;
+      },
+    );
+  }
+
   @override
   Future<T> computeWithDatabase<T>(
       Future<T> Function(sqlite.Database db) compute) async {
@@ -269,10 +298,8 @@ Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
   server.open((data) async {
     if (data is _SqliteIsolateClose) {
       if (txId != null) {
-        try {
+        if (!db.autocommit) {
           db.execute('ROLLBACK');
-        } catch (e) {
-          // Ignore
         }
         txId = null;
         txError = null;
@@ -290,7 +317,8 @@ Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
         txId = data.ctxId;
       } else if (txId != null && txId != data.ctxId) {
         // Locks should prevent this from happening
-        throw AssertionError('Mixed transactions: $txId and ${data.ctxId}');
+        throw sqlite.SqliteException(
+            0, 'Mixed transactions: $txId and ${data.ctxId}');
       } else if (data.sql == 'ROLLBACK') {
         // This is the only valid way to clear an error
         txError = null;
@@ -307,7 +335,13 @@ Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
         return result;
       } catch (err) {
         if (txId != null) {
-          txError = err;
+          if (db.autocommit) {
+            // Transaction rolled back
+            txError = sqlite.SqliteException(0,
+                'Transaction rolled back by earlier statement: ${err.toString()}');
+          } else {
+            // Recoverable error
+          }
         }
         rethrow;
       }

lib/src/sqlite_database.dart

@@ -109,6 +109,13 @@ class SqliteDatabase with SqliteQueries implements SqliteConnection {
     return _pool.closed;
   }
 
+  /// Returns true if the _write_ connection is in auto-commit mode
+  /// (no active transaction).
+  @override
+  Future<bool> getAutoCommit() {
+    return _pool.getAutoCommit();
+  }
+
   void _listenForEvents() {
     UpdateNotification? updates;
 

pubspec.yaml

@@ -1,18 +1,18 @@
 name: sqlite_async
 description: High-performance asynchronous interface for SQLite on Dart and Flutter.
-version: 0.5.2
-repository: https://github.com/journeyapps/sqlite_async.dart
+version: 0.6.0
+repository: https://github.com/powersync-ja/sqlite_async.dart
 environment:
-  sdk: '>=2.19.1 <4.0.0'
+  sdk: '>=3.2.0 <4.0.0'
 
 dependencies:
-  sqlite3: '>=1.10.1 <3.0.0'
+  sqlite3: '^2.3.0'
   async: ^2.10.0
   collection: ^1.17.0
 
 dev_dependencies:
-  lints: ^2.0.0
+  lints: ^3.0.0
   test: ^1.21.0
-  test_api: ^0.4.18
+  test_api: ^0.7.0
   glob: ^2.1.1
   benchmarking: ^0.6.1

test/basic_test.dart

@@ -222,18 +222,37 @@ void main() {
       await createTables(db);
 
       var tp = db.writeTransaction((tx) async {
-        tx.execute('INSERT INTO test_data(description) VALUES(?)', ['test1']);
-        tx.execute('INSERT INTO test_data(description) VALUES(?)', ['test2']);
-        ignore(tx.execute('INSERT INTO test_data(description) VALUES(json(?))',
-            ['test3'])); // Errors
-        // Will not be executed because of the above error
+        await tx.execute(
+            'INSERT OR ROLLBACK INTO test_data(id, description) VALUES(?, ?)',
+            [1, 'test1']);
+        await tx.execute(
+            'INSERT OR ROLLBACK INTO test_data(id, description) VALUES(?, ?)',
+            [2, 'test2']);
+        expect(await tx.getAutoCommit(), equals(false));
+        try {
+          await tx.execute(
+              'INSERT OR ROLLBACK INTO test_data(id, description) VALUES(?, ?)',
+              [2, 'test3']);
+        } catch (e) {
+          // Ignore
+        }
+        expect(await tx.getAutoCommit(), equals(true));
+        expect(tx.closed, equals(false));
+
+        // Will not be executed because of the above rollback
         ignore(tx.execute(
-            'INSERT INTO test_data(description) VALUES(?) RETURNING *',
-            ['test4']));
+            'INSERT OR ROLLBACK INTO test_data(id, description) VALUES(?, ?)',
+            [4, 'test4']));
       });
 
       // The error propagates up to the transaction
-      await expectLater(tp, throwsA((e) => e is sqlite.SqliteException));
+      await expectLater(
+          tp,
+          throwsA((e) =>
+              e is sqlite.SqliteException &&
+              e.message
+                  .contains('Transaction rolled back by earlier statement') &&
+              e.message.contains('UNIQUE constraint failed')));
 
       expect(await db.get('SELECT count() count FROM test_data'),
           equals({'count': 0}));
@@ -321,6 +340,34 @@ void main() {
       });
       expect(computed, equals(5));
     });
+
+    test('should allow resuming transaction after errors', () async {
+      final db = await setupDatabase(path: path);
+      await createTables(db);
+      SqliteWriteContext? savedTx;
+      await db.writeTransaction((tx) async {
+        savedTx = tx;
+        var caught = false;
+        try {
+          // This error does not rollback the transaction
+          await tx.execute('NOT A VALID STATEMENT');
+        } catch (e) {
+          // Ignore
+          caught = true;
+        }
+        expect(caught, equals(true));
+
+        expect(await tx.getAutoCommit(), equals(false));
+        expect(tx.closed, equals(false));
+
+        final rs = await tx.execute(
+            'INSERT INTO test_data(description) VALUES(?) RETURNING description',
+            ['Test Data']);
+        expect(rs.rows[0], equals(['Test Data']));
+      });
+      expect(await savedTx!.getAutoCommit(), equals(true));
+      expect(savedTx!.closed, equals(true));
+    });
   });
 }