Merged serializeWriter into singleton pool to ensure reentrance

Author: oskardudycz

src/packages/dumbo/src/storage/sqlite/core/pool/pool.ts

@@ -12,10 +12,6 @@ import {
   createSingletonConnectionPool,
   type ConnectionPool,
 } from '../../../../core';
-import {
-  sqliteDualConnectionPool,
-  type SQLiteDualPoolOptions,
-} from './dualPool';
 
 export type SQLiteFileNameOrConnectionString =
   | {
@@ -151,7 +147,6 @@ export type SQLitePoolOptions<
       ConnectionOptions
     >
   | SQLiteAmbientConnectionPoolOptions<SQLiteConnectionType>
-  | SQLiteDualPoolOptions<SQLiteConnectionType, ConnectionOptions>
 ) & {
   driverType: SQLiteConnectionType['driverType'];
   serializer?: JSONSerializer;
@@ -204,10 +199,7 @@ export const toSqlitePoolOptions = <
     >;
   }
 
-  return { ...rest, dual: true } as SQLitePoolOptions<
-    SQLiteConnectionType,
-    ConnectionOptions
-  >;
+  return rest as SQLitePoolOptions<SQLiteConnectionType, ConnectionOptions>;
 };
 
 export function sqlitePool<
@@ -234,12 +226,6 @@ export function sqlitePool<
       ).connection,
     });
 
