fix(replication): key logical replication leader lock on slot name

The LogicalReplicationClient leader lock (Redlock) was keyed on the client
name, but a Postgres logical replication slot allows exactly one consumer, so
the lock must serialize consumers of a given slot. When two clients target the
same slot with different names (e.g. across a rolling deploy where the name
changed but the slot did not), each acquired a distinct lock, both became
leader, and the second to run START_REPLICATION failed with 'replication slot
is active'. Since START_REPLICATION is fire-and-forget and only logged on error,
that consumer stopped and replication stalled until restarted.

Key the lock on slotName instead. Adds a regression test (two clients, same
slot, different name) verified to fail before and pass after.

Author: d-cs

.server-changes/fix-replication-leader-lock-per-slot.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Key the logical-replication leader lock on the slot name (not the client name) so consumers of the same replication slot serialize correctly across restarts and rolling deploys

internal-packages/replication/src/client.test.ts

@@ -181,4 +181,58 @@ describe("Replication Client", () => {
       expect(slotExists[0].exists).toBe(false);
     }
   );
+
+  postgresAndRedisTest(
+    "two clients on the same slot must not both lead (rolling-deploy handoff)",
+    async ({ postgresContainer, prisma, redisOptions }) => {
+      await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
+
+      const shared = {
+        slotName: "handoff_slot",
+        publicationName: "handoff_publication",
+        redisOptions,
+        table: "TaskRun",
+        pgConfig: { connectionString: postgresContainer.getConnectionUri() },
+      };
+
+      // Leader on the shared slot.
+      const a = new LogicalReplicationClient({ ...shared, name: "runs-replication" });
+      const aElections: boolean[] = [];
+      a.events.on("leaderElection", (won) => aElections.push(won));
+      a.events.on("error", () => {});
+      await a.subscribe();
+      // Let A's walsender actually attach to the slot before B races it.
+      await setTimeout(1000);
+
+      // Second client, SAME slot, DIFFERENT name — the rolling-deploy shape that
+      // regressed (name changed "runs-replication" -> "runs-replication:legacy").
+      const b = new LogicalReplicationClient({
+        ...shared,
+        name: "runs-replication:legacy",
+        leaderLockTimeoutMs: 1000,
+        leaderLockAcquireAdditionalTimeMs: 250,
+        leaderLockRetryIntervalMs: 200,
+      });
+      const bElections: boolean[] = [];
+      const bErrors: Array<unknown> = [];
+      b.events.on("leaderElection", (won) => bElections.push(won));
+      b.events.on("error", (error) => bErrors.push(error));
+      await b.subscribe();
+      await setTimeout(500);
+
+      expect(aElections).toContain(true);
+      // B must not also win leadership on the same slot, nor race START_REPLICATION
+      // into a "slot is active" error. With a name-keyed lock it did both.
+      expect(bElections).not.toContain(true);
+      expect(bElections).toContain(false);
+      expect(
+        bErrors
+          .map((e) => String((e as Error)?.message ?? e))
+          .some((m) => /replication slot .* is active|already active/i.test(m))
+      ).toBe(false);
+
+      await a.stop();
+      await b.stop();
+    }
+  );
 });

internal-packages/replication/src/client.ts

@@ -19,7 +19,8 @@ export interface LogicalReplicationClientOptions {
   pgConfig: ClientConfig;
 
   /**
-   * The name of this LogicalReplicationClient instance, used for leader election.
+   * The name of this LogicalReplicationClient instance, used for logging and the
+   * Postgres application_name. Leader election is keyed on `slotName`.
    */
   name: string;
   /**
@@ -703,8 +704,11 @@ export class LogicalReplicationClient {
 
     while (Date.now() - startTime < maxWaitTime) {
       try {
+        // Key the leader lock on the SLOT, not `name`: Postgres allows one
+        // consumer per slot, so consumers of the same slot must contend on the
+        // same lock (a name-keyed lock lets old+new pods race it across a deploy).
         this.leaderLock = await this.redlock.acquire(
-          [`logical-replication-client:${this.options.name}`],
+          [`logical-replication-client:${this.options.slotName}`],
           this.leaderLockTimeoutMs
         );