fix: add retry mechanism for ClickHouse replication leader election

When the PostgreSQL connection is lost (e.g., during database failover), the replication client would stop and never retry. This left replication permanently broken until a manual pod restart.

Changes:
- Add hasLeader() method to check if a leader lock exists in Redis
- Add automatic retry with exponential backoff when:
- Leader election fails (another pod is leader or stale lock)
- Replication client stops due to an error
- Retry delays: 35s -> 70s -> 140s -> 280s -> 300s (max 5 min)
- Skip retry attempt if a leader already exists (just reschedule)

This ensures replication automatically recovers from:
- Database failovers/restarts
- Network issues
- Pod crashes (standby pods will eventually take over)

Author: NERLOE

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

@@ -112,7 +112,7 @@ export class RunsReplicationService {
   private _lastAcknowledgedLsn: string | null = null;
   private _acknowledgeInterval: NodeJS.Timeout | null = null;
   private _leaderElectionRetryTimer: NodeJS.Timeout | null = null;
-  private _leaderElectionRetryIntervalMs: number;
+  private _leaderElectionRetryCount = 0;
   // Retry configuration
   private _insertMaxRetries: number;
   private _insertBaseDelayMs: number;
@@ -259,10 +259,16 @@ export class RunsReplicationService {
       this.logger.error("Replication client error", {
         error,
       });
+
+      if (this._replicationClient.isStopped && !this._isShuttingDown) {
+        this._isSubscribed = false;
+        this.#scheduleLeaderElectionRetry();
+      }
     });
 
     this._replicationClient.events.on("start", () => {
       this.logger.info("Replication client started");
+      this._isSubscribed = true;
     });
 
     this._replicationClient.events.on("acknowledge", ({ lsn }: { lsn: string }) => {
@@ -271,25 +277,20 @@ export class RunsReplicationService {
 
     this._replicationClient.events.on("leaderElection", (isLeader: boolean) => {
       this.logger.info("Leader election", { isLeader });
-      
+      this._isSubscribed = isLeader;
+
       // Forward the event to external listeners
       this.events.emit("leaderElection", isLeader);
-      
-      if (!isLeader) {
-        // Failed to acquire leadership - reset subscription flag and schedule retry
-        this._isSubscribed = false;
+
+      if (!isLeader && !this._isShuttingDown) {
         this.#scheduleLeaderElectionRetry();
-      } else {
-        // Successfully acquired leadership - clear any pending retries
-        this.#clearLeaderElectionRetry();
       }
     });
 
     // Initialize retry configuration
     this._insertMaxRetries = options.insertMaxRetries ?? 3;
     this._insertBaseDelayMs = options.insertBaseDelayMs ?? 100;
     this._insertMaxDelayMs = options.insertMaxDelayMs ?? 2000;
-    this._leaderElectionRetryIntervalMs = options.leaderElectionRetryIntervalMs ?? 5_000;
   }
 
   public async shutdown() {
@@ -299,10 +300,10 @@ export class RunsReplicationService {
 
     this.logger.info("Initiating shutdown of runs replication service");
 
-    // Clear any pending leader election retries
-    this.#clearLeaderElectionRetry();
-
-    this._isSubscribed = false;
+    if (this._leaderElectionRetryTimer) {
+      clearTimeout(this._leaderElectionRetryTimer);
+      this._leaderElectionRetryTimer = null;
+    }
 
     if (!this._currentTransaction) {
       this.logger.info("No transaction to commit, shutting down immediately");
@@ -334,30 +335,93 @@ export class RunsReplicationService {
 
     await this._replicationClient.subscribe(this._latestCommitEndLsn ?? undefined);
 
+    if (this._isSubscribed) {
+      this.#startAcknowledgeInterval();
+      this._concurrentFlushScheduler.start();
+    }
+  }
+
+  #startAcknowledgeInterval() {
+    if (this._acknowledgeInterval) {
+      clearInterval(this._acknowledgeInterval);
+    }
     this._acknowledgeInterval = setInterval(this.#acknowledgeLatestTransaction.bind(this), 1000);
-    this._concurrentFlushScheduler.start();
+  }
+
+  #scheduleLeaderElectionRetry() {
+    if (this._isShuttingDown || this._isShutDownComplete) {
+      return;
+    }
+
+    if (this._leaderElectionRetryTimer) {
+      clearTimeout(this._leaderElectionRetryTimer);
+    }
+
+    const baseDelayMs = (this.options.leaderLockTimeoutMs ?? 30_000) + 5_000;
+    const maxDelayMs = 5 * 60 * 1_000; // 5 minutes max
+    const retryDelayMs = Math.min(baseDelayMs * Math.pow(2, this._leaderElectionRetryCount), maxDelayMs);
+
+    this._leaderElectionRetryCount++;
+
+    this.logger.info("Scheduling leader election retry", {
+      retryDelayMs,
+      retryCount: this._leaderElectionRetryCount,
+    });
+
+    this._leaderElectionRetryTimer = setTimeout(async () => {
+      if (this._isShuttingDown || this._isShutDownComplete || this._isSubscribed) {
+        return;
+      }
+
+      const hasLeader = await this._replicationClient.hasLeader();
+
+      if (hasLeader) {
+        this.logger.info("Leader exists, skipping retry", {
+          retryCount: this._leaderElectionRetryCount,
+        });
+        this.#scheduleLeaderElectionRetry();
+        return;
+      }
+
+      this.logger.info("No leader found, attempting to become leader", {
+        retryCount: this._leaderElectionRetryCount,
+      });
+
+      await this._replicationClient.subscribe(this._latestCommitEndLsn ?? undefined);
+
+      if (this._isSubscribed) {
+        this._leaderElectionRetryCount = 0;
+        this.#startAcknowledgeInterval();
+        this._concurrentFlushScheduler.start();
+      }
+    }, retryDelayMs);
   }
 
   async stop() {
     this.logger.info("Stopping replication client");
 
-    // Clear any pending leader election retries
-    this.#clearLeaderElectionRetry();
-
-    this._isSubscribed = false;
+    if (this._leaderElectionRetryTimer) {
+      clearTimeout(this._leaderElectionRetryTimer);
+      this._leaderElectionRetryTimer = null;
+    }
 
     await this._replicationClient.stop();
 
     if (this._acknowledgeInterval) {
       clearInterval(this._acknowledgeInterval);
     }
+
+    this._isSubscribed = false;
+    this._leaderElectionRetryCount = 0;
   }
 
   async teardown() {
     this.logger.info("Teardown replication client");
 
-    // Clear any pending leader election retries
-    this.#clearLeaderElectionRetry();
+    if (this._leaderElectionRetryTimer) {
+      clearTimeout(this._leaderElectionRetryTimer);
+      this._leaderElectionRetryTimer = null;
+    }
 
     this._isSubscribed = false;
 
@@ -1005,43 +1069,6 @@ export class RunsReplicationService {
 
     return { data: parsedData };
   }
-
-  #scheduleLeaderElectionRetry() {
-    // Clear any existing retry timer
-    this.#clearLeaderElectionRetry();
-
-    // Don't retry if we're shutting down or already shut down
-    if (this._isShuttingDown || this._isShutDownComplete) {
-      return;
-    }
-
-    this.logger.info("Scheduling leader election retry", {
-      retryIntervalMs: this._leaderElectionRetryIntervalMs,
-    });
-
-    this._leaderElectionRetryTimer = setTimeout(async () => {
-      if (this._isShuttingDown || this._isShutDownComplete) {
-        return;
-      }
-
-      this.logger.info("Retrying leader election after failed attempt");
-
-      try {
-        await this.start();
-      } catch (error) {
-        this.logger.error("Error during leader election retry", { error });
-        // If start() fails, the leaderElection event will be emitted with false,
-        // which will trigger another retry via #scheduleLeaderElectionRetry
-      }
-    }, this._leaderElectionRetryIntervalMs);
-  }
-
-  #clearLeaderElectionRetry() {
-    if (this._leaderElectionRetryTimer) {
-      clearTimeout(this._leaderElectionRetryTimer);
-      this._leaderElectionRetryTimer = null;
-    }
-  }
 }
 
 export type ConcurrentFlushSchedulerConfig<T> = {

internal-packages/replication/src/client.ts

@@ -114,6 +114,12 @@ export class LogicalReplicationClient {
     return this._isStopped;
   }
 
+  public async hasLeader(): Promise<boolean> {
+    const lockKey = `logical-replication-client:${this.options.name}`;
+    const exists = await this.redis.exists(lockKey);
+    return exists === 1;
+  }
+
   constructor(options: LogicalReplicationClientOptions) {
     this.options = options;
     this.logger = options.logger ?? new Logger("LogicalReplicationClient", "info");