Add withConnectionTransaction for sibling storages on one handle.

packages/storage/CHANGELOG.md

@@ -20,6 +20,12 @@
 
 ## 0.3.45
 
+### Features
+
+#### storage
+
+- add `withConnectionTransaction` so sibling tabular storages on one SQLite, Postgres, or DuckDB handle commit and roll back together
+
 ## 0.3.44
 
 ## 0.3.43

packages/storage/src/browser.ts

@@ -8,10 +8,21 @@
 
 export * from "./common";
 
+// Connection-mutex seam. Everything below except `ConnectionReentryError` is a
+// provider-package internal, versioned in lockstep with this package rather
+// than covered by semver. Application code should use
+// `withConnectionTransaction` instead.
 export {
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
   __resetAlsForTesting,
   ConnectionReentryError,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  getAlsStore,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  isSynchronousAls,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
   runInTransactionOnConnection,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
   runOnConnection,
 } from "./tabular/ConnectionMutex.browser";
 

packages/storage/src/common-server.ts

@@ -8,13 +8,50 @@
 
 export * from "./common";
 
+// Connection-mutex seam. Everything below except `ConnectionReentryError` and
+// `NestedConnectionTransactionError` is a provider-package internal: the
+// vendor storages (`@workglow/sqlite`, `@workglow/postgres`,
+// `@workglow/duckdb`) build their `withConnectionTransaction` support on it
+// and are versioned in lockstep with this package. Application code should use
+// `withConnectionTransaction` instead; these names are not covered by semver.
 export {
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
   __resetAlsForTesting,
   ConnectionReentryError,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  getAlsStore,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  isSynchronousAls,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
   runInTransactionOnConnection,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
   runOnConnection,
 } from "./tabular/ConnectionMutex.server";
 
+export {
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  activeConnectionTxGroupHandle,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  assertSharedConnectionHandle,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  connectionTxQuery,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  deactivateConnectionTxStore,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  discardDeferredPuts,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  enqueueDeferredPut,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  isEnlistedInConnectionTx,
+  NestedConnectionTransactionError,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  runNativeConnectionTransaction,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  setConnectionTxQuery,
+  /** @internal — provider-package seam for withConnectionTransaction; not covered by semver. */
+  takeDeferredPuts,
+} from "./tabular/runNativeConnectionTransaction";
+
 export * from "./tabular/FsFolderTabularStorage";
 
 export * from "./kv/FsFolderJsonKvStorage";

packages/storage/src/common.ts

@@ -35,6 +35,7 @@ export * from "./tabular/HuggingFaceTabularStorage";
 export * from "./tabular/InMemoryTabularMigrationApplier";
 export * from "./tabular/InMemoryTabularStorage";
 export * from "./tabular/ITabularStorage";
+export * from "./tabular/withConnectionTransaction";
 export * from "./tabular/sqlMigrationDdl";
 export * from "./tabular/SqlTabularMigrationApplier";
 export * from "./tabular/StorageError";

packages/storage/src/tabular/BaseSqlTabularStorage.ts

@@ -119,6 +119,16 @@ export abstract class BaseSqlTabularStorage<
     return null;
   }
 
+  /**
+   * Public view of {@link connectionHandle} used by
+   * {@link withConnectionTransaction} to group participants. Postgres overrides
+   * this so a real `pg.Pool` still groups by pool identity even though
+   * {@link connectionHandle} stays `null` (pool writes must not serialize).
+   */
+  public sharedConnectionHandle(): object | null {
+    return this.connectionHandle();
+  }
+
   protected constructPrimaryKeyColumns($delimiter: string = ""): string {
     let cached = this._pkColsCache.get($delimiter);
     if (cached === undefined) {

packages/storage/src/tabular/ConnectionMutex.browser.ts

@@ -9,8 +9,13 @@ import { defineConnectionMutex } from "./defineConnectionMutex";
 
 export { ConnectionReentryError } from "./defineConnectionMutex";
 
-export const { runOnConnection, runInTransactionOnConnection, __resetAlsForTesting } =
-  defineConnectionMutex({
-    ensureAls,
-    __resetAlsForTesting: resetAlsForTesting,
-  });
+export const {
+  runOnConnection,
+  runInTransactionOnConnection,
+  getAlsStore,
+  isSynchronousAls,
+  __resetAlsForTesting,
+} = defineConnectionMutex({
+  ensureAls,
+  __resetAlsForTesting: resetAlsForTesting,
+});

packages/storage/src/tabular/ConnectionMutex.server.ts

@@ -9,8 +9,13 @@ import { defineConnectionMutex } from "./defineConnectionMutex";
 
 export { ConnectionReentryError } from "./defineConnectionMutex";
 
-export const { runOnConnection, runInTransactionOnConnection, __resetAlsForTesting } =
-  defineConnectionMutex({
-    ensureAls,
-    __resetAlsForTesting: resetAlsForTesting,
-  });
+export const {
+  runOnConnection,
+  runInTransactionOnConnection,
+  getAlsStore,
+  isSynchronousAls,
+  __resetAlsForTesting,
+} = defineConnectionMutex({
+  ensureAls,
+  __resetAlsForTesting: resetAlsForTesting,
+});

packages/storage/src/tabular/ITabularStorage.ts

@@ -622,6 +622,11 @@ export interface ITabularStorage<
    *
    * Use SAVEPOINT directly if you need nested rollback boundaries within a
    * single logical transaction.
+   *
+   * To commit writes across several storages that share one connection, use
+   * the free function `withConnectionTransaction(participants, fn)` instead
+   * of nesting `withTransaction` calls. Inside that callback, call ordinary
+   * methods on the original instances (there is no `tx` proxy).
    */
   withTransaction<T>(fn: (tx: this) => Promise<T>): Promise<T>;
 

packages/storage/src/tabular/__tests__/ConnectionMutex.test.ts

@@ -225,6 +225,39 @@ describe("ConnectionMutex F1: cross-instance re-entry is ALS-independent", () =>
         expect(order).toEqual(["BEGIN", "COMMIT", "SIBLING-OP"]);
       }
     );
+
+    it("inlines an enlisted sibling owner instead of throwing sibling-op", async () => {
+      const handle = {};
+      const ownerA = { table: "table_a" };
+      const ownerB = { table: "table_b" };
+      let enlistedResult: string | undefined;
+
+      await runInTransactionOnConnection(handle, [ownerA, ownerB], async () => {
+        enlistedResult = await runOnConnection(handle, ownerB, async () => "joined");
+      });
+
+      expect(enlistedResult).toBe("joined");
+    });
+
+    it("still throws sibling-op for an owner that was not enlisted", async () => {
+      const handle = {};
+      const ownerA = { table: "table_a" };
+      const ownerB = { table: "table_b" };
+      const ownerC = { table: "table_c" };
+
+      let error: unknown;
+      await runInTransactionOnConnection(handle, [ownerA, ownerB], async () => {
+        try {
+          await runOnConnection(handle, ownerC, async () => "unreachable");
+        } catch (err) {
+          error = err;
+        }
+      });
+
+      expect(error).toBeInstanceOf(ConnectionReentryError);
+      expect((error as ConnectionReentryError).mode).toBe("sibling-op");
+      expect((error as ConnectionReentryError).blockedTable).toBe("table_c");
+    });
   });
 });