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

## Problem

`LogicalReplicationClient` uses a Redlock leader lock to guarantee a
single active consumer per Postgres logical replication slot. The lock
resource was keyed on the client `name`:

```
logical-replication-client:${this.options.name}
```

A slot permits exactly one consumer, so the lock's job is to serialize
consumers **of a given slot**. Keying it on `name` breaks that whenever
two clients target the same slot with different names — most notably
across a rolling deploy where the client `name` changes but `slotName`
does not. Both acquire *distinct* locks, both consider themselves
leader, and the second to reach `START_REPLICATION` hits `replication
slot "<slot>" is active for PID <n>`. Because that query was
fire-and-forget and its failure was only logged (no retry), the consumer
stopped and replication stalled until the process was restarted.

## Fix

**1. Key the leader lock on `slotName`** — the actual single-consumer
resource:

```
logical-replication-client:${this.options.slotName}
```

Consumers of the same slot now contend on the same lock and hand off
cleanly across restarts/deploys; different slots stay independent.
`name` is kept for logging and the pg `application_name`.

**2. Self-healing resubscribe** (`resubscribeOnFailure`, opt-in) —
instead of logging-and-dying, a client re-subscribes with exponential
backoff after a lost election or a failed `START_REPLICATION`, so a
rolling deploy self-heals: the incoming pod retries until the draining
pod releases the slot, then takes over. Safety:
- `#cleanupAttempt()` unconditionally ends the pg client (freeing the
walsender) and releases the leader lock before rescheduling — retries
never leak connections/locks.
- `shutdown()` sets an intentional-stop latch re-checked after every
`await` in `subscribe()` (and aborts the lock-acquire spin), so a
resubscribe can never race or outlive an intentional shutdown.
- Backoff resets only on genuine stream start, so a permanently stuck
slot backs off to the ceiling and logs loudly rather than tight-looping;
an epoch guard neutralises stale `START_REPLICATION` catches.

Runs- and sessions-replication opt in and use `shutdown()` for all
intentional stops.

