feat(openworkflow): support transaction-scoped advisory locks

cspell.config.jsonc

@@ -33,6 +33,7 @@
     "mydb",
     "sslmode",
     "timestamptz",
+    "xact",
 
     // project vocabulary
     "checkin",

packages/openworkflow/postgres/backend.test.ts

@@ -120,6 +120,45 @@ describe("BackendPostgres.fromPool", () => {
   });
 });
 
+describe("BackendPostgres idempotency advisory locks", () => {
+  test("uses a transaction-scoped advisory lock", async () => {
+    const queries: string[] = [];
+    const pg = newPostgresMaxOne(DEFAULT_POSTGRES_URL, {
+      debug: (_connection, query) => {
+        queries.push(query);
+      },
+    });
+    const backend = BackendPostgres.fromPool(pg, {
+      namespaceId: randomUUID(),
+    });
+
+    try {
+      await backend.createWorkflowRun({
+        workflowName: randomUUID(),
+        version: null,
+        idempotencyKey: randomUUID(),
+        input: null,
+        config: {},
+        context: null,
+        parentStepAttemptNamespaceId: null,
+        parentStepAttemptId: null,
+        availableAt: null,
+        deadlineAt: null,
+      });
+
+      expect(queries).toContain("begin isolation level read committed");
+      expect(
+        queries.some((query) => query.includes("pg_advisory_xact_lock")),
+      ).toBe(true);
+      expect(
+        queries.some((query) => query.includes("pg_advisory_unlock")),
+      ).toBe(false);
+    } finally {
+      await pg.end();
+    }
+  });
+});
+
 describe("BackendPostgres schema option", () => {
   test("stores workflow data in the configured schema", async () => {
     const schema = `test_schema_${randomUUID().replaceAll("-", "_")}`;

packages/openworkflow/postgres/backend.ts

@@ -147,19 +147,18 @@ export class BackendPostgres implements Backend {
       idempotencyKey,
     });
 
-    const pgReserved = (await this.pg.reserve()) as unknown as Postgres & {
-      release: () => void;
-    };
-
-    try {
-      await pgReserved.unsafe(
-        "SELECT pg_advisory_lock(hashtextextended($1, 0::bigint))",
-        [lockScope],
-      );
+    return await this.pg.begin(
+      "isolation level read committed",
+      async (sql): Promise<WorkflowRun> => {
+        const tx = sql as unknown as Postgres;
+
+        await tx.unsafe(
+          "SELECT pg_advisory_xact_lock(hashtextextended($1, 0::bigint))",
+          [lockScope],
+        );
 
-      try {
         const existing = await this.getWorkflowRunByIdempotencyKey(
-          pgReserved,
+          tx,
           workflowName,
           idempotencyKey,
           new Date(Date.now() - DEFAULT_RUN_IDEMPOTENCY_PERIOD_MS),
@@ -168,20 +167,9 @@ export class BackendPostgres implements Backend {
           return existing;
         }
 
-        return await this.insertWorkflowRun(pgReserved, params);
-      } finally {
-        await pgReserved
-          .unsafe(
-            "SELECT pg_advisory_unlock(hashtextextended($1, 0::bigint))",
-            [lockScope],
-          )
-          .catch(() => {
-            // best effort unlock; session close also releases session advisory locks
-          });
-      }
-    } finally {
-      pgReserved.release();
-    }
+        return await this.insertWorkflowRun(tx, params);
+      },
+    );
   }
 
   private async insertWorkflowRun(