Web: In-memory connection pool

.changeset/funny-kids-fix.md

@@ -0,0 +1,5 @@
+---
+'@powersync/web': minor
+---
+
+Add `@powersync/web/extra/shared-memory-pool`, a multi-threaded in-memory connection pool useful for high-performance workloads where persistence is not required.

packages/web/package.json

@@ -23,6 +23,10 @@
       "react-native": "./dist/index.react_native_web.js",
       "default": "./lib/index.js"
     },
+    "./extra/shared-memory-pool": {
+      "types": "./lib/db/adapters/memory-pool/client.d.ts",
+      "default": "./lib/db/adapters/memory-pool/client.js"
+    },
     "./bundled_worker": {
       "types": "./lib/worker/worker.d.ts",
       "default": "./dist/worker/worker.js"
@@ -44,7 +48,7 @@
     "clean": "rm -rf lib dist tsconfig.tsbuildinfo",
     "watch": "tsc --build -w",
     "test": "pnpm build && vitest",
-    "test:exports": "attw --pack . --entrypoints . --profile=esm-only "
+    "test:exports": "attw --pack . --entrypoints . extra/shared-memory-pool --profile=esm-only "
   },
   "keywords": [
     "data sync",

packages/web/src/db/PowerSyncDatabase.ts

@@ -172,27 +172,34 @@ export class WebPowerSyncDatabase extends BasePowerSyncDatabase<WebPowerSyncData
       logger: this.logger
     };
 