-  if ('dual' in options && options.dual) {
-    return sqliteDualConnectionPool(
-      options as SQLiteDualPoolOptions<SQLiteConnectionType, ConnectionOptions>,
-    );
-  }
-
   if (
     options.singleton === true &&
     (

src/packages/dumbo/src/storage/sqlite/sqlite3/index.ts

@@ -8,6 +8,7 @@ import {
 } from '../../../core';
 import {
   DefaultSQLiteMigratorOptions,
+  isInMemoryDatabase,
   SQLiteConnectionString,
   sqliteFormatter,
   sqliteMetadata,
@@ -22,7 +23,8 @@ import {
   type SQLite3Connection,
   type SQLite3ConnectionOptions,
 } from './connections';
-import { serializeSqlite3WriterPool } from './serializeWriter';
+import { sqliteDualConnectionPool } from './pool/dualPool';
+import { sqlite3SingletonPool } from './pool/singletonPool';
 
 export type SQLite3DumboOptions = Omit<
   SQLitePoolOptions<SQLite3Connection, SQLite3ConnectionOptions>,
@@ -34,33 +36,50 @@ export type SQLite3PoolOptions = SQLite3DumboOptions;
 
 export type Sqlite3Pool = SQLitePool<SQLite3Connection>;
 
-export const sqlite3Pool = (options: SQLite3DumboOptions) => {
-  const pool = sqlitePool(
-    toSqlitePoolOptions({
-      ...options,
-      driverType: SQLite3DriverType,
-      ...('connection' in options
-        ? {}
-        : {
-            connectionOptions: options,
-            sqliteConnectionFactory: (opts: SQLite3ConnectionOptions) =>
-              sqlite3Connection({
-                ...opts,
-                serializer: options.serializer ?? JSONSerializer,
-              }),
-          }),
-    }),
-  );
+export const sqlite3Pool = (
+  options: SQLite3DumboOptions,
+): SQLitePool<SQLite3Connection> => {
+  // Ambient: caller-managed connection. No acquisition, no serialisation.
+  if ('connection' in options && options.connection) {
+    return sqlitePool(
+      toSqlitePoolOptions({
+        ...options,
+        driverType: SQLite3DriverType,
+      }),
+    );
+  }
+
+  const sqliteConnectionFactory = (opts: SQLite3ConnectionOptions) =>
+    sqlite3Connection({
+      ...opts,
+      serializer: options.serializer ?? JSONSerializer,
+    });
 
-  // Ambient pools wrap a connection the caller already holds; serialising on
-  // top of them would double-lock and defeat the purpose. Anything else gets
-  // wrapped so writer-bound calls (withConnection, withTransaction, command,
-  // batchCommand) serialise through a single TaskProcessor, with ALS-based
-  // reentrancy so nested calls from inside an active writer task bypass the
-  // queue instead of deadlocking.
-  if ('connection' in options && options.connection) return pool;
+  // Singleton-shaped: in-memory DBs, an explicit client, or `singleton: true`.
+  // One connection shared across callers — wrap it so concurrent callers
+  // serialise through a single-slot TaskProcessor with ALS-based reentrancy.
+  const isSingleton =
+    isInMemoryDatabase(options) ||
+    ('client' in options && Boolean(options.client)) ||
+    options.singleton === true;
+
+  if (isSingleton) {
+    return sqlite3SingletonPool<SQLite3Connection>({
+      driverType: SQLite3DriverType,
+      getConnection: () => sqliteConnectionFactory(options),
+    });
+  }
 
-  return serializeSqlite3WriterPool(pool);
+  // Default: file-backed dual pool. Its writer side is serialised inside
+  // sqliteDualConnectionPool via the same primitive.
+  const readerPoolSize = (options as { readerPoolSize?: number })
+    .readerPoolSize;
+  return sqliteDualConnectionPool({
+    driverType: SQLite3DriverType,
+    sqliteConnectionFactory,
+    connectionOptions: options,
+    ...(readerPoolSize !== undefined ? { readerPoolSize } : {}),
+  });
 };
 
 const tryParseConnectionString = (connectionString: string) => {

…ge/sqlite/core/pool/dualPool.int.spec.ts …sqlite/sqlite3/pool/dualPool.int.spec.tssrc/packages/dumbo/src/storage/sqlite/core/pool/dualPool.int.spec.ts renamed to src/packages/dumbo/src/storage/sqlite/sqlite3/pool/dualPool.int.spec.ts src/packages/dumbo/src/storage/sqlite/core/pool/dualPool.int.spec.ts renamed to src/packages/dumbo/src/storage/sqlite/sqlite3/pool/dualPool.int.spec.ts

@@ -1,16 +1,14 @@
 import { cpus } from 'os';
-import {
-  createBoundedConnectionPool,
-  createSingletonConnectionPool,
-} from '../../../../core';
+import { createBoundedConnectionPool } from '../../../../core';
 import { guardInitializedOnce } from '../../../../core/taskProcessing';
 import type {
   AnySQLiteConnection,
   SQLiteConnectionFactory,
   SQLiteConnectionOptions,
-} from '../connections';
-import { mapSqliteError } from '../errors';
-import type { SQLitePool } from './pool';
+  SQLitePool,
+} from '../../core';
+import { mapSqliteError } from '../../core/errors';
+import { sqlite3SingletonPool } from './singletonPool';
 
 export type SQLiteDualPoolOptions<
   SQLiteConnectionType extends AnySQLiteConnection,
@@ -89,7 +87,7 @@ export const sqliteDualConnectionPool = <
     return connection;
   };
 
-  const writerPool = createSingletonConnectionPool({
+  const writerPool = sqlite3SingletonPool<SQLiteConnectionType>({
     driverType: options.driverType,
     getConnection: () => wrappedConnectionFactory(false, connectionOptions),
   });

…src/storage/sqlite/core/pool/dualPool.ts …/storage/sqlite/sqlite3/pool/dualPool.tssrc/packages/dumbo/src/storage/sqlite/core/pool/dualPool.ts renamed to src/packages/dumbo/src/storage/sqlite/sqlite3/pool/dualPool.ts src/packages/dumbo/src/storage/sqlite/core/pool/dualPool.ts renamed to src/packages/dumbo/src/storage/sqlite/sqlite3/pool/dualPool.ts

@@ -0,0 +1,80 @@
+import { AsyncLocalStorage } from 'node:async_hooks';
+import {
+  createSingletonConnectionPool,
+  type ConnectionPool,
+} from '../../../../core';
+import { TaskProcessor } from '../../../../core/taskProcessing';
+import type { AnySQLiteConnection } from '../../core';
+
+export type SQLite3SingletonPoolOptions<
+  SQLiteConnectionType extends AnySQLiteConnection,
+> = {
+  driverType: SQLiteConnectionType['driverType'];
+  getConnection: () => SQLiteConnectionType | Promise<SQLiteConnectionType>;
+  closeConnection?: (connection: SQLiteConnectionType) => void | Promise<void>;
+  maxQueueSize?: number;
+};
+
+// Creates a singleton-connection pool whose callers serialise through a
+// single-slot TaskProcessor. An AsyncLocalStorage flag lets re-entrant calls
+// (made from inside an active task on this pool) bypass the queue instead of
+// deadlocking — that's the path emmett relies on when a workflow handler
+// internally calls back through the same pool (e.g. messageStore.appendToStream).
+export const sqlite3SingletonPool = <
+  SQLiteConnectionType extends AnySQLiteConnection,
+>(
+  options: SQLite3SingletonPoolOptions<SQLiteConnectionType>,
+): ConnectionPool<SQLiteConnectionType> => {
+  const inner = createSingletonConnectionPool<SQLiteConnectionType>({
+    driverType: options.driverType,
+    getConnection: options.getConnection,
+    ...(options.closeConnection
+      ? { closeConnection: options.closeConnection }
+      : {}),
+  });
+
+  const taskProcessor = new TaskProcessor({
+    maxActiveTasks: 1,
+    maxQueueSize: options.maxQueueSize ?? 1000,
+  });
+  const insideWriterTask = new AsyncLocalStorage<true>();
+
+  const enqueue = <Result>(op: () => Promise<Result>): Promise<Result> => {
+    if (insideWriterTask.getStore() === true) {
+      return op();
+    }
+    return taskProcessor.enqueue(({ ack }) =>
+      insideWriterTask.run(true, async () => {
+        try {
+          return await op();
+        } finally {
+          ack();
+        }
+      }),
+    );
+  };
+
+  return {
+    driverType: inner.driverType,
+    connection: inner.connection.bind(inner),
+    transaction: inner.transaction.bind(inner),
+    withConnection: (handle, connectionOptions) =>
+      enqueue(() => inner.withConnection(handle, connectionOptions)),
+    withTransaction: (handle, transactionOptions) =>
+      enqueue(() => inner.withTransaction(handle, transactionOptions)),
+    execute: {
+      query: (sql, queryOptions) =>
+        enqueue(() => inner.execute.query(sql, queryOptions)),
+      batchQuery: (sqls, queryOptions) =>
+        enqueue(() => inner.execute.batchQuery(sqls, queryOptions)),
+      command: (sql, commandOptions) =>
+        enqueue(() => inner.execute.command(sql, commandOptions)),
+      batchCommand: (sqls, commandOptions) =>
+        enqueue(() => inner.execute.batchCommand(sqls, commandOptions)),
+    },
+    close: async () => {
+      await taskProcessor.stop();
+      await inner.close();
+    },
+  };
+};