feat(webapp): gen-2 shard arms in read-through

Route read-through by `resolveShard` instead of the binary residency
classifier. A gen-2 id takes one read on its own shard's replica and
probes no other store; a gen-1 v1 id still reads new only.

Callers now declare `idKind`, because a cuid gives no way to tell a run
id from a waitpoint id and the two must route differently. A legacy
run id reads the legacy replica only, since there is no cuid run
migration and the new-store probe cannot find it. A cuid waitpoint
keeps the new-first pair probe, which is load-bearing because a cuid
waitpoint can be co-located with its run on the new store.

`ReadThroughResult` now carries `found` structurally. `source` is an
open-ended union once shards exist, so a consumer testing found-ness by
listing hit sources would read a gen-2 hit as a miss; discriminating on
`found` turns that class of bug into a compile error.

An id resolving to an unconfigured shard key logs an error and returns
not-found rather than throwing. These ids arrive from callers (a URL
param on the waitpoint route) and any base32hex core plus `[a-z0-9]`
plus "2" parses as gen-2, so a throw would be a 500 any client could
induce by guessing a shard char. Throwing stays correct on the router
path, where ids are minted rather than received.

The read-your-writes fallback for a gen-2 waitpoint now reads that
shard's own writer. The gen-1 new writer is a different database, so
reading it would miss and silently disable read-your-writes for the
freshly minted token that fallback exists to serve.

Inert while RUN_OPS_SHARDS is unset: the shard maps are empty, so every
gen-2 arm is unreachable.

Author: d-cs

apps/webapp/app/runEngine/concerns/resolveWaitpointThroughReadThrough.server.ts

@@ -1,16 +1,23 @@
+import { resolveShard, type ShardKey } from "@trigger.dev/core/v3/isomorphic";
 import type { PrismaReplicaClient } from "~/db.server";
 import {
   runOpsLegacyReplica as defaultLegacyReplica,
   runOpsNewPrisma as defaultNewPrimary,
   runOpsNewReplica as defaultNewClient,
   runOpsSplitReadEnabled as defaultSplitReadEnabled,
 } from "~/db.server";
+import {
+  runOpsShardReplicas as defaultShardReplicas,
+  runOpsShardWriters as defaultShardWriters,
+} from "~/v3/runOpsMigration/shardHandles.server";
 import { readThroughRun } from "~/v3/runOpsMigration/readThrough.server";
 
 type ResolveWaitpointDeps = {
   newClient?: PrismaReplicaClient;
   legacyReplica?: PrismaReplicaClient;
   newPrimary?: PrismaReplicaClient;
+  shardReplicas?: ReadonlyMap<ShardKey, PrismaReplicaClient>;
+  shardWriters?: ReadonlyMap<ShardKey, PrismaReplicaClient>;
   splitEnabled?: boolean;
   isPastRetention?: (id: string) => boolean;
 };
@@ -21,13 +28,17 @@ export type ResolveWaitpointReadThroughDefaults = {
   newClient: PrismaReplicaClient;
   legacyReplica: PrismaReplicaClient;
   newPrimary: PrismaReplicaClient;
+  shardReplicas: ReadonlyMap<ShardKey, PrismaReplicaClient>;
+  shardWriters: ReadonlyMap<ShardKey, PrismaReplicaClient>;
   splitEnabled: boolean;
 };
 
 const productionDefaults: ResolveWaitpointReadThroughDefaults = {
   newClient: defaultNewClient,
   legacyReplica: defaultLegacyReplica,
   newPrimary: defaultNewPrimary as unknown as PrismaReplicaClient,
+  shardReplicas: defaultShardReplicas,
+  shardWriters: defaultShardWriters as unknown as ReadonlyMap<ShardKey, PrismaReplicaClient>,
   splitEnabled: defaultSplitReadEnabled,
 };
 