**3. Observability** — the admin runs-replication status route probed
the old name-keyed Redis key (would report `leader:false` for every
source after fix #1); now probes the slot-keyed key.

## Tests

`internal-packages/replication/src/client.test.ts` (real Postgres +
Redis containers):
- same-slot/different-name → second client must not double-lead or race
into "slot is active" (the regression)
- a failing `START_REPLICATION` retry loop must not leak connections or
locks
- `shutdown()` during an in-flight `subscribe()` must not leave a zombie
leader
- `subscribe()` after `shutdown()` re-arms `resubscribeOnFailure`
- self-heals once the leader releases the slot

Plus the multi-source wiring test updated to the slot-keyed lock keys.

## Rollout

With the self-healing resubscribe, this ships as a **plain rolling
deploy** — the incoming pods retry across the one-time lock-key
transition and take over once the old pods drain (a brief replication
stall that the durable slot replays on reconnect — no data loss). No
stop-before-start required.

.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

apps/webapp/app/routes/admin.api.v1.runs-replication.status.ts

@@ -7,12 +7,15 @@ import { getRunsReplicationConfiguredSources } from "~/services/runsReplicationG
 /**
  * Probes per-source replication leadership via the redlock leader-lock key, which
  * is DOUBLE-PREFIXED with `logical-replication-client:` — once from the connection's
- * keyPrefix and once from redlock's resource string. So we prefix this connection
- * with `runs-replication:logical-replication-client:` and EXISTS on the resource
- * `logical-replication-client:runs-replication:<id>`, resolving to:
- *   runs-replication:logical-replication-client:logical-replication-client:runs-replication:<id>
+ * keyPrefix and once from redlock's resource string. The lock is keyed on the
+ * replication slot, so we prefix this connection with
+ * `runs-replication:logical-replication-client:` and EXISTS on the resource
+ * `logical-replication-client:<slotName>`, resolving to:
+ *   runs-replication:logical-replication-client:logical-replication-client:<slotName>
  */
-async function probeLeadership(sourceIds: string[]): Promise<Map<string, boolean>> {
+async function probeLeadership(
+  sources: { id: string; slotName: string }[]
+): Promise<Map<string, boolean>> {
   const leaders = new Map<string, boolean>();
 
   const redis = new Redis({
@@ -26,9 +29,9 @@ async function probeLeadership(sourceIds: string[]): Promise<Map<string, boolean
   });
 
   try {
-    for (const id of sourceIds) {
-      const exists = await redis.exists(`logical-replication-client:runs-replication:${id}`);
-      leaders.set(id, exists === 1);
+    for (const source of sources) {
+      const exists = await redis.exists(`logical-replication-client:${source.slotName}`);
+      leaders.set(source.id, exists === 1);
     }
   } finally {
     await redis.quit();
@@ -46,7 +49,7 @@ export async function loader({ request }: LoaderFunctionArgs) {
     return json({ enabled: false, sources: [] });
   }
 
-  const leaders = await probeLeadership(sources.map((s) => s.id));
+  const leaders = await probeLeadership(sources);
 
   return json({
     enabled: env.RUN_REPLICATION_ENABLED === "1" && sources.length > 0,

apps/webapp/app/services/runsReplicationService.server.ts

@@ -286,6 +286,7 @@ export class RunsReplicationService {
         table: "TaskRun",
         redisOptions: options.redisOptions,
         autoAcknowledge: false,
+        resubscribeOnFailure: true,
         publicationActions: ["insert", "update", "delete"],
         logger:
           options.logger ?? new Logger("LogicalReplicationClient", options.logLevel ?? "info"),
@@ -428,7 +429,9 @@ export class RunsReplicationService {
 
     if (!hasCurrentTransaction) {
       this.logger.info("No transaction to commit, shutting down immediately");
-      await Promise.all(Array.from(this._sources.values()).map((runtime) => runtime.client.stop()));
+      await Promise.all(
+        Array.from(this._sources.values()).map((runtime) => runtime.client.shutdown())
+      );
       this._isShutDownComplete = true;
       return;
     }
@@ -458,7 +461,7 @@ export class RunsReplicationService {
     for (const runtime of this._sources.values()) {
       this.logger.info("Stopping replication client", { sourceId: runtime.source.id });
 
-      await runtime.client.stop();
+      await runtime.client.shutdown();
 
       if (runtime.acknowledgeInterval) {
         clearInterval(runtime.acknowledgeInterval);
@@ -636,7 +639,7 @@ export class RunsReplicationService {
       // swallow client.stop() rejections so they don't surface as unhandled.
       if (!this._shutdownStopInFlight) {
         this._shutdownStopInFlight = true;
-        Promise.all(Array.from(this._sources.values()).map((r) => r.client.stop()))
+        Promise.all(Array.from(this._sources.values()).map((r) => r.client.shutdown()))
           .catch((error) => {
             this.logger.error("Error stopping replication clients during shutdown", { error });
           })

apps/webapp/app/services/sessionsReplicationService.server.ts

@@ -187,6 +187,7 @@ export class SessionsReplicationService {
       table: "Session",
       redisOptions: options.redisOptions,
       autoAcknowledge: false,
+      resubscribeOnFailure: true,
       publicationActions: ["insert", "update", "delete"],
       logger: options.logger ?? new Logger("LogicalReplicationClient", options.logLevel ?? "info"),
       leaderLockTimeoutMs: options.leaderLockTimeoutMs ?? 30_000,
@@ -265,7 +266,7 @@ export class SessionsReplicationService {
 
     if (!this._currentTransaction) {
       this.logger.info("No transaction to commit, shutting down immediately");
-      await this._replicationClient.stop();
+      await this._replicationClient.shutdown();
       this._isSubscribed = false;
       this._isShutDownComplete = true;
       return;
@@ -294,7 +295,7 @@ export class SessionsReplicationService {
   async stop() {
     this.logger.info("Stopping replication client");
 
-    await this._replicationClient.stop();
+    await this._replicationClient.shutdown();
 
     if (this._acknowledgeInterval) {
       clearInterval(this._acknowledgeInterval);
@@ -430,10 +431,15 @@ export class SessionsReplicationService {
     if (this._isShutDownComplete) return;
 
     if (this._isShuttingDown) {
-      this._replicationClient.stop().finally(() => {
-        this._isSubscribed = false;
-        this._isShutDownComplete = true;
-      });
+      this._replicationClient
+        .shutdown()
+        .catch((error) => {
+          this.logger.error("Error stopping replication client during shutdown", { error });
+        })
+        .finally(() => {
+          this._isSubscribed = false;
+          this._isShutDownComplete = true;
+        });
     }
 
     // If there are no events, do nothing

apps/webapp/test/runsReplicationInstance.test.ts

@@ -408,10 +408,12 @@ describe("RunsReplication multi-source wiring (integration)", () => {
 
         probe = new Redis(redisOptions);
 
+        // Leader lock is keyed on the slot, so each source holds a distinct
+        // slot-keyed lock (double-prefixed: connection keyPrefix + redlock resource).
         const legacyKey =
-          "runs-replication:logical-replication-client:logical-replication-client:runs-replication:legacy";
+          "runs-replication:logical-replication-client:logical-replication-client:tr_legacy_wiring";
         const newKey =
-          "runs-replication:logical-replication-client:logical-replication-client:runs-replication:new";
+          "runs-replication:logical-replication-client:logical-replication-client:tr_new_wiring";
 
         expect(await probe.exists(legacyKey)).toBe(1);
         expect(await probe.exists(newKey)).toBe(1);

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

@@ -181,4 +181,288 @@ 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();
+    }
+  );
+
+  postgresAndRedisTest(
+    "resubscribeOnFailure self-heals once the leader releases the slot",
+    async ({ postgresContainer, prisma, redisOptions }) => {
+      await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
+
+      const shared = {
+        slotName: "resub_slot",
+        publicationName: "resub_pub",
+        redisOptions,
+        table: "TaskRun",
+        pgConfig: { connectionString: postgresContainer.getConnectionUri() },
+      };
+
+      // Leader holds the slot.
+      const a = new LogicalReplicationClient({ ...shared, name: "leader-a" });
+      a.events.on("error", () => {});
+      await a.subscribe();
+      await setTimeout(1000);
+
+      // Contender with resubscribe on: loses the election while A holds the slot,
+      // then must self-heal (win) once A releases it — the rolling-deploy handoff.
+      const b = new LogicalReplicationClient({
+        ...shared,
+        name: "contender-b",
+        resubscribeOnFailure: true,
+        resubscribeMinDelayMs: 200,
+        resubscribeMaxDelayMs: 400,
+        leaderLockTimeoutMs: 500,
+        leaderLockAcquireAdditionalTimeMs: 100,
+        leaderLockRetryIntervalMs: 100,
+      });
+      const bElections: boolean[] = [];
+      b.events.on("leaderElection", (won) => bElections.push(won));
+      b.events.on("error", () => {});
+      await b.subscribe();
+      await setTimeout(1500);
+
+      // Still contending, not leader, while A holds the slot.
+      expect(bElections).toContain(false);
+      expect(bElections).not.toContain(true);
+
+      // Release the leader — a scheduled resubscribe should now win.
+      await a.shutdown();
+
+      let becameLeader = false;
+      for (let i = 0; i < 40; i++) {
+        if (bElections.includes(true)) {
+          becameLeader = true;
+          break;
+        }
+        await setTimeout(250);
+      }
+      expect(becameLeader).toBe(true);
+
+      await b.shutdown();
+    }
+  );
+
+  postgresAndRedisTest(
+    "a failing START_REPLICATION retry loop must not leak connections or locks",
+    async ({ postgresContainer, prisma, redisOptions }) => {
+      await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
+
+      const shared = {
+        slotName: "leak_slot",
+        publicationName: "leak_pub",
+        table: "TaskRun",
+        pgConfig: { connectionString: postgresContainer.getConnectionUri() },
+      };
+
+      const a = new LogicalReplicationClient({ ...shared, redisOptions, name: "leak-leader" });
+      a.events.on("error", () => {});
+      await a.subscribe();
+      await setTimeout(1000);
+
+      // B elects on a separate lock namespace so every attempt reaches
+      // START_REPLICATION and dies there ("slot is active") — the stuck-slot shape.
+      const b = new LogicalReplicationClient({
+        ...shared,
+        redisOptions: { ...redisOptions, keyPrefix: `${redisOptions.keyPrefix ?? ""}other:` },
+        name: "leak-contender",
+        resubscribeOnFailure: true,
+        resubscribeMinDelayMs: 200,
+        resubscribeMaxDelayMs: 400,
+        leaderLockTimeoutMs: 1000,
+        leaderLockAcquireAdditionalTimeMs: 300,
+        leaderLockRetryIntervalMs: 100,
+      });
+      const bErrors: Array<unknown> = [];
+      b.events.on("error", (error) => bErrors.push(error));
+      await b.subscribe();
+
+      for (let i = 0; i < 80 && bErrors.length < 3; i++) {
+        await setTimeout(250);
+      }
+      expect(bErrors.length).toBeGreaterThanOrEqual(3);
+
+      // Every failed attempt must end its pg client: at most the one in-flight
+      // attempt's backend may exist, never an accrual across cycles.
+      const backends = await prisma.$queryRaw<{ count: bigint }[]>`
+        SELECT count(*) AS count FROM pg_stat_activity WHERE application_name = 'leak-contender'
+      `;
+      expect(Number(backends[0].count)).toBeLessThanOrEqual(1);
+
+      const active = await prisma.$queryRaw<{ count: bigint }[]>`
+        SELECT count(*) AS count FROM pg_replication_slots WHERE slot_name = 'leak_slot' AND active
+      `;
+      expect(Number(active[0].count)).toBe(1);
+
+      await b.shutdown();
+      await a.shutdown();
+    }
+  );
+
+  postgresAndRedisTest(
+    "shutdown during an in-flight subscribe must not leave a zombie leader",
+    async ({ postgresContainer, prisma, redisOptions }) => {
+      await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
+
+      const shared = {
+        slotName: "zombie_slot",
+        publicationName: "zombie_pub",
+        redisOptions,
+        table: "TaskRun",
+        pgConfig: { connectionString: postgresContainer.getConnectionUri() },
+      };
+
+      const a = new LogicalReplicationClient({ ...shared, name: "zombie-leader" });
+      a.events.on("error", () => {});
+      await a.subscribe();
+      await setTimeout(1000);
+
+      // B's election spins against A's held lock; shut it down mid-subscribe.
+      const b = new LogicalReplicationClient({
+        ...shared,
+        name: "zombie-contender",
+        resubscribeOnFailure: true,
+        leaderLockTimeoutMs: 5000,
+        leaderLockAcquireAdditionalTimeMs: 5000,
+        leaderLockRetryIntervalMs: 100,
+      });
+      const bElections: boolean[] = [];
+      b.events.on("leaderElection", (won) => bElections.push(won));
+      b.events.on("error", () => {});
+
+      const inflight = b.subscribe();
+      await setTimeout(300);
+      await b.shutdown();
+
+      // Release the real leader; a zombie B would now win the lock and the slot.
+      await a.shutdown();
+      await inflight.catch(() => {});
+      await setTimeout(1500);
+
+      const zombieWon = bElections.includes(true);
+      const active = await prisma.$queryRaw<{ count: bigint }[]>`
+        SELECT count(*) AS count FROM pg_replication_slots WHERE slot_name = 'zombie_slot' AND active
+      `;
+      const backends = await prisma.$queryRaw<{ count: bigint }[]>`
+        SELECT count(*) AS count FROM pg_stat_activity WHERE application_name = 'zombie-contender'
+      `;
+      // Reap a zombie (if any) so the test exits cleanly, then assert.
+      await b.shutdown();
+
+      expect(zombieWon).toBe(false);
+      expect(Number(active[0].count)).toBe(0);
+      expect(Number(backends[0].count)).toBe(0);
+    }
+  );
+
+  postgresAndRedisTest(
+    "subscribe after shutdown re-arms resubscribeOnFailure",
+    async ({ postgresContainer, prisma, redisOptions }) => {
+      await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
+
+      const shared = {
+        slotName: "rearm_slot",
+        publicationName: "rearm_pub",
+        redisOptions,
+        table: "TaskRun",
+        pgConfig: { connectionString: postgresContainer.getConnectionUri() },
+      };
+
+      const b = new LogicalReplicationClient({
+        ...shared,
+        name: "rearm-client",
+        resubscribeOnFailure: true,
+        resubscribeMinDelayMs: 200,
+        resubscribeMaxDelayMs: 400,
+        leaderLockTimeoutMs: 500,
+        leaderLockAcquireAdditionalTimeMs: 100,
+        leaderLockRetryIntervalMs: 100,
+      });
+      const bElections: boolean[] = [];
+      b.events.on("leaderElection", (won) => bElections.push(won));
+      b.events.on("error", () => {});
+
+      // Admin stop -> start: shutdown latches the intentional stop...
+      await b.subscribe();
+      await setTimeout(500);
+      await b.shutdown();
+
+      const a = new LogicalReplicationClient({ ...shared, name: "rearm-leader" });
+      a.events.on("error", () => {});
+      await a.subscribe();
+      await setTimeout(1000);
+
+      // ...then an explicit re-subscribe loses the election and must self-heal
+      // once the leader goes away (self-heal re-armed by the subscribe).
+      bElections.length = 0;
+      await b.subscribe();
+      expect(bElections).toContain(false);
+      expect(bElections).not.toContain(true);
+
+      await a.shutdown();
+
+      let becameLeader = false;
+      for (let i = 0; i < 40; i++) {
+        if (bElections.includes(true)) {
+          becameLeader = true;
+          break;
+        }
+        await setTimeout(250);
+      }
+      expect(becameLeader).toBe(true);
+
+      await b.shutdown();
+    }
+  );
 });

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;
   /**
@@ -65,6 +66,17 @@ export interface LogicalReplicationClientOptions {
    */
   leaderLockAcquireAdditionalTimeMs?: number;
 
+  /**
+   * Auto re-subscribe (with backoff) after a lost election / failed
+   * START_REPLICATION instead of stopping. Off by default; when on, use
+   * `shutdown()` (not `stop()`) for intentional shutdown.
+   */
+  resubscribeOnFailure?: boolean;
+  /** Base delay for the resubscribe backoff (default: 1000ms). */
+  resubscribeMinDelayMs?: number;
+  /** Max delay for the resubscribe backoff (default: 30000ms). */
+  resubscribeMaxDelayMs?: number;
+
   /**
    * The interval in seconds to automatically acknowledge the last LSN if no ack has been sent (default: 10)
    */
@@ -108,6 +120,13 @@ export class LogicalReplicationClient {
   private ackIntervalTimer: NodeJS.Timeout | null = null;
   private _isStopped: boolean = false;
   private _tracer: Tracer;
+  private resubscribeOnFailure: boolean;
+  private resubscribeMinDelayMs: number;
+  private resubscribeMaxDelayMs: number;
+  private resubscribeTimer: NodeJS.Timeout | null = null;
+  private resubscribeAttempts: number = 0;
+  private _intentionalStop: boolean = false;
+  private subscribeEpoch: number = 0;
 
   public get lastLsn(): string {
     return this.lastAcknowledgedLsn ?? "0/00000000";
@@ -130,6 +149,9 @@ export class LogicalReplicationClient {
     this.leaderLockAcquireAdditionalTimeMs = options.leaderLockAcquireAdditionalTimeMs ?? 1000;
     this.leaderLockRetryIntervalMs = options.leaderLockRetryIntervalMs ?? 500;
     this.ackIntervalSeconds = options.ackIntervalSeconds ?? 10;
+    this.resubscribeOnFailure = options.resubscribeOnFailure ?? false;
+    this.resubscribeMinDelayMs = options.resubscribeMinDelayMs ?? 1000;
+    this.resubscribeMaxDelayMs = options.resubscribeMaxDelayMs ?? 30000;
 
     this.redis = createRedisClient(
       {
@@ -154,6 +176,11 @@ export class LogicalReplicationClient {
 
   public async stop(): Promise<this> {
     return await startSpan(this._tracer, "logical_replication_client.stop", async (span) => {
+      if (this.resubscribeTimer) {
+        clearTimeout(this.resubscribeTimer);
+        this.resubscribeTimer = null;
+      }
+
       if (this._isStopped) return this;
 
       span.setAttribute("replication_client.name", this.options.name);
@@ -211,37 +238,135 @@ export class LogicalReplicationClient {
     });
   }
 
-  public async teardown(): Promise<boolean> {
-    await this.stop();
+  /**
+   * Permanently stop the client and disable auto-resubscribe. Use this (not
+   * stop()) for intentional shutdown so a failure-triggered resubscribe can't
+   * race it.
+   */
+  public async shutdown(): Promise<this> {
+    this._intentionalStop = true;
+    return this.stop();
+  }
 
-    // Acquire the leaderLock
-    const leaderLockAcquired = await this.#acquireLeaderLock();
+  /**
+   * Unconditionally release the current attempt's timers, pg client and leader
+   * lock. Unlike stop() this doesn't no-op when `_isStopped` is set — a failed
+   * subscribe runs entirely in that state and would otherwise leak them.
+   */
+  async #cleanupAttempt(): Promise<void> {
+    this._isStopped = true;
+
+    if (this.leaderLockHeartbeatTimer) {
+      clearInterval(this.leaderLockHeartbeatTimer);
+      this.leaderLockHeartbeatTimer = null;
+    }
+
+    if (this.ackIntervalTimer) {
+      clearInterval(this.ackIntervalTimer);
+      this.ackIntervalTimer = null;
+    }
+
+    this.connection?.removeAllListeners();
+    this.connection = null;
+
+    if (this.client) {
+      this.client.removeAllListeners();
+
+      const [endError] = await tryCatch(this.client.end());
+
+      if (endError) {
+        this.logger.error("Failed to end client", {
+          name: this.options.name,
+          error: endError,
+        });
+      }
+      this.client = null;
+    }
+
+    await this.#releaseLeaderLock();
+  }
+
+  #scheduleResubscribe(reason: string): void {
+    if (!this.resubscribeOnFailure || this._intentionalStop) return;
+    if (this.resubscribeTimer) return;
+
+    const delay = Math.min(
+      this.resubscribeMinDelayMs * 2 ** this.resubscribeAttempts,
+      this.resubscribeMaxDelayMs
+    );
+    this.resubscribeAttempts += 1;
+
+    const payload = {
+      name: this.options.name,
+      slotName: this.options.slotName,
+      reason,
+      attempt: this.resubscribeAttempts,
+      delayMs: delay,
+    };
+    // At the ceiling the stream isn't recovering — log loudly so a genuinely
+    // stuck slot surfaces instead of hiding behind silent retries.
+    if (delay >= this.resubscribeMaxDelayMs) {
+      this.logger.error("Replication resubscribe scheduled (at max backoff)", payload);
+    } else {
+      this.logger.warn("Replication resubscribe scheduled", payload);
+    }
+
+    this.resubscribeTimer = setTimeout(() => {
+      this.resubscribeTimer = null;
+      if (this._intentionalStop) return;
+      this.subscribe(this.lastAcknowledgedLsn ?? undefined).catch((error) => {
+        this.logger.error("Replication resubscribe attempt failed", {
+          name: this.options.name,
+          error,
+        });
+        this.#scheduleResubscribe("resubscribe-threw");
+      });
+    }, delay);
+  }
+
+  public async teardown(): Promise<boolean> {
+    this._intentionalStop = true;
+    this.subscribeEpoch += 1;
+    await this.stop();
+    await this.#cleanupAttempt();
+
+    // Acquire the leaderLock (teardown itself is an intentional stop)
+    const leaderLockAcquired = await this.#acquireLeaderLock(false);
 
     if (!leaderLockAcquired) {
       return false;
     }
 
-    this.client = new Client({
-      ...this.options.pgConfig,
-      // @ts-expect-error
-      replication: "database",
-      application_name: this.options.name,
-    });
-    await this.client.connect();
+    try {
+      this.client = new Client({
+        ...this.options.pgConfig,
+        // @ts-expect-error
+        replication: "database",
+        application_name: this.options.name,
+      });
+      await this.client.connect();
 
-    // Drop the slot
-    const slotDropped = await this.#dropSlot();
-
-    await this.client.end();
-    this.client = null;
-
-    await this.#releaseLeaderLock();
-
-    return slotDropped;
+      // Drop the slot
+      return await this.#dropSlot();
+    } finally {
+      // Release the client + slot-keyed lock on both success and throw, so a
+      // mid-teardown failure can't strand the lock (blocking the slot's leader).
+      if (this.client) {
+        await tryCatch(this.client.end());
+        this.client = null;
+      }
+      await this.#releaseLeaderLock();
+    }
   }
 
   public async subscribe(startLsn?: string): Promise<this> {
+    // An explicit subscribe is intent to run: re-arm self-heal after shutdown().
+    this._intentionalStop = false;
+    const attemptEpoch = ++this.subscribeEpoch;
+
     await this.stop();
+    // stop() no-ops once stopped; a failed attempt can leave a client/lock behind.
+    await this.#cleanupAttempt();
 
     this.lastAcknowledgedLsn = startLsn ?? this.lastAcknowledgedLsn;
 
@@ -256,9 +381,16 @@ export class LogicalReplicationClient {
     // 1. Leader election
     const leaderLockAcquired = await this.#acquireLeaderLock();
 
+    if (this._intentionalStop) {
+      await this.#cleanupAttempt();
+      return this;
+    }
+
     if (!leaderLockAcquired) {
       this.events.emit("leaderElection", false);
-      return this.stop();
+      await this.#cleanupAttempt();
+      this.#scheduleResubscribe("leader-election-failed");
+      return this;
     }
 
     this.events.emit("leaderElection", true);
@@ -277,135 +409,175 @@ export class LogicalReplicationClient {
     // Start auto-acknowledge interval
     this.#startAckInterval();
 
-    // 2. Connect pg client
-    this.client = new Client({
-      ...this.options.pgConfig,
-      // @ts-expect-error
-      replication: "database",
-      application_name: this.options.name,
-    });
-    await this.client.connect();
-    // @ts-ignore
-    this.connection = this.client.connection;
+    try {
+      // 2. Connect pg client
+      this.client = new Client({
+        ...this.options.pgConfig,
+        // @ts-expect-error
+        replication: "database",
+        application_name: this.options.name,
+      });
+      await this.client.connect();
+      // @ts-ignore
+      this.connection = this.client.connection;
 
-    const publicationCreated = await this.#createPublication();
-
-    if (!publicationCreated) {
-      return this.stop();
-    }
-
-    this.logger.info("Publication created", {
-      name: this.options.name,
-      table: this.options.table,
-      slotName: this.options.slotName,
-      publicationName: this.options.publicationName,
-      startLsn,
-    });
-
-    const slotCreated = await this.#createSlot();
-
-    if (!slotCreated) {
-      return this.stop();
-    }
-
-    this.logger.info("Slot created", {
-      name: this.options.name,
-      table: this.options.table,
-      slotName: this.options.slotName,
-      publicationName: this.options.publicationName,
-      startLsn,
-    });
-
-    // 5. Start replication (pgoutput)
-    const parser = new PgoutputParser();
-    const sql = getPgoutputStartReplicationSQL(this.options.slotName, this.lastLsn, {
-      protoVersion: 1,
-      publicationNames: [this.options.publicationName],
-      messages: false,
-    });
-
-    // 6. Listen for replication events (copyData, etc.)
-    if (!this.connection) {
-      this.events.emit(
-        "error",
-        new LogicalReplicationClientError("No connection after starting replication")
-      );
-      return this.stop();
-    }
-
-    this.connection.once("replicationStart", () => {
-      this._isStopped = false;
-      this.events.emit("start");
-    });
-
-    this.connection.on(
-      "copyData",
-      async ({ chunk: buffer }: { length: number; chunk: Buffer; name: string }) => {
-        // pgoutput protocol: 0x77 = XLogData, 0x6b = Primary keepalive
-        if (buffer[0] !== 0x77 && buffer[0] !== 0x6b) {
-          this.logger.warn("Unknown replication message type", { byte: buffer[0] });
-          return;
-        }
-        const lsn =
-          buffer.readUInt32BE(1).toString(16).toUpperCase() +
-          "/" +
-          buffer.readUInt32BE(5).toString(16).toUpperCase();
-
-        if (buffer[0] === 0x77) {
-          // XLogData
-          try {
-            const start = process.hrtime.bigint();
-            const log = parser.parse(buffer.subarray(25));
-            const duration = process.hrtime.bigint() - start;
-            this.events.emit("data", { lsn, log, parseDuration: duration });
-            await this.#acknowledge(lsn);
-          } catch (err) {
-            this.logger.error("Failed to parse XLogData", { error: err });
-            this.events.emit("error", err instanceof Error ? err : new Error(String(err)));
-          }
-        } else if (buffer[0] === 0x6b) {
-          // Primary keepalive message
-          const timestamp = Math.floor(
-            buffer.readUInt32BE(9) * 4294967.296 + buffer.readUInt32BE(13) / 1000 + 946080000000
-          );
-          const shouldRespond = !!buffer.readInt8(17);
-          this.events.emit("heartbeat", { lsn, timestamp, shouldRespond });
-          if (shouldRespond) {
-            await this.#acknowledge(lsn);
-          }
-        }
-
-        this.lastAcknowledgedLsn = lsn;
+      if (this._intentionalStop) {
+        await this.#cleanupAttempt();
+        return this;
       }
-    );
 
-    // 7. Handle errors and cleanup
-    this.client.on("error", (err) => {
-      this.events.emit("error", err);
-    });
+      const publicationCreated = await this.#createPublication();
 
-    this.logger.info("Started replication", {
-      name: this.options.name,
-      table: this.options.table,
-      slotName: this.options.slotName,
-      publicationName: this.options.publicationName,
-      startLsn,
-      sql: sql.replace(/\s+/g, " "),
-    });
+      if (!publicationCreated) {
+        await this.#cleanupAttempt();
+        this.#scheduleResubscribe("create-publication-failed");
+        return this;
+      }
 
-    // Start the replication stream
-    this.client.query(sql).catch((err) => {
-      this.logger.error("Failed to start replication", {
+      this.logger.info("Publication created", {
         name: this.options.name,
         table: this.options.table,
         slotName: this.options.slotName,
         publicationName: this.options.publicationName,
-        error: err,
+        startLsn,
       });
 
-      this.events.emit("error", err);
-      return this.stop();
-    });
+      const slotCreated = await this.#createSlot();
+
+      if (!slotCreated) {
+        await this.#cleanupAttempt();
+        this.#scheduleResubscribe("create-slot-failed");
+        return this;
+      }
+
+      this.logger.info("Slot created", {
+        name: this.options.name,
+        table: this.options.table,
+        slotName: this.options.slotName,
+        publicationName: this.options.publicationName,
+        startLsn,
+      });
+
+      if (this._intentionalStop) {
+        await this.#cleanupAttempt();
+        return this;
+      }
+
+      // 5. Start replication (pgoutput)
+      const parser = new PgoutputParser();
+      const sql = getPgoutputStartReplicationSQL(this.options.slotName, this.lastLsn, {
+        protoVersion: 1,
+        publicationNames: [this.options.publicationName],
+        messages: false,
+      });
+
+      // 6. Listen for replication events (copyData, etc.)
+      if (!this.connection) {
+        this.events.emit(
+          "error",
+          new LogicalReplicationClientError("No connection after starting replication")
+        );
+        await this.#cleanupAttempt();
+        this.#scheduleResubscribe("no-connection");
+        return this;
+      }
+
+      this.connection.once("replicationStart", () => {
+        if (this._intentionalStop) {
+          // shutdown() raced the stream start — tear this attempt down.
+          void this.#cleanupAttempt();
+          return;
+        }
+        this._isStopped = false;
+        this.resubscribeAttempts = 0;
+        this.events.emit("start");
+      });
+
+      this.connection.on(
+        "copyData",
+        async ({ chunk: buffer }: { length: number; chunk: Buffer; name: string }) => {
+          // pgoutput protocol: 0x77 = XLogData, 0x6b = Primary keepalive
+          if (buffer[0] !== 0x77 && buffer[0] !== 0x6b) {
+            this.logger.warn("Unknown replication message type", { byte: buffer[0] });
+            return;
+          }
+          const lsn =
+            buffer.readUInt32BE(1).toString(16).toUpperCase() +
+            "/" +
+            buffer.readUInt32BE(5).toString(16).toUpperCase();
+
+          if (buffer[0] === 0x77) {
+            // XLogData
+            try {
+              const start = process.hrtime.bigint();
+              const log = parser.parse(buffer.subarray(25));
+              const duration = process.hrtime.bigint() - start;
+              this.events.emit("data", { lsn, log, parseDuration: duration });
+              await this.#acknowledge(lsn);
+            } catch (err) {
+              this.logger.error("Failed to parse XLogData", { error: err });
+              this.events.emit("error", err instanceof Error ? err : new Error(String(err)));
+            }
+          } else if (buffer[0] === 0x6b) {
+            // Primary keepalive message
+            const timestamp = Math.floor(
+              buffer.readUInt32BE(9) * 4294967.296 + buffer.readUInt32BE(13) / 1000 + 946080000000
+            );
+            const shouldRespond = !!buffer.readInt8(17);
+            this.events.emit("heartbeat", { lsn, timestamp, shouldRespond });
+            if (shouldRespond) {
+              await this.#acknowledge(lsn);
+            }
+          }
+
+          this.lastAcknowledgedLsn = lsn;
+        }
+      );
+
+      // 7. Handle errors and cleanup
+      this.client.on("error", (err) => {
+        this.events.emit("error", err);
+      });
+
+      this.logger.info("Started replication", {
+        name: this.options.name,
+        table: this.options.table,
+        slotName: this.options.slotName,
+        publicationName: this.options.publicationName,
+        startLsn,
+        sql: sql.replace(/\s+/g, " "),
+      });
+
+      // Start the replication stream
+      this.client.query(sql).catch(async (err) => {
+        // A newer subscribe owns the client/lock now; don't tear it down.
+        if (attemptEpoch !== this.subscribeEpoch) return;
+
+        this.logger.error("Failed to start replication", {
+          name: this.options.name,
+          table: this.options.table,
+          slotName: this.options.slotName,
+          publicationName: this.options.publicationName,
+          error: err,
+        });
+
+        this.events.emit("error", err);
+        await this.#cleanupAttempt();
+        this.#scheduleResubscribe("start-replication-failed");
+      });
+    } catch (error) {
+      this.logger.error("Subscribe failed after leader election", {
+        name: this.options.name,
+        table: this.options.table,
+        slotName: this.options.slotName,
+        publicationName: this.options.publicationName,
+        error,
+      });
+
+      await this.#cleanupAttempt();
+      this.#scheduleResubscribe("subscribe-failed");
+      throw error;
+    }
 
     return this;
   }
@@ -688,7 +860,7 @@ export class LogicalReplicationClient {
     });
   }
 
-  async #acquireLeaderLock(): Promise<boolean> {
+  async #acquireLeaderLock(abortOnIntentionalStop = true): Promise<boolean> {
     const startTime = Date.now();
     const maxWaitTime = this.leaderLockTimeoutMs + this.leaderLockAcquireAdditionalTimeMs;
 
@@ -702,9 +874,22 @@ export class LogicalReplicationClient {
     let attempt = 0;
 
     while (Date.now() - startTime < maxWaitTime) {
+      if (abortOnIntentionalStop && this._intentionalStop) {
+        this.logger.info("Leader lock acquisition aborted by shutdown", {
+          name: this.options.name,
+          slotName: this.options.slotName,
+          publicationName: this.options.publicationName,
+          attempt,
+        });
+        return false;
+      }
+
       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
         );