feat(webapp,docker): run-ops boot interlocks and migrations at N databases

apps/webapp/app/db.server.ts

@@ -599,6 +599,16 @@ export const runOpsSplitReadEnabled: boolean = computeRunOpsSplitReadEnabled({
   controlPlaneReplica: $replica,
   hasNewUrl: !!env.RUN_OPS_DATABASE_URL,
   hasLegacyUrl: !!env.RUN_OPS_LEGACY_DATABASE_URL,
+  // Observability only: a non-distinct shard handle warns and never changes the gen-1 verdict.
+  // Empty unless RUN_OPS_SHARDS is configured.
+  shardHandles: runOpsShardHandles.map((handle) => ({
+    key: handle.key,
+    writer: handle.writer,
+    replica: handle.replica,
+    // The DECLARED field, not client identity: an aliased shard shares its target's client by
+    // reference, so identity comparison cannot tell the two apart.
+    aliasOf: env.RUN_OPS_SHARDS.find((d) => d.key === handle.key)?.aliasOf,
+  })),
   logger,
 });
 

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

@@ -3,6 +3,7 @@ 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 { nonAliasedShards } from "~/v3/runOpsShards.server";
 import { meter, provider } from "~/v3/tracer.server";
 import {
   setRunsReplicationConfiguredSources,
@@ -31,6 +32,17 @@ export function buildReplicationSources(args: {
   newSlotName: string;
   newPublicationName: string;
   newOriginGeneration: number;
+  /**
+   * Gen-2 shards that own their own database, each with its own slot, publication and origin
+   * generation. An aliased shard is absent: its target's slot already covers its WAL.
+   */
+  shards?: Array<{
+    key: string;
+    url: string;
+    /** The DIRECT, non-pooled DSN. Logical replication needs a session-mode connection. */
+    directUrl?: string;
+    replication: { slotName: string; publicationName: string; originGeneration: number };
+  }>;
 }): RunsReplicationSource[] {
   const legacy: RunsReplicationSource = {
     id: "legacy",
@@ -54,7 +66,29 @@ export function buildReplicationSources(args: {
     originGeneration: args.newOriginGeneration,
   };
 
-  return [legacy, next];
+  // Shard sources come after the gen-1 pair. Reached only when the new source is on, because
+  // split is the precondition for a shard to exist at all. The origin generations come from the
+  // descriptor, which the boot parser already bounds to 2..255 and checks for duplicates; the
+  // service re-checks uniqueness across every source it is given.
+  // The DIRECT dsn, not the app writer dsn. A transaction pooler cannot serve the replication
+  // protocol, and the writer dsn is pooled in a real deployment. Gen-1 keeps the same separation
+  // through its own RUN_REPLICATION_* variables, and the migration loop prefers directUrl too.
+  const shardSources: RunsReplicationSource[] = (args.shards ?? []).map((shard) => ({
+    id: shardSourceId(shard.key),
+    pgConnectionUrl: shard.directUrl ?? shard.url,
+    slotName: shard.replication.slotName,
+    publicationName: shard.replication.publicationName,
+    originGeneration: shard.replication.originGeneration,
+  }));
+
+  return [legacy, next, ...shardSources];
+}
+
+// The replication source id for a shard. It derives the per-source client name and the key the
+// status route probes, so it must be stable and unique across sources. The leader lock is keyed on
+// the slot name, not on this id.
+function shardSourceId(key: string): string {
+  return `shard-${key}`;
 }
 
 /**
@@ -66,24 +100,106 @@ export function buildReplicationSources(args: {
  * rather than ship a fleet-wide under-count.
  */
 export class SplitReplicationMisconfiguredError extends Error {
-  constructor() {
+  constructor(message?: string) {
     super(
-      'RUN_OPS_SPLIT_ENABLED is on but the runs-replication sources[] has no "new" source: ' +
-        "run-ops 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_REPLICATION_RUN_OPS_DATABASE_URL) or turn the split off."
+      message ??
+        'RUN_OPS_SPLIT_ENABLED is on but the runs-replication sources[] has no "new" source: ' +
+          "run-ops 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_REPLICATION_RUN_OPS_DATABASE_URL) or turn the split off."
     );
     this.name = "SplitReplicationMisconfiguredError";
   }
 }
 
+/**
+ * Two sources that share an identity. The descriptor parser checks uniqueness AMONG shards only, so
+ * it cannot see the env-configured legacy and new sources: a shard can collide with either. The
+ * service has its own check, but it throws from the constructor, which the caller reaches only AFTER
+ * it has shut the bootstrap instance down — leaving the process up with NO replication at all, which
+ * is the exact silent under-count this family of errors exists to prevent. So the check runs here,
+ * at the fatal gate, before anything is torn down.
+ */
+class DuplicateReplicationIdentityError extends SplitReplicationMisconfiguredError {
+  constructor(field: string, value: unknown) {
+    super(
+      `the runs-replication sources[] has two sources with the same ${field} "${String(value)}": ` +
+        "two consumers on one WAL stream is a data race, and a shared origin generation defeats the " +
+        "ClickHouse dedup tiebreak. Give every source its own slot, publication and origin generation."
+    );
+    this.name = "DuplicateReplicationIdentityError";
+  }
+}
+
+/**
+ * A configured shard with no replication source of its own. Subclasses the split error on purpose:
+ * the boot catch site tests `instanceof SplitReplicationMisconfiguredError` to reach
+ * process.exit(1), and a shard whose runs never reach ClickHouse must take that same exit.
+ */
+class ShardReplicationMisconfiguredError extends SplitReplicationMisconfiguredError {
+  constructor(shardKey: string) {
+    super(
+      `run-ops shard ${shardKey} is configured but the runs-replication sources[] has no ` +
+        `"${shardSourceId(shardKey)}" source: runs on that shard would not replicate to ` +
+        "ClickHouse, under-counting every ClickHouse-fronted aggregate. Give the shard a " +
+        "replication slot, publication and origin generation, or remove the shard."
+    );
+    this.name = "ShardReplicationMisconfiguredError";
+  }
+}
+
+/**
+ * A shard that replicates but declares no direct dsn. Falling back to its writer dsn is a silent
+ * trap: if that dsn is pooled, the replication client throws inside start(), which is NOT a
+ * SplitReplicationMisconfiguredError, so the process stays up with EVERY source down, legacy
+ * included. Refuse the boot instead.
+ */
+class ShardDirectUrlMissingError extends SplitReplicationMisconfiguredError {
+  constructor(shardKey: string) {
+    super(
+      `run-ops shard ${shardKey} declares replication but no directUrl: logical replication needs a ` +
+        "session-mode connection, which a transaction pooler cannot serve. Give the shard a directUrl " +
+        "pointing at its direct, non-pooled endpoint."
+    );
+    this.name = "ShardDirectUrlMissingError";
+  }
+}
+
 export function assertReplicationCoversSplit(args: {
   splitEnabled: boolean;
   sources: RunsReplicationSource[];
+  /** Every configured shard, aliased ones included. An aliased shard needs no source of its own. */
+  shards?: Array<{ key: string; aliasOf?: "new"; hasDirectUrl?: boolean }>;
 }): void {
-  if (args.splitEnabled && !args.sources.some((s) => s.id === "new")) {
+  if (!args.splitEnabled) {
+    return;
+  }
+  if (!args.sources.some((s) => s.id === "new")) {
     throw new SplitReplicationMisconfiguredError();
   }
+  for (const shard of args.shards ?? []) {
+    // An aliased shard shares its target's database, so the target's slot already carries its WAL.
+    if (shard.aliasOf !== undefined) continue;
+    if (!args.sources.some((s) => s.id === shardSourceId(shard.key))) {
+      throw new ShardReplicationMisconfiguredError(shard.key);
+    }
+    if (shard.hasDirectUrl === false) {
+      throw new ShardDirectUrlMissingError(shard.key);
+    }
+  }
+
+  // Cross-source identity, over EVERY source and not only the shards. A correct two-source
+  // deployment already satisfies this, because two consumers on one WAL slot is a data race that
+  // cannot work. So this adds a loud failure for a configuration that was already broken silently.
+  for (const field of ["id", "slotName", "publicationName", "originGeneration"] as const) {
+    const seen = new Set<unknown>();
+    for (const source of args.sources) {
+      if (seen.has(source[field])) {
+        throw new DuplicateReplicationIdentityError(field, source[field]);
+      }
+      seen.add(source[field]);
+    }
+  }
 }
 
 function initializeRunsReplicationInstance() {
@@ -171,6 +287,20 @@ function initializeRunsReplicationInstance() {
     // 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.
+    // An aliased shard replicates through its target's slot, so only the shards that own their own
+    // database take a source. Coverage is then checked against EVERY descriptor, aliased included.
+    // The schema requires `replication` on every non-aliased descriptor, so the guard below is a
+    // type narrowing and not a policy.
+    const shardReplicationByKey = new Map(
+      env.RUN_OPS_SHARDS.flatMap((d) => (d.replication ? [[d.key, d.replication] as const] : []))
+    );
+    const shardsWithReplication = nonAliasedShards(env.RUN_OPS_SHARDS).flatMap((shard) => {
+      const replication = shardReplicationByKey.get(shard.key);
+      return replication
+        ? [{ key: shard.key, url: shard.url, directUrl: shard.directUrl, replication }]
+        : [];
+    });
+
     isSplitEnabled()
       .then(async (splitEnabled) => {
         const sources = buildReplicationSources({
@@ -184,10 +314,20 @@ function initializeRunsReplicationInstance() {
           newSlotName: env.RUN_REPLICATION_NEW_SLOT_NAME,
           newPublicationName: env.RUN_REPLICATION_NEW_PUBLICATION_NAME,
           newOriginGeneration: env.RUN_REPLICATION_NEW_ORIGIN_GENERATION,
+          shards: shardsWithReplication,
         });
 
-        // Refuse to start replication if split is on but `#new` is not a source.
-        assertReplicationCoversSplit({ splitEnabled, sources });
+        // Refuse to start replication if split is on but `#new` is not a source, or if any shard
+        // that owns its own database has no source of its own.
+        assertReplicationCoversSplit({
+          splitEnabled,
+          sources,
+          shards: env.RUN_OPS_SHARDS.map((d) => ({
+            key: d.key,
+            aliasOf: d.aliasOf,
+            hasDirectUrl: d.directUrl !== undefined,
+          })),
+        });
 
         if (sources.length > 1) {
           // Release the bootstrap instance's eager replication client (Redis + Redlock)

apps/webapp/app/v3/runOpsMigration/controlPlaneCoresidencySentinel.server.test.ts

@@ -34,6 +34,7 @@ describe("assertControlPlaneCoresidencyAdvisory", () => {
     const emit = vi.fn();
     await assertControlPlaneCoresidencyAdvisory({
       ...urls,
+      shards: [],
       expectSplit: false,
       probe: async () => ({ coresident: "true" }),
       emit,
@@ -46,6 +47,7 @@ describe("assertControlPlaneCoresidencyAdvisory", () => {
     await expect(
       assertControlPlaneCoresidencyAdvisory({
         ...urls,
+        shards: [],
         expectSplit: true,
         probe: async () => ({ coresident: "true" }),
         emit: vi.fn(),
@@ -58,6 +60,7 @@ describe("assertControlPlaneCoresidencyAdvisory", () => {
     const emit = vi.fn();
     await assertControlPlaneCoresidencyAdvisory({
       ...urls,
+      shards: [],
       expectSplit: true,
       probe: async () => ({ coresident: "unknown", reason: "denied" }),
       emit,
@@ -71,6 +74,7 @@ describe("assertControlPlaneCoresidencyAdvisory", () => {
     const warn = vi.fn();
     await assertControlPlaneCoresidencyAdvisory({
       ...urls,
+      shards: [],
       expectSplit: true,
       probe: async () => {
         throw new Error("probe blew up");
@@ -86,9 +90,12 @@ describe("assertControlPlaneCoresidencyAdvisory", () => {
     const emit = vi.fn();
     const probe = vi.fn();
     await assertControlPlaneCoresidencyAdvisory({
-      legacyUrl: undefined,
+      // "" and not undefined: ?? only guards nullish, so undefined would read the ambient
+      // RUN_OPS_LEGACY_DATABASE_URL and this test would depend on the developer's .env.
+      legacyUrl: "",
       controlPlaneUrl: "postgres://cp",
       expectSplit: true,
+      shards: [],
       probe,
       emit,
       log: noopLog,
@@ -97,3 +104,120 @@ describe("assertControlPlaneCoresidencyAdvisory", () => {
     expect(emit).not.toHaveBeenCalled();
   });
 });
+
+describe("assertControlPlaneCoresidencyAdvisory at N shards", () => {
+  const urls = { legacyUrl: "postgres://legacy", controlPlaneUrl: "postgres://cp" };
+  const shardA = { key: "a", url: "postgres://shard-a" };
+  const shardB = { key: "b", url: "postgres://shard-b" };
+
+  it("emits the legacy verdict with NO shard key, so today's series is unchanged", async () => {
+    const emit = vi.fn();
+    await assertControlPlaneCoresidencyAdvisory({
+      ...urls,
+      expectSplit: false,
+      shards: [],
+      probe: async () => ({ coresident: "false" }),
+      emit,
+      log: noopLog,
+    });
+    expect(emit).toHaveBeenCalledTimes(1);
+    expect(emit).toHaveBeenCalledWith("false");
+  });
+
+  it("emits one tagged verdict per shard, plus the untagged legacy verdict", async () => {
+    const emit = vi.fn();
+    await assertControlPlaneCoresidencyAdvisory({
+      ...urls,
+      expectSplit: false,
+      shards: [shardA, shardB],
+      probe: async () => ({ coresident: "false" }),
+      emit,
+      log: noopLog,
+    });
+    expect(emit).toHaveBeenCalledTimes(3);
+    expect(emit).toHaveBeenCalledWith("false");
+    expect(emit).toHaveBeenCalledWith("false", "a");
+    expect(emit).toHaveBeenCalledWith("false", "b");
+  });
+
+  it("probes each shard against the control plane", async () => {
+    const probe = vi.fn().mockResolvedValue({ coresident: "false" });
+    await assertControlPlaneCoresidencyAdvisory({
+      ...urls,
+      expectSplit: false,
+      shards: [shardA],
+      probe,
+      emit: vi.fn(),
+      log: noopLog,
+    });
+    expect(probe).toHaveBeenCalledWith("postgres://legacy", "postgres://cp", expect.anything());
+    expect(probe).toHaveBeenCalledWith("postgres://shard-a", "postgres://cp", expect.anything());
+  });
+
+  it("names the offending shard when enforcement is opted in and a shard is co-resident", async () => {
+    await expect(
+      assertControlPlaneCoresidencyAdvisory({
+        ...urls,
+        expectSplit: true,
+        shards: [shardA],
+        probe: async (url: string) =>
+          url === "postgres://shard-a"
+            ? ({ coresident: "true", reason: "same db" } as const)
+            : ({ coresident: "false" } as const),
+        emit: vi.fn(),
+        log: noopLog,
+      })
+    ).rejects.toThrow(/shard a/i);
+  });
+
+  it("emits every store before it throws, so no store loses its metric", async () => {
+    const emit = vi.fn();
+    await expect(
+      assertControlPlaneCoresidencyAdvisory({
+        ...urls,
+        expectSplit: true,
+        shards: [shardA, shardB],
+        probe: async (url: string) =>
+          url === "postgres://shard-a"
+            ? ({ coresident: "true", reason: "same db" } as const)
+            : ({ coresident: "false" } as const),
+        emit,
+        log: noopLog,
+      })
+    ).rejects.toThrow();
+    expect(emit).toHaveBeenCalledTimes(3);
+    expect(emit).toHaveBeenCalledWith("false", "b");
+  });
+
+  it("degrades one shard's throwing probe to unknown and still reports the others", async () => {
+    const emit = vi.fn();
+    await assertControlPlaneCoresidencyAdvisory({
+      ...urls,
+      expectSplit: true,
+      shards: [shardA, shardB],
+      probe: async (url: string) => {
+        if (url === "postgres://shard-a") throw new Error("probe blew up");
+        return { coresident: "false" } as const;
+      },
+      emit,
+      log: { info: () => {}, warn: () => {} },
+    });
+    expect(emit).toHaveBeenCalledWith("unknown", "a");
+    expect(emit).toHaveBeenCalledWith("false", "b");
+  });
+
+  it("still probes the shards when there is no legacy DSN", async () => {
+    const emit = vi.fn();
+    await assertControlPlaneCoresidencyAdvisory({
+      legacyUrl: "",
+      controlPlaneUrl: "postgres://cp",
+      expectSplit: false,
+      shards: [shardA],
+      probe: async () => ({ coresident: "false" }),
+      emit,
+      log: noopLog,
+    });
+    expect(emit).toHaveBeenCalledTimes(1);
+    expect(emit).toHaveBeenCalledWith("false", "a");
+  });
+});