feat(run-ops): ClickHouse multi-source replication fan-in + admin ops

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: d-cs

apps/webapp/app/routes/admin.api.v1.runs-replication.backfill.ts

@@ -5,6 +5,7 @@ import { prisma } from "~/db.server";
 import { runStore } from "~/v3/runStore.server";
 import { logger } from "~/services/logger.server";
 import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
+import { getRunsReplicationGlobal } from "~/services/runsReplicationGlobal.server";
 import { runsReplicationInstance } from "~/services/runsReplicationInstance.server";
 import { FINAL_RUN_STATUSES } from "~/v3/taskStatus";
 
@@ -40,11 +41,12 @@ export async function action({ request }: ActionFunctionArgs) {
       runs.push(...batchRuns);
     }
 
-    if (!runsReplicationInstance) {
+    const service = getRunsReplicationGlobal() ?? runsReplicationInstance;
+    if (!service) {
       throw new Error("Runs replication instance not found");
     }
 
-    await runsReplicationInstance.backfill(
+    await service.backfill(
       runs.map((run) => ({
         ...run,
         masterQueue: run.workerQueue,

apps/webapp/app/routes/admin.api.v1.runs-replication.status.ts

@@ -0,0 +1,60 @@
+import { type LoaderFunctionArgs, json } from "@remix-run/server-runtime";
+import Redis from "ioredis";
+import { env } from "~/env.server";
+import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
+import { getRunsReplicationConfiguredSources } from "~/services/runsReplicationGlobal.server";
+
+/**
+ * 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>
+ */
+async function probeLeadership(sourceIds: string[]): Promise<Map<string, boolean>> {
+  const leaders = new Map<string, boolean>();
+
+  const redis = new Redis({
+    keyPrefix: "runs-replication:logical-replication-client:",
+    port: env.RUN_REPLICATION_REDIS_PORT ?? undefined,
+    host: env.RUN_REPLICATION_REDIS_HOST ?? undefined,
+    username: env.RUN_REPLICATION_REDIS_USERNAME ?? undefined,
+    password: env.RUN_REPLICATION_REDIS_PASSWORD ?? undefined,
+    enableAutoPipelining: true,
+    ...(env.RUN_REPLICATION_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+  });
+
+  try {
+    for (const id of sourceIds) {
+      const exists = await redis.exists(`logical-replication-client:runs-replication:${id}`);
+      leaders.set(id, exists === 1);
+    }
+  } finally {
+    await redis.quit();
+  }
+
+  return leaders;
+}
+
+export async function loader({ request }: LoaderFunctionArgs) {
+  await requireAdminApiRequest(request);
+
+  const sources = getRunsReplicationConfiguredSources();
+
+  if (!sources || sources.length === 0) {
+    return json({ enabled: false, sources: [] });
+  }
+
+  const leaders = await probeLeadership(sources.map((s) => s.id));
+
+  return json({
+    enabled: env.RUN_REPLICATION_ENABLED === "1" && sources.length > 0,
+    sources: sources.map((s) => ({
+      id: s.id,
+      slotName: s.slotName,
+      originGeneration: s.originGeneration,
+      leader: leaders.get(s.id) ?? false,
+    })),
+  });
+}

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

@@ -2,10 +2,14 @@ import { RunsReplicationService } from "./runsReplicationService.server";
 
 const GLOBAL_RUNS_REPLICATION_KEY = Symbol.for("dev.trigger.ts.runs-replication");
 const GLOBAL_TCP_MONITOR_KEY = Symbol.for("dev.trigger.ts.tcp-monitor");
+const GLOBAL_RUNS_REPLICATION_SOURCES_KEY = Symbol.for("dev.trigger.ts.runs-replication-sources");
+
+export type ConfiguredSource = { id: string; slotName: string; originGeneration: number };
 
 type RunsReplicationGlobal = {
   [GLOBAL_RUNS_REPLICATION_KEY]?: RunsReplicationService;
   [GLOBAL_TCP_MONITOR_KEY]?: NodeJS.Timeout;
+  [GLOBAL_RUNS_REPLICATION_SOURCES_KEY]?: ConfiguredSource[];
 };
 
 const _globalThis = typeof globalThis === "object" ? globalThis : global;
@@ -23,6 +27,14 @@ export function unregisterRunsReplicationGlobal() {
   delete _global[GLOBAL_RUNS_REPLICATION_KEY];
 }
 
+export function getRunsReplicationConfiguredSources(): ConfiguredSource[] | undefined {
+  return _global[GLOBAL_RUNS_REPLICATION_SOURCES_KEY];
+}
+
+export function setRunsReplicationConfiguredSources(sources: ConfiguredSource[]) {
+  _global[GLOBAL_RUNS_REPLICATION_SOURCES_KEY] = sources;
+}
+
 export function getTcpMonitorGlobal(): NodeJS.Timeout | undefined {
   return _global[GLOBAL_TCP_MONITOR_KEY];
 }

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

@@ -2,15 +2,90 @@ import invariant from "tiny-invariant";
 import { env } from "~/env.server";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { singleton } from "~/utils/singleton";
+import { isSplitEnabled } from "~/v3/runOpsMigration/splitMode.server";
 import { meter, provider } from "~/v3/tracer.server";
-import { RunsReplicationService } from "./runsReplicationService.server";
+import {
+  setRunsReplicationConfiguredSources,
+  setRunsReplicationGlobal,
+} from "./runsReplicationGlobal.server";
+import {
+  RunsReplicationService,
+  type RunsReplicationSource,
+} from "./runsReplicationService.server";
 import { signalsEmitter } from "./signals.server";
 
 export const runsReplicationInstance = singleton(
   "runsReplicationInstance",
   initializeRunsReplicationInstance
 );
 
+export function buildReplicationSources(args: {
+  splitEnabled: boolean;
+  legacyUrl: string;
+  newUrl?: string;
+  /** `false` forces the new source off under split; undefined follows split. */
+  newSourceOverride?: boolean;
+  legacySlotName: string;
+  legacyPublicationName: string;
+  legacyOriginGeneration: number;
+  newSlotName: string;
+  newPublicationName: string;
+  newOriginGeneration: number;
+}): RunsReplicationSource[] {
+  const legacy: RunsReplicationSource = {
+    id: "legacy",
+    pgConnectionUrl: args.legacyUrl,
+    slotName: args.legacySlotName,
+    publicationName: args.legacyPublicationName,
+    originGeneration: args.legacyOriginGeneration,
+  };
+
+  const newSourceOn = args.splitEnabled && !!args.newUrl && args.newSourceOverride !== false;
+
+  if (!newSourceOn || !args.newUrl) {
+    return [legacy];
+  }
+
+  const next: RunsReplicationSource = {
+    id: "new",
+    pgConnectionUrl: args.newUrl,
+    slotName: args.newSlotName,
+    publicationName: args.newPublicationName,
+    originGeneration: args.newOriginGeneration,
+  };
+
+  return [legacy, next];
+}
+
+/**
+ * The residency-split gate and the `#new`->ClickHouse replication gate are
+ * independent env vars. If split is on (ksuid runs are minted on the new DB) but the
+ * constructed sources[] has no `"new"` source, every ksuid run is silently missing from
+ * ClickHouse — under-counting all CH-fronted usage/cost/metrics aggregates with no
+ * Postgres fallback. Couple the gates at boot: this misconfiguration must fail loudly
+ * rather than ship a fleet-wide under-count.
+ */
+export class SplitReplicationMisconfiguredError extends Error {
+  constructor() {
+    super(
+      "RUN_OPS_SPLIT_ENABLED is on but the runs-replication sources[] has no \"new\" source: " +
+        "ksuid runs on the new DB would not replicate to ClickHouse, under-counting every " +
+        "ClickHouse-fronted aggregate. Enable the new replication source " +
+        "(RUN_REPLICATION_NEW_ENABLED / RUN_OPS_DATABASE_URL) or turn the split off."
+    );
+    this.name = "SplitReplicationMisconfiguredError";
+  }
+}
+
+export function assertReplicationCoversSplit(args: {
+  splitEnabled: boolean;
+  sources: RunsReplicationSource[];
+}): void {
+  if (args.splitEnabled && !args.sources.some((s) => s.id === "new")) {
+    throw new SplitReplicationMisconfiguredError();
+  }
+}
+
 function initializeRunsReplicationInstance() {
   const { DATABASE_URL } = process.env;
   invariant(typeof DATABASE_URL === "string", "DATABASE_URL env var not set");
@@ -22,12 +97,11 @@ function initializeRunsReplicationInstance() {
 
   console.log("🗃️  Runs replication service enabled");
 
-  const service = new RunsReplicationService({
+  // Shared options for both the legacy-only and the multi-source constructions.
+  // Excludes per-source identity (pgConnectionUrl/slotName/publicationName/sources).
+  const baseReplicationOptions = {
     clickhouseFactory,
-    pgConnectionUrl: DATABASE_URL,
     serviceName: "runs-replication",
-    slotName: env.RUN_REPLICATION_SLOT_NAME,
-    publicationName: env.RUN_REPLICATION_PUBLICATION_NAME,
     redisOptions: {
       keyPrefix: "runs-replication:",
       port: env.RUN_REPLICATION_REDIS_PORT ?? undefined,
@@ -55,24 +129,94 @@ function initializeRunsReplicationInstance() {
     insertStrategy: env.RUN_REPLICATION_INSERT_STRATEGY,
     disablePayloadInsert: env.RUN_REPLICATION_DISABLE_PAYLOAD_INSERT === "1",
     disableErrorFingerprinting: env.RUN_REPLICATION_DISABLE_ERROR_FINGERPRINTING === "1",
+  };
+
+  // Construct the SINGLE legacy source synchronously (the split gate has not resolved
+  // yet at module-init time, and singleton(...) memoizes this synchronous return value).
+  let service = new RunsReplicationService({
+    ...baseReplicationOptions,
+    pgConnectionUrl: DATABASE_URL,
+    slotName: env.RUN_REPLICATION_SLOT_NAME,
+    publicationName: env.RUN_REPLICATION_PUBLICATION_NAME,
   });
 
+  // Register the live handle so the status route + lifecycle routes can find it.
+  setRunsReplicationGlobal(service);
+  setRunsReplicationConfiguredSources([
+    {
+      id: "legacy",
+      slotName: env.RUN_REPLICATION_SLOT_NAME,
+      originGeneration: env.RUN_REPLICATION_LEGACY_ORIGIN_GENERATION,
+    },
+  ]);
+
   if (env.RUN_REPLICATION_ENABLED === "1") {
-    clickhouseFactory
-      .isReady()
-      .then(() => service.start())
-      .then(() => {
-        console.log("🗃️ Runs replication service started");
+    // Shape B (construct-after-gate): resolve the async split gate ONCE at boot, and
+    // when both sources are enabled rebuild `service` with sources[] before starting.
+    // The legacy-only instance above is never started in the dual path (no slot/lock
+    // taken). runsReplicationService.server.ts is untouched. The create route also calls
+    // setRunsReplicationGlobal — last-writer-wins is the existing contract.
+    isSplitEnabled()
+      .then((splitEnabled) => {
+        const sources = buildReplicationSources({
+          splitEnabled,
+          legacyUrl: DATABASE_URL,
+          newUrl: env.RUN_OPS_DATABASE_URL ?? env.TASK_RUN_DATABASE_URL,
+          newSourceOverride:
+            env.RUN_REPLICATION_NEW_ENABLED === "disabled" ? false : undefined,
+          legacySlotName: env.RUN_REPLICATION_SLOT_NAME,
+          legacyPublicationName: env.RUN_REPLICATION_PUBLICATION_NAME,
+          legacyOriginGeneration: env.RUN_REPLICATION_LEGACY_ORIGIN_GENERATION,
+          newSlotName: env.RUN_REPLICATION_NEW_SLOT_NAME,
+          newPublicationName: env.RUN_REPLICATION_NEW_PUBLICATION_NAME,
+          newOriginGeneration: env.RUN_REPLICATION_NEW_ORIGIN_GENERATION,
+        });
+
+        // Refuse to start replication if split is on but `#new` is not a source.
+        assertReplicationCoversSplit({ splitEnabled, sources });
+
+        if (sources.length > 1) {
+          // The scalar pgConnectionUrl/slotName/publicationName remain required on the
+          // options type, but are ignored when sources[] is non-empty — the
+          // service normalizes off sources. Pass the legacy scalars to satisfy the type.
+          service = new RunsReplicationService({
+            ...baseReplicationOptions,
+            pgConnectionUrl: DATABASE_URL,
+            slotName: env.RUN_REPLICATION_SLOT_NAME,
+            publicationName: env.RUN_REPLICATION_PUBLICATION_NAME,
+            sources,
+          });
+          setRunsReplicationGlobal(service);
+          setRunsReplicationConfiguredSources(
+            sources.map((s) => ({
+              id: s.id,
+              slotName: s.slotName,
+              originGeneration: s.originGeneration,
+            }))
+          );
+        }
+
+        return clickhouseFactory.isReady().then(() => service.start());
       })
+      .then(() => console.log("🗃️ Runs replication service started"))
       .catch((error) => {
-        console.error("🗃️ Runs replication service failed to start", {
-          error,
-        });
+        if (error instanceof SplitReplicationMisconfiguredError) {
+          // A silent ClickHouse under-count is worse than a crash — make it fatal.
+          console.error("🚨 FATAL: run-ops split / ClickHouse replication misconfiguration", {
+            error,
+          });
+          process.exit(1);
+        }
+        console.error("🗃️ Runs replication service failed to start", { error });
       });
 
-    signalsEmitter.on("SIGTERM", service.shutdown.bind(service));
-    signalsEmitter.on("SIGINT", service.shutdown.bind(service));
+    // Closures over the `let` so SIGTERM/SIGINT hit whichever instance is live (NOT a
+    // stale .bind() to the discarded legacy-only instance).
+    signalsEmitter.on("SIGTERM", () => service.shutdown());
+    signalsEmitter.on("SIGINT", () => service.shutdown());
   }
 
+  // Returns the legacy-only instance synchronously (singleton memoizes this). Lifecycle
+  // routes read getRunsReplicationGlobal() first, so they get the live multi-source one.
   return service;
 }