Fix rollbacks after failed commits

.changeset/fuzzy-frogs-draw.md

@@ -0,0 +1,6 @@
+---
+'@powersync/kysely-driver': patch
+'@powersync/common': patch
+---
+
+Fix recovery from failing commits.

packages/common/src/db/DBAdapter.ts

@@ -258,17 +258,19 @@ class TransactionImplementation extends LockContext {
   static async runWith<T>(ctx: LockContext, fn: (tx: Transaction) => Promise<T>): Promise<T> {
     let tx = new TransactionImplementation(ctx);
 
-    try {
-      await ctx.execute('BEGIN IMMEDIATE');
+    await ctx.execute('BEGIN IMMEDIATE');
 
+    try {
       const result = await fn(tx);
       await tx.commit();
       return result;
     } catch (ex) {
+      tx.finalized = true;
+
       try {
-        await tx.rollback();
+        await ctx.execute('ROLLBACK');
       } catch (ex2) {
-        // In rare cases, a rollback may fail.
+        // In rare cases, a rollback may fail (e.g. if user code invoked commit() and then threw afterwards).
         // Safe to ignore.
       }
       throw ex;

packages/kysely-driver/src/sqlite/sqlite-connection.ts

@@ -1,4 +1,4 @@
-import { type AbstractPowerSyncDatabase, type Transaction } from '@powersync/common';
+import type { CommonPowerSyncDatabase, Transaction } from '@powersync/common';
 import { CompiledQuery, DatabaseConnection, QueryResult } from 'kysely';
 
 /**
@@ -11,14 +11,14 @@ import { CompiledQuery, DatabaseConnection, QueryResult } from 'kysely';
  * This allows us to bypass write locks when doing pure select queries outside a transaction.
  */
 export class PowerSyncConnection implements DatabaseConnection {
-  readonly #db: AbstractPowerSyncDatabase;
-  #completeTransaction: (() => void) | null;
+  readonly #db: CommonPowerSyncDatabase;
+  #completeOuterTransaction: (() => void) | null = null;
+  #failOuterTransaction: ((error: unknown) => void) | null = null;
   #tx: Transaction | null;
 
-  constructor(db: AbstractPowerSyncDatabase) {
+  constructor(db: CommonPowerSyncDatabase) {
     this.#db = db;
     this.#tx = null;
-    this.#completeTransaction = null;
   }
 
   async executeQuery<O>(compiledQuery: CompiledQuery): Promise<QueryResult<O>> {
@@ -69,13 +69,11 @@ export class PowerSyncConnection implements DatabaseConnection {
           // Set the current active transaction
           this.#tx = tx;
 
-          /**
-           * Wait for this transaction to be completed
-           * Rejecting would cause any uncommitted changes to be
-           * rolled back.
-           */
-          const transactionCompleted = new Promise<void>((resolve) => {
-            this.#completeTransaction = resolve;
+          // Wait for this transaction to be completed.
+          // Rejecting causes any uncommitted changes to be rolled back.
+          const transactionCompleted = new Promise<void>((resolve, reject) => {
+            this.#completeOuterTransaction = resolve;
+            this.#failOuterTransaction = reject;
           });
 
           // Allow this transaction to be used externally
@@ -94,23 +92,30 @@ export class PowerSyncConnection implements DatabaseConnection {
 
     try {
       await this.#tx.commit();
-    } finally {
-      // The SQLite transaction is over whether or not COMMIT succeeded. Holding the write lock
-      // any longer would block every other write.
-      this.releaseTransaction();
+    } catch (e) {
+      // A failed COMMIT may leave the transaction open. Rejecting makes writeTransaction roll it back
+      // before releasing the write lock.
+      this.#completeError(e);
+      throw e;
     }
+    this.#complete();
   }
 
   async rollbackTransaction(): Promise<void> {
     if (!this.#tx) {
-      throw new Error('Transaction is not defined');
+      // Kysely calls this after a failed commitTransaction, which already handed the rollback to
+      // writeTransaction. Throwing here would hide the original COMMIT error.
+      return;
     }
 
     try {
       await this.#tx.rollback();
-    } finally {
-      this.releaseTransaction();
+    } catch (e) {
+      // Let writeTransaction retry the rollback so that the changes are never committed.
+      this.#completeError(e);
+      throw e;
     }
+    this.#complete();
   }
 
   async releaseConnection(): Promise<void> {
@@ -119,17 +124,27 @@ export class PowerSyncConnection implements DatabaseConnection {
     if (this.#tx) {
       try {
         await this.#tx.rollback();
-      } catch {
-        // The transaction may already be over, nothing left to roll back.
-      } finally {
-        this.releaseTransaction();
+        this.#complete();
+      } catch (e) {
+        // Let writeTransaction retry the rollback. If the transaction is already over, that's a no-op.
+        this.#completeError(e);
       }
     }
   }
 
-  private releaseTransaction() {
-    this.#completeTransaction?.();
-    this.#completeTransaction = null;
+  #complete() {
+    this.#completeOuterTransaction?.();
+    this.#markOuterTransactionReleased();
+  }
+
+  #completeError(cause: unknown) {
+    this.#failOuterTransaction?.(cause);
+    this.#markOuterTransactionReleased();
+  }
+
+  #markOuterTransactionReleased() {
+    this.#completeOuterTransaction = null;
+    this.#failOuterTransaction = null;
     this.#tx = null;
   }
 }

packages/kysely-driver/tests/sqlite/transaction-release.test.ts

@@ -66,4 +66,26 @@ describe('transaction lock release', () => {
 
     await expect(writeWithTimeout(powerSyncDb)).resolves.toBeDefined();
   });
+
+  it('should roll back when COMMIT fails with the transaction still open', async () => {
+    // A deferred foreign key violation makes COMMIT fail while leaving the transaction open.
+    await powerSyncDb.writeLock((tx) => tx.execute('PRAGMA foreign_keys = ON'));
+    await powerSyncDb.execute('CREATE TABLE parent (id TEXT PRIMARY KEY)');
+    await powerSyncDb.execute('CREATE TABLE child (id TEXT REFERENCES parent(id) DEFERRABLE INITIALLY DEFERRED)');
+
+    await expect(
+      db.transaction().execute(async (trx) => {
+        await sql`
+          INSERT INTO
+            child
+          VALUES
+            ('missing')
+        `.execute(trx);
+      })
+    ).rejects.toThrow(/FOREIGN KEY/);
+
+    // The transaction should have been rolled back, so the row should not exist.
+    expect(await powerSyncDb.writeLock((tx) => tx.getAll('SELECT * FROM child'))).toHaveLength(0);
+    await expect(writeWithTimeout(powerSyncDb)).resolves.toBeDefined();
+  });
 });

packages/node/tests/PowerSyncDatabase.test.ts

@@ -338,6 +338,23 @@ databaseTest('soft clear', async ({ database }) => {
   expect(await database.getAll('SELECT name FROM ps_buckets')).toStrictEqual([]);
 });
 
+databaseTest('rolls back failing transactions', async ({ database }) => {
+  // Regression test for https://github.com/powersync-ja/powersync-js/issues/1176.
+  await database.writeLock((tx) => tx.execute('PRAGMA foreign_keys = ON'));
+  await database.execute('CREATE TABLE parent (id TEXT PRIMARY KEY)');
+  await database.execute('CREATE TABLE child (id TEXT REFERENCES parent(id) DEFERRABLE INITIALLY DEFERRED)');
+
+  await expect(() =>
+    database.writeTransaction((tx) => tx.execute("INSERT INTO child VALUES ('missing')"))
+  ).rejects.toThrow();
+
+  // Transaction should have been rolled back, so the row should not exist.
+  expect(await database.writeLock((tx) => tx.getAll('SELECT * FROM child'))).toHaveLength(0);
+
+  // We should still be able to issue transactions.
+  await database.writeTransaction((tx) => tx.execute('SELECT 1'));
+});
+
 describe('onChangeWithAsyncGenerator', () => {
   databaseTest('receives change events', async ({ database }) => {
     const changes: WatchOnChangeEvent[] = [];