diff --git a/cspell.config.jsonc b/cspell.config.jsonc index 1626af07..225688c8 100644 --- a/cspell.config.jsonc +++ b/cspell.config.jsonc @@ -33,6 +33,7 @@ "mydb", "sslmode", "timestamptz", + "xact", // project vocabulary "checkin", diff --git a/packages/openworkflow/postgres/backend.test.ts b/packages/openworkflow/postgres/backend.test.ts index 636f28fe..b037ecb1 100644 --- a/packages/openworkflow/postgres/backend.test.ts +++ b/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("-", "_")}`; diff --git a/packages/openworkflow/postgres/backend.ts b/packages/openworkflow/postgres/backend.ts index 42b05917..8fe0ef24 100644 --- a/packages/openworkflow/postgres/backend.ts +++ b/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 => { + 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(