fix(webapp): scan every run-ops store for the batches list (#4806)

## Summary

Batches created on a run-ops store other than the two the list reads
were missing from the Batches page. No error, nothing logged: the page
just showed fewer batches than exist. This is only reachable once
additional run-ops stores are configured, so nothing changes for anyone
today.

## Fix

The list scanned exactly two databases and merged them by keyset. It now
covers one leg per configured store, in ascending precedence order, all
issued together.

The existing keyset merge generalises without change. Every leg runs the
same query, with the same cursor predicate, ordering and over-fetch, so
a row's rank within its own leg is never worse than its global rank, and
the merged first page is still the true first page. That argument holds
for any number of legs, not just two.

The empty-state check keeps its existing sequential pair, since a
project with no batches is the common case for that path, then issues
the remaining checks in a single round trip.

A store that declares itself an alias of another shares its client by
reference, so it contributes no leg. Scanning it would query the same
database twice for rows the other leg already returned. This matches how
the routing store and the boot checks treat an alias.

The fan-out deliberately fails the page if any store is unreachable,
rather than returning a short page. A tolerant merge would recreate the
same silent absence this change removes, with a wider blast radius.

## Verification

Covered by container tests against real databases: gen-1, legacy and
additional stores merged into one ordered page, paging forward and back
across a boundary that spans stores, and the empty-state check.

Also verified end to end against a live environment with a real corpus:
the missing rows reproduce with the new leg removed and appear correctly
with it present, ordering interleaves across stores as expected, paging
across a store boundary loses and repeats nothing, and the page is
byte-identical to before when no extra store is configured.

Merge precedence is pinned by its own test: one id seeded on two stores,
asserting the higher-authority copy is the one shown. Verified by
mutation, since a union-only test passes regardless of leg order.

## Boot interlocks

Two related boot checks changed alongside the read path, since
configuring an extra store is what makes them reachable.

A store configured while split reads are disabled is dropped in silence:
no client is built, no leg is added, and rows already resident there
disappear from every list with no error. The other two ways the split
ends up disabled already refuse to start; this closes the one that did
not, and names the stores it is refusing. A store that declares itself
an alias of another owns no database, so it is exempt.

The distinct-database probe fails closed, which meant one store being
briefly unreachable collapsed the deployment to single-DB and then
refused the boot entirely. Each target now gets a bounded number of
attempts with a short backoff before the probe gives up. Failing closed
is unchanged once that budget is exhausted, and a genuine duplicate is
still a final answer that is never retried.

Author: d-cs

apps/webapp/app/db.server.ts

@@ -28,6 +28,7 @@ import { singleton } from "./utils/singleton";
 import { registerDatabaseMetricsSource } from "./utils/databaseMetrics.server";
 import {
   isSplitEnabled,
+  assertShardsRequireSplit,
   assertSplitRealtimeInterlock,
 } from "./v3/runOpsMigration/splitMode.server";
 import { computeRunOpsSplitReadEnabled } from "./v3/runOpsMigration/runOpsSplitReadGate";
@@ -617,6 +618,12 @@ export const runOpsSplitReadEnabled: boolean = computeRunOpsSplitReadEnabled({
 // interlock). Async, so it cannot live in the synchronous singleton factory — called
 // fire-and-forget from the eager-boot path (routing is wired synchronously at module load).
 export async function assertRunOpsSplitSentinel(): Promise<void> {
+  // Shard interlock first: shard clients are only built on the split-on arm, so this case has to be
+  // checked BEFORE the split-off early return below, which would otherwise skip it in silence.
+  assertShardsRequireSplit({
+    splitFlagEnabled: env.RUN_OPS_SPLIT_ENABLED,
+    shards: env.RUN_OPS_SHARDS,
+  });
   if (!env.RUN_OPS_SPLIT_ENABLED) return;
   // Realtime interlock (synchronous): Electric replicates only from the control-plane
   // DB, so split-on without the native realtime backend leaves NEW-resident runs

apps/webapp/app/presenters/v3/BatchListPresenter.server.ts

@@ -75,6 +75,7 @@ export class BatchListPresenter extends BasePresenter {
       runOpsNew?: RunOpsPrismaClient; // new run-ops client (run-ops brand ⇒ guard classifies as runops)
       runOpsLegacyReplica?: RunOpsPrismaClient; // legacy run-ops READ REPLICA only — never the legacy primary
       controlPlaneReplica?: PrismaClientOrTransaction; // control-plane DB (for project)
+      shardReplicas?: ReadonlyArray<{ key: string; replica: RunOpsPrismaClient }>;
       splitEnabled?: boolean; // resolved boot constant
     }
   ) {
@@ -110,9 +111,11 @@ export class BatchListPresenter extends BasePresenter {
     // unsound across the residency split: legacy cuid ids ("c…") sort ABOVE new run-ops ids ("0…")
     // under id order, so a new-only page can hide pre-flip legacy batches that belong ahead of it.
     // Ordering is by createdAt (id tiebreak), which is chronologically correct across both schemes.
-    const [newRows, legacyRows] = await Promise.all([
+    const shardReplicas = this.readRoute.shardReplicas ?? [];
+    const [newRows, legacyRows, ...shardRows] = await Promise.all([
       scan(this.readRoute.runOpsNew ?? passthrough),
       scan(this.readRoute.runOpsLegacyReplica ?? passthrough),
+      ...shardReplicas.map((shard) => scan(shard.replica)),
     ]);
 
     // De-dupe by id (new wins), re-sort under the page's keyset order, re-apply the over-fetch LIMIT.
@@ -125,6 +128,11 @@ export class BatchListPresenter extends BasePresenter {
         byId.set(row.id, row);
       }
     }
+    for (const rows of shardRows) {
+      for (const row of rows) {
+        byId.set(row.id, row);
+      }
+    }
 
     // forward => newest-first (createdAt DESC), backward => oldest-first (ASC); id is the stable
     // tiebreak (ASCII codepoint, NEVER localeCompare).
@@ -167,7 +175,23 @@ export class BatchListPresenter extends BasePresenter {
     ).batchTaskRun.findFirst({
       where: { runtimeEnvironmentId: environmentId },
     });
-    return Boolean(onLegacy);
+    if (onLegacy) {
+      return true;
+    }
+
+    const shardReplicas = this.readRoute.shardReplicas ?? [];
+    if (shardReplicas.length === 0) {
+      return false;
+    }
+
+    const onShards = await Promise.all(
+      shardReplicas.map((shard) =>
+        shard.replica.batchTaskRun.findFirst({
+          where: { runtimeEnvironmentId: environmentId },
+        })
+      )
+    );
+    return onShards.some(Boolean);
   }
 
   public async call({

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.batches/route.tsx

@@ -56,6 +56,7 @@ import {
   runOpsSplitReadEnabled,
   type PrismaClientOrTransaction,
 } from "~/db.server";
+import { runOpsNonAliasedShardReplicas } from "~/v3/runOpsMigration/shardHandles.server";
 import {
   docsPath,
   EnvironmentParamSchema,
@@ -104,6 +105,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     runOpsNew: runOpsNewReplicaClient,
     runOpsLegacyReplica: runOpsLegacyReplicaClient,
     controlPlaneReplica: $replica as unknown as PrismaClientOrTransaction,
+    shardReplicas: runOpsNonAliasedShardReplicas,
     splitEnabled: runOpsSplitReadEnabled,
   });
   const list = await presenter.call({

apps/webapp/app/v3/runOpsMigration/distinctDbSentinel.server.ts

@@ -64,6 +64,48 @@ export async function probeControlPlaneCoresidency(
 
 export type DistinctTarget = { id: string; url: string };
 
+/** Injection seam for the retry tests: no containers, no real waiting. */
+export type DistinctProbeOptions = {
+  logger?: { warn: (msg: string, meta?: Record<string, unknown>) => void };
+  readFingerprint?: (url: string) => Promise<DatabaseFingerprint>;
+  /** Total attempts per target, including the first. Bounded so boot latency stays bounded. */
+  attempts?: number;
+  sleep?: (ms: number) => Promise<void>;
+};
+
+const DEFAULT_PROBE_ATTEMPTS = 3;
+const RETRY_BASE_DELAY_MS = 250;
+
+const defaultSleep = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));
+
+/**
+ * Read one fingerprint, retrying a bounded number of times.
+ *
+ * The probe fails CLOSED, and that must not change: "distinct" is a positive claim a failed probe
+ * cannot support. But failing closed on the first blip means one shard being briefly unreachable
+ * collapses the deployment to single-DB, and the boot interlock then refuses the boot for the whole
+ * fleet. A transient error deserves a retry; a persistent one still fails closed, just later.
+ */
+async function readFingerprintWithRetry(
+  url: string,
+  read: (url: string) => Promise<DatabaseFingerprint>,
+  attempts: number,
+  sleep: (ms: number) => Promise<void>
+): Promise<DatabaseFingerprint> {
+  let lastError: unknown;
+  for (let attempt = 1; attempt <= attempts; attempt++) {
+    try {
+      return await read(url);
+    } catch (error) {
+      lastError = error;
+      if (attempt < attempts) {
+        await sleep(RETRY_BASE_DELAY_MS * attempt);
+      }
+    }
+  }
+  throw lastError;
+}
+
 /**
  * Set uniqueness over every store that owns its own database. Fail-closed: a probe that cannot
  * answer returns NOT distinct, because "distinct" is a positive claim a failed probe cannot support.
@@ -77,14 +119,22 @@ export type DistinctTarget = { id: string; url: string };
  */
 export async function probeDistinctStores(
   targets: DistinctTarget[],
-  opts?: { logger?: { warn: (msg: string, meta?: Record<string, unknown>) => void } }
+  opts?: DistinctProbeOptions
 ): Promise<{ distinct: true } | { distinct: false; reason: string }> {
   if (targets.length < 2) {
     return { distinct: true };
   }
 
+  const read = opts?.readFingerprint ?? readDatabaseFingerprint;
+  const attempts = opts?.attempts ?? DEFAULT_PROBE_ATTEMPTS;
+  const sleep = opts?.sleep ?? defaultSleep;
+
   try {
-    const fingerprints = await Promise.all(targets.map((t) => readDatabaseFingerprint(t.url)));
+    // Retry per TARGET, not around the whole set: one slow shard must not re-probe the stores that
+    // already answered. A duplicate verdict below is final and is never retried.
+    const fingerprints = await Promise.all(
+      targets.map((t) => readFingerprintWithRetry(t.url, read, attempts, sleep))
+    );
 
     const seen = new Map<string, string>();
     for (const [index, target] of targets.entries()) {
@@ -104,7 +154,9 @@ export async function probeDistinctStores(
 
     return { distinct: true };
   } catch (error) {
-    const reason = `distinct-db sentinel probe failed; failing closed (single-DB). ${String(error)}`;
+    const reason =
+      `distinct-db sentinel probe failed after ${opts?.attempts ?? DEFAULT_PROBE_ATTEMPTS} ` +
+      `attempt(s); failing closed (single-DB). ${String(error)}`;
     opts?.logger?.warn(reason, { error });
     return { distinct: false, reason };
   }
@@ -115,7 +167,7 @@ export async function probeDistinctStores(
 export async function probeDistinctDatabases(
   legacyUrl: string,
   newUrl: string,
-  opts?: { logger?: { warn: (msg: string, meta?: Record<string, unknown>) => void } }
+  opts?: DistinctProbeOptions
 ): Promise<{ distinct: true } | { distinct: false; reason: string }> {
   return probeDistinctStores(
     [

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

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "vitest";
-import { buildShardHandleMaps } from "./shardHandles.server";
+import { buildShardHandleMaps, nonAliasedShardReplicas } from "./shardHandles.server";
 
 // Two distinct sentinels per shard: the maps must not cross writer and replica.
 function handle(key: string) {
@@ -35,3 +35,26 @@ describe("buildShardHandleMaps", () => {
     expect(replicas.get("a")).not.toEqual({ tag: "a-writer" });
   });
 });
+
+describe("nonAliasedShardReplicas", () => {
+  it("yields an empty list when no shard is configured", () => {
+    expect(nonAliasedShardReplicas([])).toEqual([]);
+  });
+
+  it("keeps the configured order and carries each shard's replica", () => {
+    expect(nonAliasedShardReplicas([handle("b"), handle("a")])).toEqual([
+      { key: "b", replica: { tag: "b-replica" } },
+      { key: "a", replica: { tag: "a-replica" } },
+    ]);
+  });
+
+  it("drops a shard that declares aliasOf", () => {
+    expect(nonAliasedShardReplicas([{ ...handle("a"), aliasOf: "new" }, handle("b")])).toEqual([
+      { key: "b", replica: { tag: "b-replica" } },
+    ]);
+  });
+
+  it("never carries a writer in place of a replica", () => {
+    expect(nonAliasedShardReplicas([handle("a")])[0]?.replica).not.toEqual({ tag: "a-writer" });
+  });
+});

apps/webapp/app/v3/runOpsMigration/shardHandles.server.ts

@@ -5,14 +5,16 @@
  * is what keeps every gen-2 arm unreachable today.
  */
 import type { PrismaClient } from "@trigger.dev/database";
+import type { RunOpsPrismaClient } from "@internal/run-ops-database";
 import type { ShardKey } from "@trigger.dev/core/v3/isomorphic";
 import type { PrismaReplicaClient } from "~/db.server";
 import { runOpsShardHandles } from "~/db.server";
 
 type ShardHandle = {
   key: string;
-  writer: unknown;
-  replica: unknown;
+  writer: RunOpsPrismaClient;
+  replica: RunOpsPrismaClient;
+  aliasOf?: string;
 };
 
 export function buildShardHandleMaps(handles: ShardHandle[]): {
@@ -22,8 +24,8 @@ export function buildShardHandleMaps(handles: ShardHandle[]): {
   const replicas = new Map<ShardKey, PrismaReplicaClient>();
   const writers = new Map<ShardKey, PrismaClient>();
   for (const handle of handles) {
-    replicas.set(handle.key, handle.replica as PrismaReplicaClient);
-    writers.set(handle.key, handle.writer as PrismaClient);
+    replicas.set(handle.key, handle.replica as unknown as PrismaReplicaClient);
+    writers.set(handle.key, handle.writer as unknown as PrismaClient);
   }
   return { replicas, writers };
 }
@@ -40,7 +42,17 @@ function resolveShardHandles(): ShardHandle[] {
   }
 }
 
-const maps = buildShardHandleMaps(resolveShardHandles());
+export function nonAliasedShardReplicas<TClient>(
+  handles: ReadonlyArray<{ key: string; replica: TClient; aliasOf?: string }>
+): ReadonlyArray<{ key: string; replica: TClient }> {
+  return handles
+    .filter((handle) => handle.aliasOf === undefined)
+    .map((handle) => ({ key: handle.key, replica: handle.replica }));
+}
+
+const handles = resolveShardHandles();
+const maps = buildShardHandleMaps(handles);
 
 export const runOpsShardReplicas = maps.replicas;
 export const runOpsShardWriters = maps.writers;
+export const runOpsNonAliasedShardReplicas = nonAliasedShardReplicas(handles);

apps/webapp/app/v3/runOpsMigration/splitMode.server.ts

@@ -7,7 +7,11 @@
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { probeDistinctStores as defaultProbe } from "./distinctDbSentinel.server";
-import { nonAliasedShards, type ShardTarget } from "~/v3/runOpsShards.server";
+import {
+  nonAliasedShards,
+  type RunOpsShardDescriptor,
+  type ShardTarget,
+} from "~/v3/runOpsShards.server";
 
 export type SplitModeConfig = {
   flagEnabled: boolean;
@@ -73,6 +77,35 @@ export function assertSplitRealtimeInterlock(config: SplitRealtimeInterlockConfi
   }
 }
 
+export type ShardsRequireSplitConfig = {
+  splitFlagEnabled: boolean;
+  /** Raw descriptors. The alias exemption is applied here so no call site can forget it. */
+  shards: RunOpsShardDescriptor[];
+};
+
+/**
+ * Boot-time shard interlock (pure predicate). Shard clients are only built on the split-on arm of
+ * `selectRunOpsTopology`, so a shard configured while the split flag is off is dropped in silence:
+ * no client, no fan-out leg, and any row already resident on that database vanishes from every
+ * list with no error. The other two ways split can end up disabled (URLs missing, sentinel not
+ * distinct) already refuse to boot; this closes the one that does not.
+ */
+export function assertShardsRequireSplit(config: ShardsRequireSplitConfig): void {
+  if (config.splitFlagEnabled) {
+    return;
+  }
+  // An aliased shard owns no database: it shares its target's client by reference, so its rows are
+  // still read with the split off and nothing is dropped. Exempt here exactly as it is exempt from
+  // the distinctness sentinel, the coresidency loop and replication.
+  const owning = nonAliasedShards(config.shards).map((shard) => shard.key);
+  if (owning.length === 0) {
+    return;
+  }
+  throw new Error(
+    `RUN_OPS_SHARDS configures shard(s) ${owning.join(", ")} but RUN_OPS_SPLIT_ENABLED is off, so no shard client is built and rows on those databases would be silently missing; refusing to start.`
+  );
+}
+
 let cached: Promise<boolean> | undefined;
 
 export function isSplitEnabled(): Promise<boolean> {