fix(webapp): reschedule reconnect when subscribe() throws

Addresses PR review feedback:

- LogicalReplicationClient.subscribe() can throw before its internal
  "error" listener is wired up (notably when pg client.connect() fails
  mid-failover). The reconnect strategy's catch block only logged, so
  recovery silently stopped. Now also calls scheduleReconnect(err) — the
  pendingReconnect guard makes it idempotent if an error event was also
  emitted.
- Reject negative values for the new replication-recovery env vars and
  cap exit codes at 255.
- Convert the new ReplicationErrorRecovery{Deps,} interfaces to type
  aliases to match the repo's TypeScript style.
- Tighten the reconnect dep comment to drop a stale "lastAcknowledgedLsn"
  reference (the wrapper-tracked resume LSN is what callers actually pass).
- Restore process.exit after service.shutdown() in the exit-strategy
  test so a delayed exit timer can't terminate the test worker.

Author: ericallam

apps/webapp/app/env.server.ts

@@ -1314,8 +1314,8 @@ const EnvironmentSchema = z
     RUN_REPLICATION_ERROR_STRATEGY: z
       .enum(["reconnect", "exit", "log"])
       .default("reconnect"),
-    RUN_REPLICATION_EXIT_DELAY_MS: z.coerce.number().int().default(5_000),
-    RUN_REPLICATION_EXIT_CODE: z.coerce.number().int().default(1),
+    RUN_REPLICATION_EXIT_DELAY_MS: z.coerce.number().int().min(0).default(5_000),
+    RUN_REPLICATION_EXIT_CODE: z.coerce.number().int().min(0).max(255).default(1),
 
     // Session replication (Postgres → ClickHouse sessions_v1). Shares Redis
     // with the runs replicator for leader locking but has its own slot and
@@ -1352,15 +1352,15 @@ const EnvironmentSchema = z
     SESSION_REPLICATION_ERROR_STRATEGY: z
       .enum(["reconnect", "exit", "log"])
       .default("reconnect"),
-    SESSION_REPLICATION_EXIT_DELAY_MS: z.coerce.number().int().default(5_000),
-    SESSION_REPLICATION_EXIT_CODE: z.coerce.number().int().default(1),
+    SESSION_REPLICATION_EXIT_DELAY_MS: z.coerce.number().int().min(0).default(5_000),
+    SESSION_REPLICATION_EXIT_CODE: z.coerce.number().int().min(0).max(255).default(1),
 
     // Reconnect tuning shared across both replication services. Only
     // applies when error strategy is `reconnect`. Max attempts of 0 means
     // unlimited (default).
-    REPLICATION_RECONNECT_INITIAL_DELAY_MS: z.coerce.number().int().default(1_000),
-    REPLICATION_RECONNECT_MAX_DELAY_MS: z.coerce.number().int().default(60_000),
-    REPLICATION_RECONNECT_MAX_ATTEMPTS: z.coerce.number().int().default(0),
+    REPLICATION_RECONNECT_INITIAL_DELAY_MS: z.coerce.number().int().min(0).default(1_000),
+    REPLICATION_RECONNECT_MAX_DELAY_MS: z.coerce.number().int().min(0).default(60_000),
+    REPLICATION_RECONNECT_MAX_ATTEMPTS: z.coerce.number().int().min(0).default(0),
 
     // Clickhouse
     CLICKHOUSE_URL: z.string(),

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

@@ -27,26 +27,26 @@ export type ReplicationErrorRecoveryStrategy =
     }
   | { type: "log" };
 
-export interface ReplicationErrorRecoveryDeps {
+export type ReplicationErrorRecoveryDeps = {
   strategy: ReplicationErrorRecoveryStrategy;
   logger: Logger;
   // Re-subscribe the underlying replication client. Implementations should
-  // call client.subscribe(lastAcknowledgedLsn) and resolve once that returns.
+  // call client.subscribe(...) and resolve once the stream is started.
   reconnect: () => Promise<void>;
   // True once the host service has begun graceful shutdown — recovery
   // suppresses all work in that state.
   isShuttingDown: () => boolean;
-}
+};
 
-export interface ReplicationErrorRecovery {
+export type ReplicationErrorRecovery = {
   // Called from the replication client's "error" event handler.
   handle(error: unknown): void;
   // Called from the replication client's "start" event handler. Resets the
   // reconnect attempt counter so the next failure starts from initialDelayMs.
   notifyStreamStarted(): void;
   // Cancel any pending reconnect/exit timer. Called from shutdown().
   dispose(): void;
-}
+};
 
 export function createReplicationErrorRecovery(
   deps: ReplicationErrorRecoveryDeps
@@ -91,13 +91,17 @@ export function createReplicationErrorRecovery(
         // Success path is handled by notifyStreamStarted, which fires from
         // the replication client's "start" event after the stream is live.
       } catch (err) {
-        // subscribe() emits an "error" event of its own on failure, so the
-        // next attempt is scheduled by handle(). Log here anyway so reconnect
-        // failures stay visible even if the error event is suppressed.
+        // subscribe() can throw without first emitting an "error" event —
+        // notably when the initial pg client.connect() fails because Postgres
+        // is still unreachable mid-failover. Schedule the next attempt
+        // ourselves so recovery doesn't silently stop. If subscribe() did
+        // also emit an "error" event, handle() will call scheduleReconnect()
+        // first; the guard on pendingReconnect makes this idempotent.
         logger.error("Replication reconnect attempt failed", {
           attempt,
           error: err,
         });
+        scheduleReconnect(err);
       }
     }, delay);
   }

apps/webapp/test/runsReplicationService.errorRecovery.test.ts

@@ -126,8 +126,10 @@ describe("RunsReplicationService error recovery", () => {
 
         expect(exitSpy).toHaveBeenCalledWith(1);
       } finally {
-        exitSpy.mockRestore();
+        // shutdown() before mockRestore() so any in-flight exit timer can
+        // be disposed without terminating the Vitest worker.
         await service.shutdown();
+        exitSpy.mockRestore();
       }
     }
   );