feat(webapp): build one run-ops client pair per shard descriptor

selectRunOpsTopology gains a shard loop and returns a keyed shard map. An
aliasOf:"new" descriptor reuses the new store's clients by reference and opens
no pool. Each real shard gets its own resilience budget and the new-role pool
knobs merged with its per-shard overrides.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: d-cs

apps/webapp/app/db.server.ts

@@ -32,6 +32,7 @@ import {
 } from "./v3/runOpsMigration/splitMode.server";
 import { computeRunOpsSplitReadEnabled } from "./v3/runOpsMigration/runOpsSplitReadGate";
 import { resolveRunOpsPoolKnobs } from "./v3/runOpsPoolKnobs.server";
+import { resolveShardResilience } from "./v3/transactionResilience.server";
 import { assertControlPlaneCoresidencyAdvisory } from "./v3/runOpsMigration/controlPlaneCoresidencySentinel.server";
 import { DATASOURCE_CONTEXT_KEY, startActiveSpan } from "./v3/tracer.server";
 import {
@@ -276,10 +277,19 @@ export const webhookReplica: WebhookReplicaDatabase = singleton("webhookReplica"
 
 type RunOpsClients = { writer: PrismaClient; replica: PrismaReplicaClient };
 type NewRunOpsClients = { writer: RunOpsPrismaClient; replica: RunOpsPrismaClient };
+export type ShardTopologyDescriptor = {
+  key: string;
+  url?: string;
+  replicaUrl?: string;
+  aliasOf?: "new";
+};
 export type RunOpsTopology = {
   newRunOps: NewRunOpsClients;
   legacyRunOps: RunOpsClients;
   controlPlane: RunOpsClients;
+  // One client pair per gen-2 shard descriptor. Empty unless RUN_OPS_SHARDS is configured. An
+  // aliasOf:"new" descriptor maps to the newRunOps pair BY REFERENCE (no new pool).
+  shards: Map<string, NewRunOpsClients>;
 };
 export type SelectRunOpsTopologyConfig = {
   splitEnabled: boolean;
@@ -289,6 +299,7 @@ export type SelectRunOpsTopologyConfig = {
   newReplicaUrl?: string;
   // When true, legacy reuses the control-plane client instead of opening its own pool. Defaults to false.
   legacySharesControlPlane?: boolean;
+  shards?: ShardTopologyDescriptor[];
 };
 export type RunOpsClientBuilders = {
   controlPlane: RunOpsClients;
@@ -298,6 +309,10 @@ export type RunOpsClientBuilders = {
   // RunOpsPrismaClient double-cast needed): the legacy DB carries the full control-plane schema.
   buildLegacyWriter: (url: string, clientType: string) => PrismaClient;
   buildLegacyReplica: (url: string, clientType: string) => PrismaReplicaClient;
+  // Receive the whole descriptor so the singleton can resolve per-shard knobs and resilience by key.
+  // Optional so the existing test literals (which build no shards) need no change.
+  buildShardWriter?: (shard: ShardTopologyDescriptor) => RunOpsPrismaClient;
+  buildShardReplica?: (shard: ShardTopologyDescriptor) => RunOpsPrismaClient;
 };
 
 // Pure run-ops client selector. No env, no isSplitEnabled() — those
@@ -316,11 +331,11 @@ export function selectRunOpsTopology(
   };
 
   if (!config.splitEnabled) {
-    return { newRunOps: cpFallback, legacyRunOps: controlPlane, controlPlane };
+    return { newRunOps: cpFallback, legacyRunOps: controlPlane, controlPlane, shards: new Map() };
   }
 
   if (!config.legacyUrl || !config.newUrl) {
-    return { newRunOps: cpFallback, legacyRunOps: controlPlane, controlPlane };
+    return { newRunOps: cpFallback, legacyRunOps: controlPlane, controlPlane, shards: new Map() };
   }
 
   // Same-DB legacy reuses the control-plane pool; only build a separate pool once the DSNs diverge.
@@ -339,12 +354,28 @@ export function selectRunOpsTopology(
   const newReplica: RunOpsPrismaClient = config.newReplicaUrl
     ? builders.buildNewReplica(config.newReplicaUrl, "run-ops-replica")
     : newWriter;
+  const newRunOps: NewRunOpsClients = { writer: newWriter, replica: newReplica };
+
+  const shards = new Map<string, NewRunOpsClients>();
+  for (const shard of config.shards ?? []) {
+    if (shard.aliasOf === "new") {
+      // Aliased: share the new store's clients by reference. No builder, no new pool — the soak path.
+      shards.set(shard.key, newRunOps);
+      continue;
+    }
+    if (!shard.url || !builders.buildShardWriter || !builders.buildShardReplica) {
+      throw new Error(
+        `selectRunOpsTopology: shard "${shard.key}" needs a url and shard builders when not aliased`
+      );
+    }
+    const shardWriter = builders.buildShardWriter(shard);
+    const shardReplica: RunOpsPrismaClient = shard.replicaUrl
+      ? builders.buildShardReplica(shard)
+      : shardWriter;
+    shards.set(shard.key, { writer: shardWriter, replica: shardReplica });
+  }
 
-  return {
-    newRunOps: { writer: newWriter, replica: newReplica },
-    legacyRunOps,
-    controlPlane,
-  };
+  return { newRunOps, legacyRunOps, controlPlane, shards };
 }
 
 // The env-bound run-ops topology singleton. The split decision uses
@@ -378,6 +409,7 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => {
   }
 
   const newPoolKnobs = resolveRunOpsPoolKnobs("new");
+  const shardDescriptorsByKey = new Map(env.RUN_OPS_SHARDS.map((d) => [d.key, d]));
 
   return selectRunOpsTopology(
     {
@@ -387,6 +419,12 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => {
       newUrl,
       newReplicaUrl: env.RUN_OPS_DATABASE_READ_REPLICA_URL,
       legacySharesControlPlane,
+      shards: env.RUN_OPS_SHARDS.map((d) => ({
+        key: d.key,
+        url: d.url,
+        replicaUrl: d.replicaUrl,
+        aliasOf: d.aliasOf,
+      })),
     },
     {
       controlPlane: { writer: prisma, replica: $replica },
@@ -461,6 +499,50 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => {
             )
           )
         ),
+      // A gen-2 shard is a dedicated run-ops DB, so it mirrors buildNewWriter/buildNewReplica: same
+      // client class, same wrapper stack, its OWN resilience budget, and the "new"-role pool knobs
+      // merged with the descriptor's per-shard overrides. Shards share the run-ops datasource tag.
+      buildShardWriter: (shard) => {
+        const descriptor = shardDescriptorsByKey.get(shard.key);
+        const knobs = resolveRunOpsPoolKnobs("new", descriptor?.knobs);
+        return registerTransactionResilience(
+          captureInfraErrorsRunOps(
+            tagDatasourceRunOps(
+              "run-ops-writer",
+              buildRunOpsClient({
+                url: shard.url!,
+                clientType: `run-ops-shard-${shard.key}-writer`,
+                role: "writer",
+                connectionLimit: knobs.connectionLimit,
+                poolTimeout: knobs.writerPoolTimeout,
+                connectTimeout: knobs.writerConnectionTimeout,
+                useDriverAdapter: knobs.writerDriverAdapter,
+              })
+            )
+          ),
+          resolveShardResilience(shard.key, descriptor?.knobs)
+        );
+      },
+      buildShardReplica: (shard) => {
+        const descriptor = shardDescriptorsByKey.get(shard.key);
+        const knobs = resolveRunOpsPoolKnobs("new", descriptor?.knobs);
+        return markReadReplicaClient(
+          captureInfraErrorsRunOps(
+            tagDatasourceRunOps(
+              "run-ops-replica",
+              buildRunOpsClient({
+                url: shard.replicaUrl!,
+                clientType: `run-ops-shard-${shard.key}-replica`,
+                role: "replica",
+                connectionLimit: knobs.replicaConnectionLimit,
+                poolTimeout: knobs.replicaPoolTimeout,
+                connectTimeout: knobs.replicaConnectionTimeout,
+                useDriverAdapter: knobs.replicaDriverAdapter,
+              })
+            )
+          )
+        );
+      },
     }
   );
 });

apps/webapp/app/v3/transactionResilience.server.ts

@@ -67,6 +67,44 @@ export const runOpsTransactionResilience = resolveTransactionResilience("run-ops
   budgetBurst: env.RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BUDGET_BURST,
 });
 
