fix(replication): probe leadership + assert lock keys by slot name

Follow-on to keying the leader lock on slotName: the admin runs-replication
status route probed the old name-keyed Redis key (would report leader:false
for every source), and the multi-source wiring test asserted the old
name-keyed lock keys (CI-red). Both updated to the slot-keyed format.

Author: d-cs

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/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);