fix(webapp): drop bogus isStopped check, route leader-lock failure through handle()

The previous post-subscribe() isStopped check was always true on the
happy path: subscribe() calls stop() up front (setting _isStopped=true)
and only resets the flag inside the replicationStart event, which fires
asynchronously after subscribe() returns. So the check threw on every
successful reconnect, the catch rescheduled, the next attempt tore down
the just-built client, and the cycle continued — replication briefly
worked between teardowns, which is why the integration test passed.

Replace it with the correct nudge: subscribe to leaderElection and call
the recovery handler on isLeader=false. That's the only subscribe()
exit path that doesn't either throw or emit an "error" event (the other
silent-return paths emit "error" first via createPublication/createSlot
failures).

Author: ericallam

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

@@ -264,13 +264,6 @@ export class RunsReplicationService {
       logger: this.logger,
       reconnect: async () => {
         await this._replicationClient.subscribe(this._latestCommitEndLsn ?? undefined);
-        if (this._replicationClient.isStopped) {
-          // subscribe() can resolve without throwing or emitting an "error"
-          // event when leader-lock acquisition fails (see LogicalReplication-
-          // Client.subscribe leader-election branch). Throw here so the
-          // recovery handler reschedules the next attempt.
-          throw new Error("Replication client stopped after subscribe()");
-        }
       },
       isShuttingDown: () => this._isShuttingDown || this._isShutDownComplete,
     });
@@ -293,6 +286,15 @@ export class RunsReplicationService {
 
     this._replicationClient.events.on("leaderElection", (isLeader) => {
       this.logger.info("Leader election", { isLeader });
+      if (!isLeader) {
+        // Failed leader election doesn't throw or emit an "error" event —
+        // subscribe() just emits leaderElection(false), calls stop(), and
+        // returns. Nudge the recovery handler so reconnect doesn't silently
+        // stall when another instance holds the lock.
+        this._errorRecovery.handle(
+          new Error("Failed to acquire replication leader lock")
+        );
+      }
     });
 
     // Initialize retry configuration

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

@@ -245,11 +245,6 @@ export class SessionsReplicationService {
       logger: this.logger,
       reconnect: async () => {
         await this._replicationClient.subscribe(this._latestCommitEndLsn ?? undefined);
-        if (this._replicationClient.isStopped) {
-          // See RunsReplicationService for the rationale: subscribe() can
-          // resolve without throwing when leader-lock acquisition fails.
-          throw new Error("Replication client stopped after subscribe()");
-        }
       },
       isShuttingDown: () => this._isShuttingDown || this._isShutDownComplete,
     });
@@ -272,6 +267,12 @@ export class SessionsReplicationService {
 
     this._replicationClient.events.on("leaderElection", (isLeader) => {
       this.logger.info("Leader election", { isLeader });
+      if (!isLeader) {
+        // See RunsReplicationService for the rationale.
+        this._errorRecovery.handle(
+          new Error("Failed to acquire replication leader lock")
+        );
+      }
     });
 
     // Initialize retry configuration