fix(replication,webapp): recover leadership after a failed leader-lock extend

Run replication now recovers on its own after a Redis restart or outage.
The leader could previously lose its lock without noticing and log
"Cannot extend an already-expired lock" every few seconds while holding
the replication slot open, until the server was restarted. It now
re-acquires the lock, or steps down once and re-elects.

Deployments running under a process supervisor can set
`RUN_REPLICATION_MAX_RESUBSCRIBE_ATTEMPTS` (or the session equivalent)
to exit and be restarted when a replication stream cannot recover. It
defaults to 0, which keeps retrying and preserves existing behavior.

Mono-RevId: 465d1f9bd0bb720320c8dd294dde3c3a9b88f1d9

Author: ericallam

.server-changes/replication-leader-lock-recovery.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Run replication now recovers on its own after a Redis restart or outage, in place of logging "Cannot extend an already-expired lock" and holding the replication slot open until the server is restarted. Deployments running under a process supervisor can set `RUN_REPLICATION_MAX_RESUBSCRIBE_ATTEMPTS` to exit and be restarted when a stream cannot recover.

apps/webapp/app/env.server.ts

@@ -2015,6 +2015,18 @@ const EnvironmentSchema = z
     RUN_REPLICATION_FLUSH_BATCH_SIZE: z.coerce.number().int().default(100),
     RUN_REPLICATION_MAX_POISON_STRIPS_PER_BATCH: z.coerce.number().int().default(1),
     RUN_REPLICATION_LEADER_LOCK_TIMEOUT_MS: z.coerce.number().int().default(30_000),
+
+    // Replication self-heals in-process: a lost stream or leader lock is retried
+    // with backoff. These bound that. 0 (the default) retries forever, which is
+    // right without a supervisor; set a budget where something can restart the
+    // process (k8s, ECS, systemd) so an unrecoverable stream gets a clean slate
+    // instead of retrying against a slot that is never coming back.
+    RUN_REPLICATION_MAX_RESUBSCRIBE_ATTEMPTS: z.coerce.number().int().min(0).default(0),
+    // Grace period before exiting, so the final logs can flush.
+    RUN_REPLICATION_EXIT_DELAY_MS: z.coerce.number().int().min(0).default(5_000),
+    // Capped at 255: POSIX masks the code to `code & 0xff`, so anything larger
+    // could silently become 0 and read as a clean exit to a supervisor.
+    RUN_REPLICATION_EXIT_CODE: z.coerce.number().int().min(0).max(255).default(1),
     RUN_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS: z.coerce.number().int().default(10_000),
     RUN_REPLICATION_ACK_INTERVAL_SECONDS: z.coerce.number().int().default(10),
     RUN_REPLICATION_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
@@ -2093,6 +2105,18 @@ const EnvironmentSchema = z
     SESSION_REPLICATION_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
     SESSION_REPLICATION_FLUSH_BATCH_SIZE: z.coerce.number().int().default(100),
     SESSION_REPLICATION_LEADER_LOCK_TIMEOUT_MS: z.coerce.number().int().default(30_000),
+
+    // Replication self-heals in-process: a lost stream or leader lock is retried
+    // with backoff. These bound that. 0 (the default) retries forever, which is
+    // right without a supervisor; set a budget where something can restart the
+    // process (k8s, ECS, systemd) so an unrecoverable stream gets a clean slate
+    // instead of retrying against a slot that is never coming back.
+    SESSION_REPLICATION_MAX_RESUBSCRIBE_ATTEMPTS: z.coerce.number().int().min(0).default(0),
+    // Grace period before exiting, so the final logs can flush.
+    SESSION_REPLICATION_EXIT_DELAY_MS: z.coerce.number().int().min(0).default(5_000),
+    // Capped at 255: POSIX masks the code to `code & 0xff`, so anything larger
+    // could silently become 0 and read as a clean exit to a supervisor.
+    SESSION_REPLICATION_EXIT_CODE: z.coerce.number().int().min(0).max(255).default(1),
     SESSION_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS: z.coerce.number().int().default(10_000),
     SESSION_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS: z.coerce.number().int().default(10_000),
     SESSION_REPLICATION_LEADER_LOCK_RETRY_INTERVAL_MS: z.coerce.number().int().default(500),

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