@@ -43,30 +54,40 @@ export async function resolveWaitpointThroughReadThrough<T>(opts: {
   const splitEnabled = opts.deps?.splitEnabled ?? defaults.splitEnabled;
 
   const result = await readThroughRun({
-    runId: opts.waitpointId,
+    id: opts.waitpointId,
+    idKind: "waitpoint",
     environmentId: opts.environmentId,
     readNew: (client) => opts.read(client),
     readLegacy: (replica) => opts.read(replica),
     deps: {
       splitEnabled,
       newClient: opts.deps?.newClient ?? defaults.newClient,
       legacyReplica: opts.deps?.legacyReplica ?? defaults.legacyReplica,
+      shardReplicas: opts.deps?.shardReplicas ?? defaults.shardReplicas,
       isPastRetention: opts.deps?.isPastRetention,
     },
   });
 
-  if (result.source === "new" || result.source === "legacy-replica") {
+  if (result.found) {
     return result.value;
   }
   // past-retention is an intentional not-found: the token is gone.
-  if (result.source === "past-retention") {
+  if (result.reason === "past-retention") {
     return null;
   }
 
   // Read-your-writes fallback for a token completed immediately after mint, before it replicated:
-  // re-read from the run-ops PRIMARY only. We deliberately never read the control-plane/legacy
+  // re-read from the owning store's PRIMARY only. We deliberately never read the control-plane/legacy
   // primary here (that is the load the replica-only read-through exists to shed), so a legacy-resident
   // token that misses its replica stays a miss and the caller retries, rather than adding primary load.
+  const shardKey = resolveShard(opts.waitpointId);
+  if (shardKey !== "new" && shardKey !== "legacy") {
+    // A gen-2 token's primary is its OWN shard's writer. The gen-1 new writer is a different
+    // database, so reading it would miss and silently disable read-your-writes here.
+    const shardWriter = (opts.deps?.shardWriters ?? defaults.shardWriters).get(shardKey);
+    return shardWriter ? await opts.read(shardWriter) : null;
+  }
+
   const fromNewPrimary = await opts.read(opts.deps?.newPrimary ?? defaults.newPrimary);
   if (fromNewPrimary != null) {
     return fromNewPrimary;

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

@@ -35,7 +35,8 @@ export async function readRunForEvent<S extends Prisma.TaskRunSelect>(
   deps: EventReadDeps
 ): Promise<Prisma.TaskRunGetPayload<{ select: S }> | null> {
   const result = await readThroughRun<Prisma.TaskRunGetPayload<{ select: S }>>({
-    runId,
+    id: runId,
+    idKind: "run",
     environmentId,
     readNew: (client) => deps.store.findRun({ id: runId }, { select }, client),
     readLegacy: (replica) => deps.store.findRun({ id: runId }, { select }, replica),
@@ -47,7 +48,7 @@ export async function readRunForEvent<S extends Prisma.TaskRunSelect>(
     },
   });
 
-  return result.source === "not-found" || result.source === "past-retention" ? null : result.value;
+  return result.found ? result.value : null;
 }
 
 /**

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

@@ -13,6 +13,21 @@ vi.setConfig({ testTimeout: 60_000 });
 // 25-char cuid body → LEGACY residency. 26-char v1 body (version "1" at index 25) → NEW residency.
 const LEGACY_RUN_ID = "run_" + "a".repeat(25);
 const NEW_RUN_ID = "run_" + "b".repeat(24) + "01";
+// 26-char gen-2 body: shard char at index 24, version "2" at index 25.
+const SHARD_A_RUN_ID = "run_" + "c".repeat(24) + "a2";
+const SHARD_Z_RUN_ID = "run_" + "c".repeat(24) + "z2";
+const LEGACY_WAITPOINT_ID = "waitpoint_" + "d".repeat(25);
+
+function throwingClient(label: string) {
+  return vi.fn(async (): Promise<{ marker: number } | null> => {
+    throw new Error(`${label} must never be read`);
+  });
+}
+
+function collectingLogger() {
+  const errors: { message: string; meta?: unknown }[] = [];
+  return { errors, error: (message: string, meta?: unknown) => errors.push({ message, meta }) };
+}
 
 // Lightweight real read: a trivial `$queryRaw` that genuinely hits the given container.
 // `hit` controls whether the read "finds" the run, so we exercise routing without
@@ -28,14 +43,7 @@ async function realRead(
 // A presenter-shaped mapping: both "not-found" and "past-retention" collapse to the
 // same 404-ish surface, so an old run after termination yields the normal response.
 function toHttpish<T>(result: ReadThroughResult<T>): { status: number; value?: T } {
-  switch (result.source) {
-    case "new":
-    case "legacy-replica":
-      return { status: 200, value: result.value };
-    case "not-found":
-    case "past-retention":
-      return { status: 404 };
-  }
+  return result.found ? { status: 200, value: result.value } : { status: 404 };
 }
 
 describe("readThroughRun (legacy replica + new DB)", () => {
@@ -46,7 +54,8 @@ describe("readThroughRun (legacy replica + new DB)", () => {
       // read resolving through `legacyReplica` (prisma14) IS the structural guarantee
       // that the primary is never touched.
       const result = await readThroughRun({
-        runId: LEGACY_RUN_ID,
+        id: LEGACY_RUN_ID,
+        idKind: "run",
         environmentId: "env_1",
         readNew: (c) => realRead(c, false),
         readLegacy: (c) => realRead(c, true),
@@ -57,7 +66,7 @@ describe("readThroughRun (legacy replica + new DB)", () => {
         },
       });
 
-      expect(result.source).toBe("legacy-replica");
+      expect(result.found && result.source).toBe("legacy-replica");
       expect(toHttpish(result).status).toBe(200);
     }
   );
@@ -66,7 +75,8 @@ describe("readThroughRun (legacy replica + new DB)", () => {
     "post-termination past-retention returns the normal not-found surface",
     async ({ prisma14, prisma17 }) => {
       const pastRetentionResult = await readThroughRun({
-        runId: LEGACY_RUN_ID,
+        id: LEGACY_RUN_ID,
+        idKind: "run",
         environmentId: "env_1",
         readNew: (c) => realRead(c, false),
         readLegacy: (c) => realRead(c, false), // legacy gone / retention elapsed
@@ -78,11 +88,12 @@ describe("readThroughRun (legacy replica + new DB)", () => {
         },
       });
 
-      expect(pastRetentionResult.source).toBe("past-retention");
+      expect(pastRetentionResult.found === false && pastRetentionResult.reason).toBe("past-retention");
 
       // A run that is simply absent (not past retention) yields not-found.
       const notFoundResult = await readThroughRun({
-        runId: LEGACY_RUN_ID,
+        id: LEGACY_RUN_ID,
+        idKind: "run",
         environmentId: "env_1",
         readNew: (c) => realRead(c, false),
         readLegacy: (c) => realRead(c, false),
@@ -94,7 +105,7 @@ describe("readThroughRun (legacy replica + new DB)", () => {
         },
       });
 
-      expect(notFoundResult.source).toBe("not-found");
+      expect(notFoundResult.found === false && notFoundResult.reason).toBe("not-found");
       // Both collapse to the same 404-ish surface.
       expect(toHttpish(pastRetentionResult).status).toBe(toHttpish(notFoundResult).status);
       expect(toHttpish(pastRetentionResult).status).toBe(404);
@@ -110,7 +121,8 @@ describe("readThroughRun (legacy replica + new DB)", () => {
       const newRead = vi.fn((c: PrismaReplicaClient) => realRead(c, true));
 
       const result = await readThroughRun({
-        runId: LEGACY_RUN_ID,
+        id: LEGACY_RUN_ID,
+        idKind: "run",
         environmentId: "env_1",
         readNew: newRead,
         readLegacy: throwingLegacy,
@@ -121,7 +133,7 @@ describe("readThroughRun (legacy replica + new DB)", () => {
         },
       });
 
-      expect(result.source).toBe("new");
+      expect(result.found && result.source).toBe("new");
       expect(newRead).toHaveBeenCalledTimes(1);
       expect(throwingLegacy).not.toHaveBeenCalled();
     }
@@ -135,7 +147,152 @@ describe("readThroughRun (legacy replica + new DB)", () => {
       });
 
       const result = await readThroughRun({
-        runId: NEW_RUN_ID,
+        id: NEW_RUN_ID,
+        idKind: "run",
+        environmentId: "env_1",
+        readNew: (c) => realRead(c, true),
+        readLegacy: throwingLegacy,
+        deps: {
+          splitEnabled: true,
+          newClient: prisma17 as unknown as PrismaReplicaClient,
+          legacyReplica: prisma14 as unknown as PrismaReplicaClient,
+        },
+      });
+
+      expect(result.found && result.source).toBe("new");
+      expect(throwingLegacy).not.toHaveBeenCalled();
+    }
+  );
+
+  heteroPostgresTest(
+    "gen-2 id reads its OWN shard replica once and probes no other store",
+    async ({ prisma14, prisma17 }) => {
+      const throwingNew = throwingClient("the gen-1 new store");
+      const throwingLegacy = throwingClient("the legacy replica");
+      const shardRead = vi.fn((c: PrismaReplicaClient) => realRead(c, true));
+
+      const result = await readThroughRun({
+        id: SHARD_A_RUN_ID,
+        idKind: "run",
+        environmentId: "env_1",
+        // One closure serves both the gen-1 new store and a shard: a shard is the same
+        // dedicated schema. The throwing clients prove WHICH client it was handed.
+        readNew: (c) => shardRead(c),
+        readLegacy: throwingLegacy,
+        deps: {
+          splitEnabled: true,
+          newClient: throwingNew as unknown as PrismaReplicaClient,
+          legacyReplica: prisma14 as unknown as PrismaReplicaClient,
+          shardReplicas: new Map([["a", prisma17 as unknown as PrismaReplicaClient]]),
+        },
+      });
+
+      expect(result.found && result.source).toBe("shard:a");
+      expect(shardRead).toHaveBeenCalledTimes(1);
+      // Identity, not deep equality: a Prisma client is too large to deep-compare.
+      expect(shardRead.mock.calls[0][0]).toBe(prisma17);
+      expect(throwingLegacy).not.toHaveBeenCalled();
+    }
+  );
+
+  heteroPostgresTest(
+    "gen-2 id on an UNCONFIGURED shard key logs an error and returns not-found, never throws",
+    async ({ prisma14, prisma17 }) => {
+      const logger = collectingLogger();
+      const throwingLegacy = throwingClient("the legacy replica");
+      const newRead = vi.fn((c: PrismaReplicaClient) => realRead(c, true));
+
+      // Shard "z" is not configured. A 500 here would be inducible by any caller that
+      // guesses a shard char, so the layer must degrade rather than throw.
+      const result = await readThroughRun({
+        id: SHARD_Z_RUN_ID,
+        idKind: "run",
+        environmentId: "env_1",
+        readNew: newRead,
+        readLegacy: throwingLegacy,
+        deps: {
+          splitEnabled: true,
+          newClient: prisma17 as unknown as PrismaReplicaClient,
+          legacyReplica: prisma14 as unknown as PrismaReplicaClient,
+          shardReplicas: new Map([["a", prisma17 as unknown as PrismaReplicaClient]]),
+          logger,
+        },
+      });
+
+      expect(result.found).toBe(false);
+      expect(result.found === false && result.reason).toBe("not-found");
+      expect(logger.errors).toHaveLength(1);
+      expect(logger.errors[0].meta).toMatchObject({ shardKey: "z", configured: ["a"] });
+      // It must not silently fall back onto a gen-1 store.
+      expect(newRead).not.toHaveBeenCalled();
+      expect(throwingLegacy).not.toHaveBeenCalled();
+    }
+  );
+
+  heteroPostgresTest(
+    "gen-1 RUN id reads the legacy replica only and never probes the new store",
+    async ({ prisma14 }) => {
+      const throwingNew = throwingClient("the new store");
+      const legacyRead = vi.fn((c: PrismaReplicaClient) => realRead(c, true));
+
+      const result = await readThroughRun({
+        id: LEGACY_RUN_ID,
+        idKind: "run",
+        environmentId: "env_1",
+        readNew: throwingNew,
+        readLegacy: legacyRead,
+        deps: {
+          splitEnabled: true,
+          newClient: prisma14 as unknown as PrismaReplicaClient,
+          legacyReplica: prisma14 as unknown as PrismaReplicaClient,
+        },
+      });
+
+      expect(result.found && result.source).toBe("legacy-replica");
+      expect(throwingNew).not.toHaveBeenCalled();
+      expect(legacyRead).toHaveBeenCalledTimes(1);
+    }
+  );
+
+  heteroPostgresTest(
+    "cuid WAITPOINT id keeps the new-FIRST pair probe (frozen: cuid waitpoints co-locate on new)",
+    async ({ prisma14, prisma17 }) => {
+      const calls: string[] = [];
+      const newRead = vi.fn(async (c: PrismaReplicaClient) => {
+        calls.push("new");
+        return realRead(c, false);
+      });
+      const legacyRead = vi.fn(async (c: PrismaReplicaClient) => {
+        calls.push("legacy");
+        return realRead(c, true);
+      });
+
+      const result = await readThroughRun({
+        id: LEGACY_WAITPOINT_ID,
+        idKind: "waitpoint",
+        environmentId: "env_1",
+        readNew: newRead,
+        readLegacy: legacyRead,
+        deps: {
+          splitEnabled: true,
+          newClient: prisma17 as unknown as PrismaReplicaClient,
+          legacyReplica: prisma14 as unknown as PrismaReplicaClient,
+        },
+      });
+
+      expect(result.found && result.source).toBe("legacy-replica");
+      expect(calls).toEqual(["new", "legacy"]);
+    }
+  );
+
+  heteroPostgresTest(
+    "a cuid waitpoint found on the new store returns it without touching legacy",
+    async ({ prisma14, prisma17 }) => {
+      const throwingLegacy = throwingClient("the legacy replica");
+
+      const result = await readThroughRun({
+        id: LEGACY_WAITPOINT_ID,
+        idKind: "waitpoint",
         environmentId: "env_1",
         readNew: (c) => realRead(c, true),
         readLegacy: throwingLegacy,
@@ -146,7 +303,7 @@ describe("readThroughRun (legacy replica + new DB)", () => {
         },
       });
 
-      expect(result.source).toBe("new");
+      expect(result.found && result.source).toBe("new");
       expect(throwingLegacy).not.toHaveBeenCalled();
     }
   );