Merge pull request #226 from powersync-ja/fix-web-locks

Fix web lock issues

.changeset/real-bottles-mate.md

@@ -0,0 +1,5 @@
+---
+'@powersync/web': patch
+---
+
+Fix read statements not using the transaction lock

packages/kysely-driver/tests/setup/db.ts

@@ -18,5 +18,3 @@ export const getPowerSyncDb = () => {
 
   return database;
 };
-
-export const getKyselyDb = wrapPowerSyncWithKysely<Database>(getPowerSyncDb());

packages/kysely-driver/tests/sqlite/sqlite-connection.test.ts

@@ -21,7 +21,6 @@ describe('PowerSyncConnection', () => {
 
     it('should execute a select query using getAll from the table', async () => {
       await powerSyncDb.execute('INSERT INTO users (id, name) VALUES(uuid(), ?)', ['John']);
-
       const getAllSpy = vi.spyOn(powerSyncDb, 'getAll');
 
       const compiledQuery: CompiledQuery = {
@@ -35,7 +34,7 @@ describe('PowerSyncConnection', () => {
 
       expect(rows.length).toEqual(1);
       expect(rows[0].name).toEqual('John');
-      expect(getAllSpy).toHaveBeenCalledTimes(1);
+      expect(getAllSpy).toHaveBeenCalledWith('SELECT * From users', []);
     });
 
     it('should insert to the table', async () => {

packages/web/src/db/adapters/wa-sqlite/WASQLiteDBAdapter.ts

@@ -46,7 +46,9 @@ export class WASQLiteDBAdapter extends BaseObserver<DBAdapterListener> implement
     this.dbGetHelpers = null;
     this.methods = null;
     this.initialized = this.init();
-    this.dbGetHelpers = this.generateDBHelpers({ execute: this._execute.bind(this) });
+    this.dbGetHelpers = this.generateDBHelpers({
+      execute: (query, params) => this.acquireLock(() => this._execute(query, params))
+    });
   }
 
   get name() {

packages/web/src/shared/open-db.ts

@@ -2,6 +2,7 @@ import * as SQLite from '@journeyapps/wa-sqlite';
 import '@journeyapps/wa-sqlite';
 import * as Comlink from 'comlink';
 import type { DBFunctionsInterface, OnTableChangeCallback, WASQLExecuteResult } from './types';
+import { Mutex } from 'async-mutex';
 
 let nextId = 1;
 
@@ -18,6 +19,7 @@ export async function _openDB(
   sqlite3.vfs_register(vfs, true);
 
   const db = await sqlite3.open_v2(dbFileName);
+  const statementMutex = new Mutex();
 
   /**
    * Listeners are exclusive to the DB connection.
@@ -40,10 +42,10 @@ export async function _openDB(
 
   /**
    * This requests a lock for executing statements.
-   * Should only be used interanlly.
+   * Should only be used internally.
    */
-  const _acquireExecuteLock = (callback: () => Promise<any>): Promise<any> => {
-    return navigator.locks.request(`db-execute-${dbFileName}`, callback);
+  const _acquireExecuteLock = <T>(callback: () => Promise<T>): Promise<T> => {
+    return statementMutex.runExclusive(callback);
   };
 
   /**
@@ -115,7 +117,7 @@ export async function _openDB(
    * This executes SQL statements in a batch.
    */
   const executeBatch = async (sql: string, bindings?: any[][]): Promise<WASQLExecuteResult> => {
-    return _acquireExecuteLock(async () => {
+    return _acquireExecuteLock(async (): Promise<WASQLExecuteResult> => {
       let affectedRows = 0;
 
       const str = sqlite3.str_new(db, sql);
@@ -127,7 +129,8 @@ export async function _openDB(
         const prepared = await sqlite3.prepare_v2(db, query);
         if (prepared === null) {
           return {
-            rowsAffected: 0
+            rowsAffected: 0,
+            rows: { _array: [], length: 0 }
           };
         }
         const wrappedBindings = bindings ? bindings : [];
@@ -158,13 +161,15 @@ export async function _openDB(
       } catch (err) {
         await executeSingleStatement('ROLLBACK');
         return {
-          rowsAffected: 0
+          rowsAffected: 0,
+          rows: { _array: [], length: 0 }
         };
       } finally {
         sqlite3.str_finish(str);
       }
       const result = {
-        rowsAffected: affectedRows
+        rowsAffected: affectedRows,
+        rows: { _array: [], length: 0 }
       };
 
       return result;

packages/web/tests/crud.test.ts

@@ -3,6 +3,7 @@ import { AbstractPowerSyncDatabase, Column, ColumnType, CrudEntry, Schema, Table
 import { PowerSyncDatabase } from '@powersync/web';
 import { v4 as uuid } from 'uuid';
 import { generateTestDb } from './utils/testDb';
+import pDefer from 'p-defer';
 
 const testId = '2290de4f-0488-4e50-abed-f8e8eb1d0b42';
 
@@ -289,4 +290,25 @@ describe('CRUD Tests', () => {
     await tx2.complete();
     expect(await powersync.getNextCrudTransaction()).equals(null);
   });
+
+  it('Transaction exclusivity', async () => {
+    const outside = pDefer();
+    const inTx = pDefer();
+
+    const txPromise = powersync.writeTransaction(async (tx) => {
+      await tx.execute('INSERT INTO assets(id, description) VALUES(?, ?)', [testId, 'test1']);
+      inTx.resolve();
+      await outside.promise;
+      await tx.rollback();
+    });
+
+    await inTx.promise;
+
+    const r = powersync.getOptional<any>('SELECT * FROM assets WHERE id = ?', [testId]);
+    await new Promise((resolve) => setTimeout(resolve, 10));
+    outside.resolve();
+
+    await txPromise;
+    expect(await r).toEqual(null);
+  });
 });