@@ -0,0 +1,40 @@
+// What to do when a replication client has exhausted its in-process
+// self-healing. The client retries a lost stream or leader lock on its own with
+// backoff; once it gives up, the stream is not coming back without help (a
+// dropped slot, a publication that no longer exists, credentials that no longer
+// work). Retrying past that point hides a dead replica behind a healthy-looking
+// process.
+//
+// Exiting hands the problem to whatever supervises us (Kubernetes, ECS,
+// systemd), which can give the process a clean slate. Deployments without a
+// supervisor leave *_REPLICATION_MAX_RESUBSCRIBE_ATTEMPTS at 0, so the client
+// never gives up and this never runs.
+//
+// The process-control decision lives here, at the composition root, rather than
+// inside the replication services: they take an `onUnrecoverable` callback so
+// they stay testable without touching globals.
+
+// One exit, however many sources give up. The first failure is the useful one;
+// the rest would only race the same timer.
+let exitScheduled = false;
+
+export function scheduleReplicationExit(options: {
+  label: string;
+  delayMs: number;
+  exitCode: number;
+  details: Record<string, unknown>;
+}): void {
+  const { label, delayMs, exitCode, details } = options;
+
+  if (exitScheduled) return;
+  exitScheduled = true;
+
+  console.error(
+    `🗃️ ${label}: replication is unrecoverable, exiting in ${delayMs}ms so the supervisor can restart this process`,
+    { ...details, exitCode }
+  );
+
+  // Delayed so the logs above can flush; unref'd so this timer alone never
+  // keeps an otherwise-finished process alive.
+  setTimeout(() => process.exit(exitCode), delayMs).unref();
+}

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

@@ -10,6 +10,7 @@ import {
   setRunsReplicationGlobal,
 } from "./runsReplicationGlobal.server";
 import { runsReplicationSourceMetrics } from "./runsReplicationMetrics.server";