+// A gen-2 shard's resilience. Defaults to the RUN_OPS_DATABASE_TRANSACTION_* values (so a shard with
+// no overrides matches the gen-1 new store), then applies the descriptor's per-shard overrides. Each
+// call builds its OWN budget, so a storm on one shard cannot drain another's.
+export function resolveShardResilience(
+  key: string,
+  overrides?: {
+    transactionMaxWaitMs?: number;
+    transactionStartRetryEnabled?: boolean;
+    transactionStartRetryMaxAttempts?: number;
+    transactionStartRetryBackoffMinMs?: number;
+    transactionStartRetryBackoffMaxMs?: number;
+    transactionStartRetryBudgetPerSec?: number;
+    transactionStartRetryBudgetBurst?: number;
+  }
+): TransactionResilienceConfig {
+  return resolveTransactionResilience(`run-ops-shard-${key}`, {
+    maxWaitMs: overrides?.transactionMaxWaitMs ?? env.RUN_OPS_DATABASE_TRANSACTION_MAX_WAIT_MS,
+    enabled:
+      overrides?.transactionStartRetryEnabled ??
+      env.RUN_OPS_DATABASE_TRANSACTION_START_RETRY_ENABLED,
+    maxAttempts:
+      overrides?.transactionStartRetryMaxAttempts ??
+      env.RUN_OPS_DATABASE_TRANSACTION_START_RETRY_MAX_ATTEMPTS,
+    backoffMinMs:
+      overrides?.transactionStartRetryBackoffMinMs ??
+      env.RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MIN_MS,
+    backoffMaxMs:
+      overrides?.transactionStartRetryBackoffMaxMs ??
+      env.RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MAX_MS,
+    budgetPerSec:
+      overrides?.transactionStartRetryBudgetPerSec ??
+      env.RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BUDGET_PER_SEC,
+    budgetBurst:
+      overrides?.transactionStartRetryBudgetBurst ??
+      env.RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BUDGET_BURST,
+  });
+}
+
 export const runOpsLegacyTransactionResilience = resolveTransactionResilience("run-ops-legacy", {
   maxWaitMs: env.RUN_OPS_LEGACY_DATABASE_TRANSACTION_MAX_WAIT_MS,
   enabled: env.RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_ENABLED,

apps/webapp/test/runOpsDbTopology.test.ts

@@ -142,6 +142,57 @@ describe("selectRunOpsTopology (pure)", () => {
     expect(topo.legacyRunOps.replica).toBe(legacyWriter);
     expect(buildLegacyReplica).not.toHaveBeenCalled();
   });
+
+  const baseSplit = {
+    splitEnabled: true,
+    legacyUrl: "postgres://legacy",
+    newUrl: "postgres://new",
+  };
+  const baseBuilders = () => ({
+    controlPlane: cp,
+    buildNewWriter: vi.fn().mockReturnValue({ tag: "nw" } as any),
+    buildNewReplica: vi.fn().mockReturnValue({ tag: "nr" } as any),
+    buildLegacyWriter: vi.fn().mockReturnValue({ tag: "lw" } as any),
+    buildLegacyReplica: vi.fn().mockReturnValue({ tag: "lr" } as any),
+  });
+
+  it("no descriptors: the shards map is empty", () => {
+    const topo = selectRunOpsTopology(baseSplit, baseBuilders());
+    expect(topo.shards.size).toBe(0);
+  });
+
+  it("two descriptors: two shard client pairs, each built once", () => {
+    const buildShardWriter = vi.fn((s: any) => ({ tag: `w:${s.key}` }) as any);
+    const buildShardReplica = vi.fn((s: any) => ({ tag: `r:${s.key}` }) as any);
+    const topo = selectRunOpsTopology(
+      {
+        ...baseSplit,
+        shards: [
+          { key: "a", url: "postgres://a", replicaUrl: "postgres://a-r" },
+          { key: "b", url: "postgres://b" },
+        ],
+      },
+      { ...baseBuilders(), buildShardWriter, buildShardReplica }
+    );
+    expect(topo.shards.size).toBe(2);
+    expect(topo.shards.get("a")!.writer).toEqual({ tag: "w:a" });
+    // b has no replicaUrl, so its replica falls back to its writer (buildShardReplica not called for b).
+    expect(topo.shards.get("b")!.replica).toEqual({ tag: "w:b" });
+    expect(buildShardWriter).toHaveBeenCalledTimes(2);
+    expect(buildShardReplica).toHaveBeenCalledTimes(1);
+  });
+
+  it("an alias descriptor reuses newRunOps by reference and calls no shard builder", () => {
+    const buildShardWriter = vi.fn();
+    const buildShardReplica = vi.fn();
+    const topo = selectRunOpsTopology(
+      { ...baseSplit, shards: [{ key: "a", aliasOf: "new" }] },
+      { ...baseBuilders(), buildShardWriter, buildShardReplica }
+    );
+    expect(topo.shards.get("a")).toBe(topo.newRunOps);
+    expect(buildShardWriter).not.toHaveBeenCalled();
+    expect(buildShardReplica).not.toHaveBeenCalled();
+  });
 });
 
 describe("sameDatabaseTarget", () => {