From 7acb7eab296940af6ef483c6e4c259d025bf0862 Mon Sep 17 00:00:00 2001 From: Stephen Demjanenko Date: Mon, 20 Jul 2026 09:13:01 -0700 Subject: [PATCH] Add opt-in PostgreSQL advisory lock namespacing by table prefix. Running multiple clusters on one Postgres DB, where each cluster has a different prefix leads to collision of advisory locks. We fix that with an opt-in advisory lock change that keeps locks for separate prefixes separate. --- .../cluster-namespace-advisory-locks.md | 5 ++ .../src/unstable/cluster/SqlRunnerStorage.ts | 37 ++++++-- .../cluster/SqlRunnerStorageNamespace.test.ts | 16 ++++ .../test/cluster/SqlRunnerStorage.test.ts | 85 ++++++++++++++++++- 4 files changed, 137 insertions(+), 6 deletions(-) create mode 100644 .changeset/cluster-namespace-advisory-locks.md create mode 100644 packages/effect/test/cluster/SqlRunnerStorageNamespace.test.ts diff --git a/.changeset/cluster-namespace-advisory-locks.md b/.changeset/cluster-namespace-advisory-locks.md new file mode 100644 index 00000000000..5205b761974 --- /dev/null +++ b/.changeset/cluster-namespace-advisory-locks.md @@ -0,0 +1,5 @@ +--- +"effect": minor +--- + +Add opt-in `namespaceAdvisoryLocks` to `SqlRunnerStorage` so PostgreSQL advisory locks are keyed by table prefix. Enable when multiple clusters share one Postgres database with different prefixes; default lock behavior is unchanged. diff --git a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts index c5787475413..8779232460c 100644 --- a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts @@ -18,6 +18,7 @@ import * as SqlClient from "../sql/SqlClient.ts" import type { SqlError } from "../sql/SqlError.ts" import type * as Statement from "../sql/Statement.ts" import { PersistenceError } from "./ClusterError.ts" +import { hashString } from "./internal/hash.ts" import { ResourceRef } from "./internal/resourceRef.ts" import * as RunnerStorage from "./RunnerStorage.ts" import * as ShardId from "./ShardId.ts" @@ -42,10 +43,17 @@ const withTracerDisabled = Effect.withTracerEnabled(false) * locks unless `ShardingConfig.shardLockDisableAdvisory` is enabled; other * dialects use rows in the locks table. * + * Set `namespaceAdvisoryLocks` when multiple clusters share one PostgreSQL + * database with different `prefix` values. Without it, PostgreSQL advisory + * lock keys ignore `prefix` and clusters can contend for the same shard locks. + * Enabling the flag changes the lock key space — restart the fleet together. + * * **Gotchas** * * Changing `prefix` changes both generated table names, so runners using - * different prefixes do not share registrations or shard locks. + * different prefixes do not share registrations or shard locks. On PostgreSQL, + * table prefix alone does not isolate advisory locks unless + * `namespaceAdvisoryLocks` is enabled. * * @see {@link layer} for the default SQL-backed storage layer * @see {@link layerWith} for a SQL-backed storage layer with a custom table prefix @@ -55,6 +63,7 @@ const withTracerDisabled = Effect.withTracerEnabled(false) */ export const make = Effect.fnUntraced(function*(options: { readonly prefix?: string | undefined + readonly namespaceAdvisoryLocks?: boolean | undefined }) { const config = yield* ShardingConfig.ShardingConfig const shardGroups = ShardingConfig.shardGroupConfig(config) @@ -62,6 +71,8 @@ export const make = Effect.fnUntraced(function*(options: { const disableAdvisoryLocks = config.shardLockDisableAdvisory const sql = (yield* SqlClient.SqlClient).withoutTransforms() const prefix = options?.prefix ?? "cluster" + const namespaceAdvisoryLocks = options?.namespaceAdvisoryLocks === true + const lockNamespace = hashString(prefix) const table = (name: string) => `${prefix}_${name}` const acquireLockConn = sql.onDialectOrElse({ @@ -323,7 +334,9 @@ export const make = Effect.fnUntraced(function*(options: { const acquiredShardIds: Array = [] const toAcquire = new Map(shardIds.map((shardId) => [lockNumbers.get(shardId)!, shardId])) const takenLocks = yield* conn.executeValues( - `SELECT objid FROM pg_locks WHERE locktype = 'advisory' AND granted = true AND pid = ${pid} ORDER BY objid`, + namespaceAdvisoryLocks + ? `SELECT objid FROM pg_locks WHERE locktype = 'advisory' AND granted = true AND pid = ${pid} AND classid = ${lockNamespace} ORDER BY objid` + : `SELECT objid FROM pg_locks WHERE locktype = 'advisory' AND granted = true AND pid = ${pid} ORDER BY objid`, [] ) for (let i = 0; i < takenLocks.length; i++) { @@ -460,7 +473,10 @@ export const make = Effect.fnUntraced(function*(options: { const pgLocks = (shardIdsMap: Map) => Array.from( shardIdsMap.entries(), - ([lockNum, shardId]) => `pg_try_advisory_lock(${lockNum}) AS "${shardId}"` + ([lockNum, shardId]) => + namespaceAdvisoryLocks + ? `pg_try_advisory_lock(${lockNamespace}, ${lockNum}) AS "${shardId}"` + : `pg_try_advisory_lock(${lockNum}) AS "${shardId}"` ).join(", ") const mysqlLocks = (shardIds: ReadonlyArray) => @@ -592,9 +608,16 @@ export const make = Effect.fnUntraced(function*(options: { const lockNum = lockNumbers.get(shardId)! for (let i = 0; i < 5; i++) { const [conn] = yield* lockConn!.await - yield* conn.executeRaw(`SELECT pg_advisory_unlock(${lockNum})`, []) + yield* conn.executeRaw( + namespaceAdvisoryLocks + ? `SELECT pg_advisory_unlock(${lockNamespace}, ${lockNum})` + : `SELECT pg_advisory_unlock(${lockNum})`, + [] + ) const takenLocks = yield* conn.executeValues( - `SELECT 1 FROM pg_locks WHERE locktype = 'advisory' AND granted = true AND pid = pg_backend_pid() AND objid = ${lockNum}`, + namespaceAdvisoryLocks + ? `SELECT 1 FROM pg_locks WHERE locktype = 'advisory' AND granted = true AND pid = pg_backend_pid() AND classid = ${lockNamespace} AND objid = ${lockNum}` + : `SELECT 1 FROM pg_locks WHERE locktype = 'advisory' AND granted = true AND pid = pg_backend_pid() AND objid = ${lockNum}`, [] ) if (takenLocks.length === 0) return @@ -695,10 +718,14 @@ export const layer: Layer.Layer< /** * Layer that provides SQL-backed `RunnerStorage` using a custom table prefix. * + * Pass `namespaceAdvisoryLocks: true` when multiple clusters share one + * PostgreSQL database with different prefixes so advisory locks are isolated. + * * @category layers * @since 4.0.0 */ export const layerWith = (options: { readonly prefix?: string | undefined + readonly namespaceAdvisoryLocks?: boolean | undefined }): Layer.Layer => Layer.effect(RunnerStorage.RunnerStorage)(make(options)) diff --git a/packages/effect/test/cluster/SqlRunnerStorageNamespace.test.ts b/packages/effect/test/cluster/SqlRunnerStorageNamespace.test.ts new file mode 100644 index 00000000000..a474bd825c2 --- /dev/null +++ b/packages/effect/test/cluster/SqlRunnerStorageNamespace.test.ts @@ -0,0 +1,16 @@ +import { assert, describe, it } from "@effect/vitest" +import { hashString } from "../../src/unstable/cluster/internal/hash.ts" + +describe("SqlRunnerStorage advisory lock namespace", () => { + it("derives distinct namespaces from different prefixes", () => { + assert.notStrictEqual(hashString("cluster_a"), hashString("cluster_b")) + assert.strictEqual(hashString("cluster"), hashString("cluster")) + }) + + it("returns a stable int32-compatible value for a given prefix", () => { + const namespace = hashString("cluster") + assert.strictEqual(namespace, hashString("cluster")) + assert.isTrue(Number.isInteger(namespace)) + assert.isTrue(namespace >= -0x80000000 && namespace <= 0x7fffffff) + }) +}) diff --git a/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts b/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts index 913899875e5..712fc52fa20 100644 --- a/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts +++ b/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts @@ -1,6 +1,6 @@ import { NodeFileSystem } from "@effect/platform-node" import { SqliteClient } from "@effect/sql-sqlite-node" -import { describe, expect, it } from "@effect/vitest" +import { assert, describe, expect, it } from "@effect/vitest" import { Effect, FileSystem, Layer } from "effect" import { Runner, @@ -88,9 +88,92 @@ describe("SqlRunnerStorage", () => { })) }) }) + + it.layer( + Layer.merge( + Layer.orDie(PgContainer.layerClient), + ShardingConfig.layer() + ), + { timeout: 60000 } + )("pg namespaced advisory locks", (it) => { + const shards = [ + ShardId.make("default", 1), + ShardId.make("default", 2), + ShardId.make("default", 3) + ] + + it.effect("isolates shard locks across prefixes when enabled", () => + Effect.gen(function*() { + const storageA = yield* SqlRunnerStorage.make({ + prefix: "cluster_a", + namespaceAdvisoryLocks: true + }) + const storageB = yield* SqlRunnerStorage.make({ + prefix: "cluster_b", + namespaceAdvisoryLocks: true + }) + + const acquiredA = yield* storageA.acquire(runnerAddress1, shards) + const acquiredB = yield* storageB.acquire(runnerAddress2, shards) + assert.deepStrictEqual(acquiredA.map((shard) => shard.id), [1, 2, 3]) + assert.deepStrictEqual(acquiredB.map((shard) => shard.id), [1, 2, 3]) + }).pipe(Effect.scoped)) + + it.effect("keeps locks exclusive for the same prefix when enabled", () => + Effect.gen(function*() { + const storageA = yield* SqlRunnerStorage.make({ + prefix: "shared", + namespaceAdvisoryLocks: true + }) + const storageB = yield* SqlRunnerStorage.make({ + prefix: "shared", + namespaceAdvisoryLocks: true + }) + + const acquiredA = yield* storageA.acquire(runnerAddress1, shards) + const acquiredB = yield* storageB.acquire(runnerAddress2, shards) + assert.deepStrictEqual(acquiredA.map((shard) => shard.id), [1, 2, 3]) + assert.deepStrictEqual(acquiredB, []) + }).pipe(Effect.scoped)) + + it.effect("releases namespaced locks so another runner can acquire", () => + Effect.gen(function*() { + const storageA = yield* SqlRunnerStorage.make({ + prefix: "release_ns", + namespaceAdvisoryLocks: true + }) + const storageB = yield* SqlRunnerStorage.make({ + prefix: "release_ns", + namespaceAdvisoryLocks: true + }) + + assert.deepStrictEqual( + (yield* storageA.acquire(runnerAddress1, shards)).map((shard) => shard.id), + [1, 2, 3] + ) + yield* storageA.release(runnerAddress1, ShardId.make("default", 2)) + + const reacquired = yield* storageB.acquire(runnerAddress2, [ShardId.make("default", 2)]) + assert.deepStrictEqual(reacquired.map((shard) => shard.id), [2]) + }).pipe(Effect.scoped)) + + it.effect("without the flag, different prefixes still share advisory locks", () => + Effect.gen(function*() { + const storageA = yield* SqlRunnerStorage.make({ prefix: "legacy_a" }) + const storageB = yield* SqlRunnerStorage.make({ prefix: "legacy_b" }) + + const acquiredA = yield* storageA.acquire(runnerAddress1, shards) + const acquiredB = yield* storageB.acquire(runnerAddress2, shards) + assert.deepStrictEqual(acquiredA.map((shard) => shard.id), [1, 2, 3]) + // Documents the pre-fix collision: prefix-only isolation does not + // apply to PostgreSQL advisory lock keys unless namespacing is enabled. + assert.deepStrictEqual(acquiredB, []) + }).pipe(Effect.scoped)) + }) }) const runnerAddress1 = RunnerAddress.make("localhost", 1234) +const runnerAddress2 = RunnerAddress.make("localhost", 1235) const SqliteLayer = Effect.gen(function*() { const fs = yield* FileSystem.FileSystem