Add opt-in PostgreSQL advisory lock namespacing by table prefix.

.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.

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,13 +63,16 @@ 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)
   const availableShardGroups = Array.from(shardGroups.available)
   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<string> = []
         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<number, string>) =>
     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<string>) =>
@@ -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<RunnerStorage.RunnerStorage, SqlError, SqlClient.SqlClient | ShardingConfig.ShardingConfig> =>
   Layer.effect(RunnerStorage.RunnerStorage)(make(options))

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)
+  })
+})

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