Add unit tests and autocommit check for tests

Author: stevensJourney

packages/web/src/db/adapters/AsyncDatabaseConnection.ts

@@ -35,6 +35,11 @@ export interface AsyncDatabaseConnection<Config extends ResolvedWebSQLOpenOption
    * @param holdId The hold ID to release.
    */
   releaseHold(holdId: string): Promise<void>;
+  /**
+   * Checks if the database connection is in autocommit mode.
+   * @returns true if in autocommit mode, false if in a transaction
+   */
+  isAutoCommit(): Promise<boolean>;
   execute(sql: string, params?: any[]): Promise<ProxiedQueryResult>;
   executeRaw(sql: string, params?: any[]): Promise<any[][]>;
   executeBatch(sql: string, params?: any[]): Promise<ProxiedQueryResult>;

packages/web/src/db/adapters/WorkerWrappedAsyncDatabaseConnection.ts

@@ -61,11 +61,15 @@ export class WorkerWrappedAsyncDatabaseConnection<Config extends ResolvedWebSQLO
   }
 
   markHold(): Promise<string> {
-    return this.baseConnection.markHold();
+    return this.withRemote(() => this.baseConnection.markHold());
   }
 
   releaseHold(holdId: string): Promise<void> {
-    return this.baseConnection.releaseHold(holdId);
+    return this.withRemote(() => this.baseConnection.releaseHold(holdId));
+  }
+
+  isAutoCommit(): Promise<boolean> {
+    return this.baseConnection.isAutoCommit();
   }
 
   private withRemote<T>(workerPromise: () => Promise<T>): Promise<T> {
@@ -74,6 +78,8 @@ export class WorkerWrappedAsyncDatabaseConnection<Config extends ResolvedWebSQLO
       return new Promise((resolve, reject) => {
         if (controller.signal.aborted) {
           reject(new Error('Called operation on closed remote'));
+          // Don't run the operation if we're going to reject
+          return;
         }
 
         function handleAbort() {

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

@@ -193,6 +193,14 @@ export class WASqliteConnection
     return this._dbP;
   }
 
+  /**
+   * Checks if the database connection is in autocommit mode.
+   * @returns true if in autocommit mode, false if in a transaction
+   */
+  async isAutoCommit(): Promise<boolean> {
+    return this.sqliteAPI.get_autocommit(this.dbP) != 0;
+  }
+
   async markHold(): Promise<string> {
     const previousHoldId = this._holdId;
     this._holdId = `${++this._holdCounter}`;

packages/web/src/worker/db/SharedWASQLiteConnection.ts

@@ -34,6 +34,22 @@ export class SharedWASQLiteConnection implements AsyncDatabaseConnection {
     this.activeHoldId = null;
   }
 
+  protected get logger() {
+    return this.options.logger;
+  }
+
+  protected get dbEntry() {
+    return this.options.dbMap.get(this.options.dbFilename)!;
+  }
+
+  protected get connection() {
+    return this.dbEntry.db;
+  }
+
+  protected get clientIds() {
+    return this.dbEntry.clientIds;
+  }
+
   async init(): Promise<void> {
     // No-op since the connection is already initialized when it was created
   }
@@ -51,20 +67,8 @@ export class SharedWASQLiteConnection implements AsyncDatabaseConnection {
     }
   }
 
-  protected get logger() {
-    return this.options.logger;
-  }
-
-  protected get dbEntry() {
-    return this.options.dbMap.get(this.options.dbFilename)!;
-  }
-
-  protected get connection() {
-    return this.dbEntry.db;
-  }
-
-  protected get clientIds() {
-    return this.dbEntry.clientIds;
+  async isAutoCommit(): Promise<boolean> {
+    return this.connection.isAutoCommit();
   }
 
   /**

packages/web/src/worker/db/WASQLiteDB.worker.ts

@@ -9,7 +9,7 @@ import { AsyncDatabaseConnection } from '../../db/adapters/AsyncDatabaseConnecti
 import { WorkerDBOpenerOptions } from '../../db/adapters/wa-sqlite/WASQLiteOpenFactory';
 import { getNavigatorLocks } from '../../shared/navigator';
 import { SharedDBWorkerConnection, SharedWASQLiteConnection } from './SharedWASQLiteConnection';
-import { WorkerWASQLiteConnection, proxyWASQLiteConnection } from './WorkerWASQLiteConnection';
+import { WorkerWASQLiteConnection } from './WorkerWASQLiteConnection';
 
 const baseLogger = createBaseLogger();
 baseLogger.useDefaults();
@@ -59,7 +59,7 @@ const openDBShared = async (options: WorkerDBOpenerOptions): Promise<AsyncDataba
       logger
     });
 
-    return proxyWASQLiteConnection(sharedConnection);
+    return Comlink.proxy(sharedConnection);
   });
 };
 

packages/web/src/worker/db/WorkerWASQLiteConnection.ts

@@ -1,26 +1,11 @@
 import * as Comlink from 'comlink';
-import { AsyncDatabaseConnection, OnTableChangeCallback } from '../../db/adapters/AsyncDatabaseConnection';
+import { OnTableChangeCallback } from '../../db/adapters/AsyncDatabaseConnection';
 import { WASqliteConnection } from '../../db/adapters/wa-sqlite/WASQLiteConnection';
 
 /**
- * Fully proxies a WASQLiteConnection to be used as an AsyncDatabaseConnection.
+ * A Small proxy wrapper around the WASqliteConnection.
+ * This ensures that certain return types are properly proxied.
  */
-export function proxyWASQLiteConnection(connection: AsyncDatabaseConnection): AsyncDatabaseConnection {
-  return Comlink.proxy({
-    init: Comlink.proxy(() => connection.init()),
-    close: Comlink.proxy(() => connection.close()),
-    markHold: Comlink.proxy(() => connection.markHold()),
-    releaseHold: Comlink.proxy((holdId: string) => connection.releaseHold(holdId)),
-    execute: Comlink.proxy((sql: string, params?: any[]) => connection.execute(sql, params)),
-    executeRaw: Comlink.proxy((sql: string, params?: any[]) => connection.executeRaw(sql, params)),
-    executeBatch: Comlink.proxy((sql: string, params?: any[]) => connection.executeBatch(sql, params)),
-    registerOnTableChange: Comlink.proxy((callback: OnTableChangeCallback) =>
-      connection.registerOnTableChange(callback)
-    ),
-    getConfig: Comlink.proxy(() => connection.getConfig())
-  });
-}
-
 export class WorkerWASQLiteConnection extends WASqliteConnection {
   async registerOnTableChange(callback: OnTableChangeCallback): Promise<() => void> {
     // Proxy the callback remove function

packages/web/tests/multiple_instances.test.ts

@@ -3,26 +3,31 @@ import {
   createBaseLogger,
   createLogger,
   DEFAULT_CRUD_UPLOAD_THROTTLE_MS,
-  DEFAULT_STREAMING_SYNC_OPTIONS,
   SqliteBucketStorage,
   SyncStatus
 } from '@powersync/common';
 import {
+  OpenAsyncDatabaseConnection,
   SharedWebStreamingSyncImplementation,
   SharedWebStreamingSyncImplementationOptions,
+  WASqliteConnection,
   WebRemote
 } from '@powersync/web';
-
-import { beforeAll, describe, expect, it, vi } from 'vitest';
+import * as Comlink from 'comlink';
+import { beforeAll, describe, expect, it, onTestFinished, vi } from 'vitest';
+import { LockedAsyncDatabaseAdapter } from '../src/db/adapters/LockedAsyncDatabaseAdapter';
 import { WebDBAdapter } from '../src/db/adapters/WebDBAdapter';
+import { WorkerWrappedAsyncDatabaseConnection } from '../src/db/adapters/WorkerWrappedAsyncDatabaseConnection';
 import { TestConnector } from './utils/MockStreamOpenFactory';
 import { generateTestDb, testSchema } from './utils/testDb';
 
+const DB_FILENAME = 'test-multiple-instances.db';
+
 describe('Multiple Instances', { sequential: true }, () => {
   const openDatabase = () =>
     generateTestDb({
       database: {
-        dbFilename: `test-multiple-instances.db`
+        dbFilename: DB_FILENAME
       },
       schema: testSchema
     });
@@ -99,6 +104,106 @@ describe('Multiple Instances', { sequential: true }, () => {
     await createAsset(powersync2);
   });
 
+  it('should handled interrupted transactions', { timeout: Infinity }, async () => {
+    //Create a shared PowerSync database. We'll just use this for internally managing connections.
+    const powersync = openDatabase();
+    await powersync.init();
+
+    // Now get a shared connection to the same database
+    const webAdapter = powersync.database as WebDBAdapter;
+
+    // Allow us to share the connection. This is what shared sync workers will use.
+    const shared = await webAdapter.shareConnection();
+    const config = webAdapter.getConfiguration();
+    const opener = Comlink.wrap<OpenAsyncDatabaseConnection>(shared.port);
+
+    // Open up a shared connection
+    const initialSharedConnection = (await opener(config)) as Comlink.Remote<WASqliteConnection>;
+    onTestFinished(async () => {
+      await initialSharedConnection.close();
+    });
+
+    // This will simulate another subsequent shared connection
+    const subsequentSharedConnection = (await opener(config)) as Comlink.Remote<WASqliteConnection>;
+    onTestFinished(async () => {
+      await subsequentSharedConnection.close();
+    });
+
+    // In the beginning, we should not be in a transaction
+    const isAutoCommit = await initialSharedConnection.isAutoCommit();
+    // Should be true initially
+    expect(isAutoCommit).true;
+
+    // Now we'll simulate the locked connections which are used by the shared sync worker
+    const wrappedInitialSharedConnection = new WorkerWrappedAsyncDatabaseConnection({
+      baseConnection: initialSharedConnection,
+      identifier: DB_FILENAME,
+      remoteCanCloseUnexpectedly: true,
+      remote: opener
+    });
+
+    // Wrap the second connection in a locked adapter, this simulates the actual use case
+    const lockedInitialConnection = new LockedAsyncDatabaseAdapter({
+      name: DB_FILENAME,
+      openConnection: async () => wrappedInitialSharedConnection
+    });
+
+    // Allows us to unblock a transaction which is awaiting a promise
+    let unblockTransaction: (() => void) | undefined;
+
+    // Start a transaction that will be interrupted
+    const transactionPromise = lockedInitialConnection.writeTransaction(async (tx) => {
+      // Transaction should be started now
+
+      // Wait till we are unblocked. Keep this transaction open.
+      await new Promise<void>((resolve) => {
+        unblockTransaction = resolve;
+      });
+
+      // This should throw if the db was closed
+      await tx.get('SELECT 1');
+    });
+
+    // Wait for the transaction to have started
+    await vi.waitFor(() => expect(unblockTransaction).toBeDefined(), { timeout: 2000 });
+
+    // Since we're in a transaction from above
+    expect(await initialSharedConnection.isAutoCommit()).false;
+
+    // The in-use connection should be closed now
+    // This simulates a tab being closed.
+    await wrappedInitialSharedConnection.close();
+    wrappedInitialSharedConnection.markRemoteClosed();
+
+    // The transaction should be unblocked now
+    unblockTransaction?.();
+
+    // Since we closed while in the transaction, the execution call should have thrown
+    await expect(transactionPromise).rejects.toThrow('Called operation on closed remote');
+
+    // It will still be false until we request a new hold
+    // Requesting a new hold will cleanup the previous transaction.
+    expect(await subsequentSharedConnection.isAutoCommit()).false;
+
+    // Allows us to simulate a new locked shared connection.
+    const lockedSubsequentConnection = new LockedAsyncDatabaseAdapter({
+      name: DB_FILENAME,
+      openConnection: async () =>
+        new WorkerWrappedAsyncDatabaseConnection({
+          baseConnection: subsequentSharedConnection,
+          identifier: DB_FILENAME,
+          remoteCanCloseUnexpectedly: true,
+          remote: opener
+        })
+    });
+
+    // Starting a new transaction should work cleanup the old and work as expected
+    await lockedSubsequentConnection.writeTransaction(async (tx) => {
+      await tx.get('SELECT 1');
+      expect(await subsequentSharedConnection.isAutoCommit()).false;
+    });
+  });
+
   it('should watch table changes between instances', async () => {
     const db1 = openDatabase();
     const db2 = openDatabase();