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

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> {