-    switch (true) {
-      case this.resolvedOpenOptions.ssrMode:
-        return new SSRStreamingSyncImplementation();
-      case this.resolvedOpenOptions.enableMultiTabs:
-        if (!this.enableBroadcastLogs) {
-          const warning = `
+    if (this.resolvedOpenOptions.ssrMode) {
+      return new SSRStreamingSyncImplementation();
+    } else if (this.resolvedOpenOptions.enableMultiTabs) {
+      if (!this.enableBroadcastLogs) {
+        const warning = `
             Multiple tabs are enabled, but broadcasting of logs is disabled.
             Logs for shared sync worker will only be available in the shared worker context
           `;
-          const logger = this.options.logger;
-          logger ? logger.log({ level: LogLevels.warn, message: warning }) : console.warn(warning);
-        }
+        const logger = this.options.logger;
+        logger ? logger.log({ level: LogLevels.warn, message: warning }) : console.warn(warning);
+      }
+
+      if ('shareConnection' in this.database) {
         return new SharedWebStreamingSyncImplementation({
           ...syncOptions,
           db: this.database as WebDBAdapter, // This should always be the case
           logLevel: this.options.sync?.logLevel ?? LogLevels.info,
           enableBroadcastLogs: this.enableBroadcastLogs
         });
-      default:
-        return new TabLocalStreamingSyncImplementation(syncOptions);
+      }
+
+      this.logger.log({
+        level: LogLevels.warn,
+        message: "Not using a shared sync worker because the database adapter doesn't support it."
+      });
     }
+
+    return new TabLocalStreamingSyncImplementation(syncOptions);
   }
 }
 /**

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

@@ -1,7 +1,8 @@
 import { DBAdapter, DBAdapterListener, DBLockOptions, LockContext } from '@powersync/common';
-import { Mutex, Semaphore, UnlockFn } from '@powersync/shared-internals';
-import { SharedConnectionWorker, WebDBAdapter, WebDBAdapterConfiguration } from './WebDBAdapter.js';
+import { Mutex, Semaphore } from '@powersync/shared-internals';
+import { SharedConnectionWorker, WebDBAdapterConfiguration } from './WebDBAdapter.js';
 import { DatabaseClient } from './wa-sqlite/DatabaseClient.js';
+import { acquireFromPool } from './acquireFromPool.js';
 
 type PendingListener = { listener: Partial<DBAdapterListener>; closeAfterRegisteredOnResolvedPool?: () => void };
 
@@ -126,65 +127,16 @@ function readWritePoolState(writer: DatabaseClient, readers: DatabaseClient[]):
   return {
     writer,
     async withConnection(allowReadOnly, fn, options) {
-      const abortController = new AbortController();
-      const abortSignal = abortController.signal;
-
-      let timeout: any = null;
-      let release: UnlockFn | undefined;
-      if (options?.timeoutMs) {
-        timeout = setTimeout(() => abortController.abort('requesting database timed out'), options.timeoutMs);
-      }
-
-      try {
-        if (allowReadOnly) {
-          let connection: DatabaseClient;
-
-          // Even if we have a pool of read connections, it's typically very small and we assume that most queries are
-          // reads. So, we want to request any connection from the read pool and the dedicated write connection (which
-          // can also serve reads). We race for the first connection we can obtain this way, and then abort the other
-          // request.
-          [connection, release] = await new Promise<[DatabaseClient, UnlockFn]>((resolve, reject) => {
-            let didComplete = false;
-            function complete() {
-              didComplete = true;
-              abortController.abort();
-            }
-
-            function completeSuccess(connection: DatabaseClient, returnFn: UnlockFn) {
-              if (didComplete) {
-                // We're not going to use this connection, so return it immediately.
-                returnFn();
-              } else {
-                complete();
-                resolve([connection, returnFn]);
-              }
-            }
-
-            function completeError(error: unknown) {
-              // We either have a working connection already, or we've rejected the promise. Either way, we don't need
-              // to do either thing again.
-              if (didComplete) return;
-
-              complete();
-              reject(error);
-            }
-
-            writerMutex.acquire(abortSignal).then((unlock) => completeSuccess(writer, unlock), completeError);
-            readerSemaphore
-              .requestOne(abortSignal)
-              .then(({ item, release }) => completeSuccess(item, release), completeError);
-          });
-
-          return await connection.readLock(fn);
-        } else {
-          return await writerMutex.runExclusive(() => writer.writeLock(fn), abortSignal);
-        }
-      } finally {
-        if (timeout != null) {
-          clearTimeout(timeout);
-        }
-        release?.();
-      }
+      return acquireFromPool(
+        writerMutex,
+        writer,
+        readerSemaphore,
+        (connection) => {
+          return allowReadOnly ? connection.readLock(fn) : connection.writeLock(fn);
+        },
+        options,
+        allowReadOnly
+      );
     },
     async close() {
       await writer.close();

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

@@ -0,0 +1,73 @@
+import type { DBLockOptions } from '@powersync/common';
+import type { Mutex, Semaphore, UnlockFn } from '@powersync/shared-internals';
+
+/**
+ * Internal helper function to acquire a connection from a pool that has a designated writer, additional readers, and
+ * also allows dispatching reads to the writer.
+ */
+export async function acquireFromPool<Connection, Res>(
+  writerMutex: Mutex,
+  writer: Connection,
+  readers: Semaphore<Connection> | undefined,
+  callback: (connection: Connection) => Promise<Res>,
+  options: DBLockOptions | undefined,
+  allowReadOnly: boolean
+): Promise<Res> {
+  const abortController = new AbortController();
+  const abortSignal = abortController.signal;
+
+  let timeout: any = null;
+  let release: UnlockFn | undefined;
+  if (options?.timeoutMs) {
+    timeout = setTimeout(() => abortController.abort('requesting database timed out'), options.timeoutMs);
+  }
+
+  try {
+    if (allowReadOnly) {
+      let connection: Connection;
+
+      // Even if we have a pool of read connections, it's typically very small and we assume that most queries are
+      // reads. So, we want to request any connection from the read pool and the dedicated write connection (which
+      // can also serve reads). We race for the first connection we can obtain this way, and then abort the other
+      // request.
+      [connection, release] = await new Promise<[Connection, UnlockFn]>((resolve, reject) => {
+        let didComplete = false;
+        function complete() {
+          didComplete = true;
+          abortController.abort();
+        }
+
+        function completeSuccess(connection: Connection, returnFn: UnlockFn) {
+          if (didComplete) {
+            // We're not going to use this connection, so return it immediately.
+            returnFn();
+          } else {
+            complete();
+            resolve([connection, returnFn]);
+          }
+        }
+
+        function completeError(error: unknown) {
+          // We either have a working connection already, or we've rejected the promise. Either way, we don't need
+          // to do either thing again.
+          if (didComplete) return;
+
+          complete();
+          reject(error);
+        }
+
+        writerMutex.acquire(abortSignal).then((unlock) => completeSuccess(writer, unlock), completeError);
+        readers?.requestOne(abortSignal).then(({ item, release }) => completeSuccess(item, release), completeError);
+      });
+
+      return await callback(connection);
+    } else {
+      return await writerMutex.runExclusive(() => callback(writer), abortSignal);
+    }
+  } finally {
+    if (timeout != null) {
+      clearTimeout(timeout);
+    }
+    release?.();
+  }
+}