Fix a few minor concurrency issues

Author: simolus3

.changeset/backport-1053-crudlock.md

@@ -0,0 +1,5 @@
+---
+'@powersync/common': patch
+---
+
+Internal: Fix obtaining crud lock not being abortable.

.changeset/backport-1053-timeout.md

@@ -0,0 +1,5 @@
+---
+'@powersync/web': patch
+---
+
+Fix timeout option having no effect with OPFS WriteAhead VFS.

.changeset/backport-1053-worker-errors.md

@@ -0,0 +1,5 @@
+---
+'@powersync/web': patch
+---
+
+Log error when a worker fails to load.

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -970,7 +970,7 @@ SELECT * FROM crud_entries;
 
   /**
    * Open a read-only transaction.
-   * Read transactions can run concurrently to a write transaction.
+   * When multiple connections are available, read transactions can run concurrently to a write transaction.
    * Changes from any write transaction are not visible to read transactions started before it.
    *
    * @param callback - Function to execute within the transaction

packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -379,6 +379,7 @@ export abstract class AbstractStreamingSyncImplementation
   private async _uploadAllCrud(signal: AbortSignal): Promise<void> {
     return this.obtainLock({
       type: LockType.CRUD,
+      signal,
       callback: async () => {
         /**
          * Keep track of the first item in the CRUD queue for the last `uploadCrud` iteration.

packages/common/src/utils/ControlledExecutor.ts

@@ -65,14 +65,17 @@ export class ControlledExecutor<T> {
 
   private async execute(param: T) {
     this.runningTask = this.task(param);
-    await this.runningTask;
-    this.runningTask = undefined;
+    try {
+      await this.runningTask;
+    } finally {
+      this.runningTask = undefined;
 
-    if (this.pendingTaskParam) {
-      const pendingParam = this.pendingTaskParam;
-      this.pendingTaskParam = undefined;
+      if (this.pendingTaskParam) {
+        const pendingParam = this.pendingTaskParam;
+        this.pendingTaskParam = undefined;
 
-      this.execute(pendingParam);
+        this.execute(pendingParam);
+      }
     }
   }
 }

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

@@ -126,7 +126,7 @@ function readWritePoolState(writer: DatabaseClient, readers: DatabaseClient[]):
       let timeout: any = null;
       let release: UnlockFn | undefined;
       if (options?.timeoutMs) {
-        timeout = setTimeout(() => abortController.abort, options.timeoutMs);
+        timeout = setTimeout(() => abortController.abort('requesting database timed out'), options.timeoutMs);
       }
 
       try {

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

@@ -135,16 +135,24 @@ export class WASQLiteOpenFactory implements SQLOpenFactory {
       ): Promise<DatabaseClient> => {
         const workerPort =
           typeof optionsDbWorker == 'function'
-            ? resolveWorkerDatabasePortFactory(() =>
-                optionsDbWorker({
-                  ...this.options,
-                  temporaryStorage,
-                  cacheSizeKb,
-                  flags: this.resolvedFlags,
-                  encryptionKey
-                })
+            ? resolveWorkerDatabasePortFactory(
+                () =>
+                  optionsDbWorker({
+                    ...this.options,
+                    temporaryStorage,
+                    cacheSizeKb,
+                    flags: this.resolvedFlags,
+                    encryptionKey
+                  }),
+                this.logger
               )
-            : openWorkerDatabasePort(this.options.dbFilename, enableMultiTabs, optionsDbWorker, this.waOptions.vfs);
+            : openWorkerDatabasePort(
+                this.options.dbFilename,
+                enableMultiTabs,
+                optionsDbWorker,
+                this.waOptions.vfs,
+                this.logger
+              );
 
         const source = Comlink.wrap<OpenWorkerConnection>(workerPort);
         const closeSignal = new AbortController();

packages/web/src/db/sync/SharedWebStreamingSyncImplementation.ts

@@ -5,7 +5,6 @@ import {
   SyncStatusOptions
 } from '@powersync/common';
 import * as Comlink from 'comlink';
-import { getNavigatorLocks } from '../../shared/navigator.js';
 import { AbstractSharedSyncClientProvider } from '../../worker/sync/AbstractSharedSyncClientProvider.js';
 import { ManualSharedSyncPayload, SharedSyncClientEvent } from '../../worker/sync/SharedSyncImplementation.js';
 import { WorkerClient } from '../../worker/sync/WorkerClient.js';
@@ -16,6 +15,7 @@ import {
   WebStreamingSyncImplementationOptions
 } from './WebStreamingSyncImplementation.js';
 import { generateTabCloseSignal } from '../../shared/tab_close_signal.js';
+import { logWorkerErrors } from '../../worker/errors.js';
 
 /**
  * The shared worker will trigger methods on this side of the message port
@@ -128,22 +128,23 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
     const syncWorker = options.sync?.worker;
     if (syncWorker) {
       if (typeof syncWorker === 'function') {
-        this.messagePort = syncWorker(resolvedWorkerOptions).port;
+        this.messagePort = this.workerPort(syncWorker(resolvedWorkerOptions));
       } else {
-        this.messagePort = new SharedWorker(`${syncWorker}`, {
-          /* @vite-ignore */
-          name: `shared-sync-${this.webOptions.identifier}`
-        }).port;
+        this.messagePort = this.workerPort(
+          new SharedWorker(`${syncWorker}`, {
+            /* @vite-ignore */
+            name: `shared-sync-${this.webOptions.identifier}`
+          })
+        );
       }
     } else {
-      this.messagePort = new SharedWorker(
-        new URL('../../worker/sync/SharedSyncImplementation.worker.js', import.meta.url),
-        {
+      this.messagePort = this.workerPort(
+        new SharedWorker(new URL('../../worker/sync/SharedSyncImplementation.worker.js', import.meta.url), {
           /* @vite-ignore */
           name: `shared-sync-${this.webOptions.identifier}`,
           type: 'module'
-        }
-      ).port;
+        })
+      );
     }
 
     /**
@@ -179,6 +180,11 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
     this.isInitialized = this._init();
   }
 
+  private workerPort(worker: SharedWorker): MessagePort {
+    logWorkerErrors(worker, this.logger);
+    return worker.port;
+  }
+
   protected async _init() {
     /**
      * The general flow of initialization is:

packages/web/src/worker/db/open-worker-database.ts

@@ -1,6 +1,8 @@
 import * as Comlink from 'comlink';
 import { vfsRequiresDedicatedWorkers, WASQLiteVFS } from '../../db/adapters/wa-sqlite/vfs.js';
 import { OpenWorkerConnection } from '../../db/adapters/wa-sqlite/DatabaseClient.js';
+import type { ILogger } from '@powersync/common';
+import { logWorkerErrors } from '../errors.js';
 
 /**
  * Opens a shared or dedicated worker which exposes opening of database connections
@@ -9,39 +11,45 @@ export function openWorkerDatabasePort(
   workerIdentifier: string,
   multipleTabs = true,
   worker: string | URL = '',
-  vfs?: WASQLiteVFS
+  vfs?: WASQLiteVFS,
+  logger?: ILogger
 ) {
   const needsDedicated = vfs && vfsRequiresDedicatedWorkers(vfs);
+  let resolvedWorker: Worker | SharedWorker;
 
   if (worker) {
-    return !needsDedicated && multipleTabs
-      ? new SharedWorker(`${worker}`, {
-          /* @vite-ignore */
-          name: `shared-DB-worker-${workerIdentifier}`
-        }).port
-      : new Worker(`${worker}`, {
-          /* @vite-ignore */
-          name: `DB-worker-${workerIdentifier}`
-        });
+    resolvedWorker =
+      !needsDedicated && multipleTabs
+        ? new SharedWorker(`${worker}`, {
+            /* @vite-ignore */
+            name: `shared-DB-worker-${workerIdentifier}`
+          })
+        : new Worker(`${worker}`, {
+            /* @vite-ignore */
+            name: `DB-worker-${workerIdentifier}`
+          });
   } else {
     /**
      *  Webpack V5 can bundle the worker automatically if the full Worker constructor syntax is used
      *  https://webpack.js.org/guides/web-workers/
      *  This enables multi tab support by default, but falls back if SharedWorker is not available
      *  (in the case of Android)
      */
-    return !needsDedicated && multipleTabs
-      ? new SharedWorker(new URL('./WASQLiteDB.worker.js', import.meta.url), {
-          /* @vite-ignore */
-          name: `shared-DB-worker-${workerIdentifier}`,
-          type: 'module'
-        }).port
-      : new Worker(new URL('./WASQLiteDB.worker.js', import.meta.url), {
-          /* @vite-ignore */
-          name: `DB-worker-${workerIdentifier}`,
-          type: 'module'
-        });
+    resolvedWorker =
+      !needsDedicated && multipleTabs
+        ? new SharedWorker(new URL('./WASQLiteDB.worker.js', import.meta.url), {
+            /* @vite-ignore */
+            name: `shared-DB-worker-${workerIdentifier}`,
+            type: 'module'
+          })
+        : new Worker(new URL('./WASQLiteDB.worker.js', import.meta.url), {
+            /* @vite-ignore */
+            name: `DB-worker-${workerIdentifier}`,
+            type: 'module'
+          });
   }
+
+  return resolveWorkerDatabasePortFactory(() => resolvedWorker, logger);
 }
 
 /**
@@ -52,8 +60,12 @@ export function getWorkerDatabaseOpener(workerIdentifier: string, multipleTabs =
   return Comlink.wrap<OpenWorkerConnection>(openWorkerDatabasePort(workerIdentifier, multipleTabs, worker));
 }
 
-export function resolveWorkerDatabasePortFactory(worker: () => Worker | SharedWorker) {
+export function resolveWorkerDatabasePortFactory(worker: () => Worker | SharedWorker, logger?: ILogger) {
   const workerInstance = worker();
+  if (logger) {
+    logWorkerErrors(workerInstance, logger);
+  }
+
   return isSharedWorker(workerInstance) ? workerInstance.port : workerInstance;
 }