Added single writer processing with isolation to SQLite3 connection pool

Author: oskardudycz

src/packages/dumbo/src/core/taskProcessing/executionGuards.ts

@@ -1,4 +1,5 @@
 import { v7 as uuid } from 'uuid';
+import { DumboError, TransientDatabaseError } from '../errors';
 import { TaskProcessor } from './taskProcessor';
 
 export type ExclusiveAccessGuard = {

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

@@ -22,6 +22,7 @@ import {
   type SQLite3Connection,
   type SQLite3ConnectionOptions,
 } from './connections';
+import { serializeSqlite3WriterPool } from './serializeWriter';
 
 export type SQLite3DumboOptions = Omit<
   SQLitePoolOptions<SQLite3Connection, SQLite3ConnectionOptions>,
@@ -33,8 +34,8 @@ export type SQLite3PoolOptions = SQLite3DumboOptions;
 
 export type Sqlite3Pool = SQLitePool<SQLite3Connection>;
 
-export const sqlite3Pool = (options: SQLite3DumboOptions) =>
-  sqlitePool(
+export const sqlite3Pool = (options: SQLite3DumboOptions) => {
+  const pool = sqlitePool(
     toSqlitePoolOptions({
       ...options,
       driverType: SQLite3DriverType,
@@ -51,6 +52,17 @@ export const sqlite3Pool = (options: SQLite3DumboOptions) =>
     }),
   );
 
+  // 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;
+
+  return serializeSqlite3WriterPool(pool);
+};
+
 const tryParseConnectionString = (connectionString: string) => {
   try {
     return SQLiteConnectionString(connectionString);

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

@@ -0,0 +1,58 @@
+import { AsyncLocalStorage } from 'node:async_hooks';
+import { TaskProcessor } from '../../../core/taskProcessing';
+import type { SQLitePool } from '../core';
+import type { SQLite3Connection } from './connections';
+
+export const serializeSqlite3WriterPool = (
+  pool: SQLitePool<SQLite3Connection>,
+  options?: { maxQueueSize?: number },
+): SQLitePool<SQLite3Connection> => {
+  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: pool.driverType,
+    connection: pool.connection.bind(pool),
+    transaction: pool.transaction.bind(pool),
+    withConnection: ((handle, connectionOptions) =>
+      connectionOptions?.readonly
+        ? pool.withConnection(handle, connectionOptions)
+        : enqueue(() =>
+            pool.withConnection(handle, connectionOptions),
+          )) as SQLitePool<SQLite3Connection>['withConnection'],
+    withTransaction: ((handle, transactionOptions) =>
+      enqueue(() =>
+        pool.withTransaction(handle, transactionOptions),
+      )) as SQLitePool<SQLite3Connection>['withTransaction'],
+    execute: {
+      query: pool.execute.query.bind(pool.execute),
+      batchQuery: pool.execute.batchQuery.bind(pool.execute),
+      command: (sql, commandOptions) =>
+        enqueue(() => pool.execute.command(sql, commandOptions)),
+      batchCommand: (sqls, commandOptions) =>
+        enqueue(() => pool.execute.batchCommand(sqls, commandOptions)),
+    },
+    close: async () => {
+      await taskProcessor.stop();
+      await pool.close();
+    },
+  };
+};

src/packages/dumbo/src/storage/sqlite/sqlite3/transactions/transactions.int.spec.ts

@@ -552,12 +552,8 @@ describe('SQLite3 Transactions', () => {
         }
       });
 
-      // TODO: parallel pool.withTransaction calls share one underlying SQLite
-      // transaction on the singleton writer connection. A rollback in one
-      // sibling rolls back the other's writes. Fix requires a writer queue
-      // (one real BEGIN/COMMIT per call), not just the finally-block tweak.
-      it.fails(
-        `isolates a failing parallel transaction from a successful one with ${testName} database`,
+      it(
+        `keeps a plain command outside of an open transaction with ${testName} database`,
         async () => {
           const pool = sqlite3Pool({
             fileName,
@@ -566,37 +562,87 @@ describe('SQLite3 Transactions', () => {
 
           try {
             await pool.execute.command(
-              SQL`CREATE TABLE isolation_test (id INTEGER PRIMARY KEY, value TEXT)`,
+              SQL`CREATE TABLE plain_isolation (id INTEGER PRIMARY KEY, value TEXT)`,
             );
 
-            let releaseA: () => void = () => {};
-            let releaseB: () => void = () => {};
-            const aMayProceed = new Promise<void>((r) => {
-              releaseA = r;
+            let releaseTx: () => void = () => {};
+            let plainMayStart: () => void = () => {};
+            const txMayProceed = new Promise<void>((r) => {
+              releaseTx = r;
+            });
+            const plainCanStart = new Promise<void>((r) => {
+              plainMayStart = r;
             });
-            const bMayProceed = new Promise<void>((r) => {
-              releaseB = r;
+
+            const txPromise = pool.withTransaction(async (tx) => {
+              await tx.execute.command(
+                SQL`INSERT INTO plain_isolation (id, value) VALUES (1, 'tx-row')`,
+              );
+              plainMayStart();
+              await txMayProceed;
+              throw new Error('tx intentional rollback');
             });
 
+            await plainCanStart;
+
+            const plainPromise = pool.execute.command(
+              SQL`INSERT INTO plain_isolation (id, value) VALUES (2, 'plain-row')`,
+            );
+
+            releaseTx();
+
+            await assert.rejects(txPromise, /tx intentional rollback/);
+            await plainPromise;
+
+            const rows = await pool.execute.query<{
+              id: number;
+              value: string;
+            }>(SQL`SELECT id, value FROM plain_isolation ORDER BY id`);
+
+            assert.deepStrictEqual(
+              rows.rows.map((r) => r.value),
+              ['plain-row'],
+            );
+          } finally {
+            await pool.close();
+          }
+        },
+      );
+
+      it(
+        `isolates a failing parallel transaction from a successful one with ${testName} database`,
+        async () => {
+          const pool = sqlite3Pool({
+            fileName,
+            transactionOptions: { allowNestedTransactions: true },
+          });
+
+          try {
+            await pool.execute.command(
+              SQL`CREATE TABLE isolation_test (id INTEGER PRIMARY KEY, value TEXT)`,
+            );
+
             const a = pool.withTransaction(async (tx) => {
               await tx.execute.command(
                 SQL`INSERT INTO isolation_test (id, value) VALUES (1, 'a-commits')`,
               );
-              releaseB();
-              await aMayProceed;
             });
 
             const b = pool.withTransaction(async (tx) => {
-              await bMayProceed;
               await tx.execute.command(
                 SQL`INSERT INTO isolation_test (id, value) VALUES (2, 'b-rolls-back')`,
               );
               throw new Error('B intentional failure');
             });
 
-            await assert.rejects(b, /B intentional failure/);
-            releaseA();
-            await a;
+            const results = await Promise.allSettled([a, b]);
+
+            assert.strictEqual(results[0]?.status, 'fulfilled');
+            assert.strictEqual(results[1]?.status, 'rejected');
+            assert.match(
+              (results[1] as PromiseRejectedResult).reason.message,
+              /B intentional failure/,
+            );
 
             const rows = await pool.execute.query<{
               id: number;