+import { scheduleReplicationExit } from "./replicationUnrecoverableExit.server";
 import {
   RunsReplicationService,
   type RunsReplicationSource,
@@ -253,6 +254,22 @@ function initializeRunsReplicationInstance() {
     // A source whose publication carries no usable table logs every 30s and replicates nothing.
     // Boot cannot see it (the source IS configured), so the counter is the alarmable signal.
     onSourceError: runsReplicationSourceMetrics.recordSourceError,
+    maxResubscribeAttempts: env.RUN_REPLICATION_MAX_RESUBSCRIBE_ATTEMPTS,
+    onUnrecoverable: ({
+      sourceId,
+      reason,
+      attempts,
+    }: {
+      sourceId: string;
+      reason: string;
+      attempts: number;
+    }) =>
+      scheduleReplicationExit({
+        label: "Runs replication",
+        delayMs: env.RUN_REPLICATION_EXIT_DELAY_MS,
+        exitCode: env.RUN_REPLICATION_EXIT_CODE,
+        details: { sourceId, reason, attempts },
+      }),
   };
 
   // Construct the SINGLE legacy source synchronously (the split gate has not resolved

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

@@ -122,6 +122,13 @@ export type RunsReplicationServiceOptions = {
    * count it on.
    */
   onSourceError?: (info: { sourceId: string; error: unknown }) => void;
+  /** 0 (default) retries a broken stream forever; above that the client gives up. */
+  maxResubscribeAttempts?: number;
+  /**
+   * Self-healing for a source has been exhausted. Injected rather than exiting
+   * here, so the service stays free of process control and testable.
+   */
+  onUnrecoverable?: (info: { sourceId: string; reason: string; attempts: number }) => void;
 };
 
 type PostgresTaskRun = TaskRun & { masterQueue: string };
@@ -367,6 +374,7 @@ export class RunsReplicationService {
         redisOptions: options.redisOptions,
         autoAcknowledge: false,
         resubscribeOnFailure: true,
+        maxResubscribeAttempts: options.maxResubscribeAttempts,
         publicationActions: ["insert", "update", "delete"],
         logger:
           options.logger ?? new Logger("LogicalReplicationClient", options.logLevel ?? "info"),
@@ -490,6 +498,15 @@ export class RunsReplicationService {
     client.events.on("leaderElection", (isLeader) => {
       this.logger.info("Leader election", { sourceId: source.id, isLeader });
     });
+
+    client.events.on("unrecoverable", ({ reason, attempts }) => {
+      this.logger.error("Replication client gave up; source is down", {
+        sourceId: source.id,
+        reason,
+        attempts,
+      });
+      this.options.onUnrecoverable?.({ sourceId: source.id, reason, attempts });
+    });
   }
 
   /** Exposed for tests and metrics — batches where nothing landed even after stripping JSON. */

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

@@ -3,6 +3,7 @@ import { env } from "~/env.server";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { singleton } from "~/utils/singleton";
 import { meter, provider } from "~/v3/tracer.server";
+import { scheduleReplicationExit } from "./replicationUnrecoverableExit.server";
 import { SessionsReplicationService } from "./sessionsReplicationService.server";
 import { signalsEmitter } from "./signals.server";
 
@@ -45,6 +46,14 @@ function initializeSessionsReplicationInstance() {
     leaderLockExtendIntervalMs: env.SESSION_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS,
     leaderLockAcquireAdditionalTimeMs: env.SESSION_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS,
     leaderLockRetryIntervalMs: env.SESSION_REPLICATION_LEADER_LOCK_RETRY_INTERVAL_MS,
+    maxResubscribeAttempts: env.SESSION_REPLICATION_MAX_RESUBSCRIBE_ATTEMPTS,
+    onUnrecoverable: ({ reason, attempts }) =>
+      scheduleReplicationExit({
+        label: "Sessions replication",
+        delayMs: env.SESSION_REPLICATION_EXIT_DELAY_MS,
+        exitCode: env.SESSION_REPLICATION_EXIT_CODE,
+        details: { reason, attempts },
+      }),
     ackIntervalSeconds: env.SESSION_REPLICATION_ACK_INTERVAL_SECONDS,
     logLevel: env.SESSION_REPLICATION_LOG_LEVEL,
     waitForAsyncInsert: env.SESSION_REPLICATION_WAIT_FOR_ASYNC_INSERT === "1",

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

@@ -54,6 +54,13 @@ export type SessionsReplicationServiceOptions = {
   leaderLockExtendIntervalMs?: number;
   leaderLockAcquireAdditionalTimeMs?: number;
   leaderLockRetryIntervalMs?: number;
+  /** 0 (default) retries a broken stream forever; above that the client gives up. */
+  maxResubscribeAttempts?: number;
+  /**
+   * Self-healing has been exhausted. Injected rather than exiting here, so the
+   * service stays free of process control and testable.
+   */
+  onUnrecoverable?: (info: { reason: string; attempts: number }) => void;
   ackIntervalSeconds?: number;
   acknowledgeTimeoutMs?: number;
   logger?: Logger;
@@ -188,6 +195,7 @@ export class SessionsReplicationService {
       redisOptions: options.redisOptions,
       autoAcknowledge: false,
       resubscribeOnFailure: true,
+      maxResubscribeAttempts: options.maxResubscribeAttempts,
       publicationActions: ["insert", "update", "delete"],
       logger: options.logger ?? new Logger("LogicalReplicationClient", options.logLevel ?? "info"),
       leaderLockTimeoutMs: options.leaderLockTimeoutMs ?? 30_000,
@@ -251,6 +259,14 @@ export class SessionsReplicationService {
       this.logger.info("Leader election", { isLeader });
     });
 
+    this._replicationClient.events.on("unrecoverable", ({ reason, attempts }) => {
+      this.logger.error("Replication client gave up; sessions replication is down", {
+        reason,
+        attempts,
+      });
+      options.onUnrecoverable?.({ reason, attempts });
+    });
+
     // Initialize retry configuration
     this._insertMaxRetries = options.insertMaxRetries ?? 3;
     this._insertBaseDelayMs = options.insertBaseDelayMs ?? 100;

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

@@ -198,6 +198,13 @@ export class WebhookDeliveriesReplicationService {
       publishViaPartitionRoot: options.publishViaPartitionRoot,
       redisOptions: options.redisOptions,
       autoAcknowledge: false,
+      // Without this, losing the leader lock stops this client for good: it
+      // tears the stream down (a non-leader must not hold the slot open) and
+      // has nothing to re-contend with. The runs and sessions services already
+      // enable it. Note that every teardown path below must then call the
+      // client's shutdown(), not stop(): only shutdown() latches the intentional
+      // stop that keeps a pending resubscribe from reviving the stream.
+      resubscribeOnFailure: true,
       publicationActions: ["insert", "update", "delete"],
       logger: options.logger ?? new Logger("LogicalReplicationClient", options.logLevel ?? "info"),
       leaderLockTimeoutMs: options.leaderLockTimeoutMs ?? 30_000,
@@ -276,7 +283,7 @@ export class WebhookDeliveriesReplicationService {
 
     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;
@@ -305,7 +312,7 @@ export class WebhookDeliveriesReplicationService {
   async stop() {
     this.logger.info("Stopping replication client");
 
-    await this._replicationClient.stop();
+    await this._replicationClient.shutdown();
 
     if (this._acknowledgeInterval) {
       clearInterval(this._acknowledgeInterval);
@@ -441,7 +448,7 @@ export class WebhookDeliveriesReplicationService {
     if (this._isShutDownComplete) return;
 
     if (this._isShuttingDown) {
-      this._replicationClient.stop().finally(() => {
+      this._replicationClient.shutdown().finally(() => {
         this._isSubscribed = false;
         this._isShutDownComplete = true;
       });