perf(webapp,run-store): grouped run-ops reads + mint-kind flip grace (#4227)

## Summary

Two threads on the run-ops split path.

Read path: per-item run reads are batched into grouped queries, a
waitpoint's connected-run reads are bounded, and the dedicated-schema
relation hydrators fetch only the requested columns instead of whole
rows. Retrieve also falls back to the other database when a routed read
misses, so a run whose physical residency diverges from its id shape is
still found rather than returning a spurious not-found. Fewer and
lighter queries on the run read path, with no change to results.

Mint-kind flip safety: flipping which database new runs mint to is now a
deterministic wall-clock cutover, for both per-org and global flips. For
a grace window every process resolves the same database, so a flip
cannot route two concurrent triggers that share an idempotency key to
different databases (which would bypass the per-database unique
constraint and create a duplicate run).

Supersedes the earlier #4205 and #4208.

Draft: validation in progress.

Author: d-cs

apps/webapp/app/env.server.ts

@@ -1755,6 +1755,10 @@ const EnvironmentSchema = z
     RUN_OPS_MINT_ENABLED: BoolEnv.default(false),
     RUN_OPS_MINT_FLAG_CACHE_TTL_MS: z.coerce.number().int().default(30_000),
     RUN_OPS_MINT_FLAG_CACHE_MAX_ENTRIES: z.coerce.number().int().default(10_000),
+    // Deterministic wall-clock cutover after a runOpsMintKind flip. Must exceed the sum
+    // of RUN_OPS_MINT_FLAG_CACHE_TTL_MS and the control-plane cache TTL so every process
+    // (stale or fresh) resolves to the same kind for the whole window. See mintFlipGrace.ts.
+    RUN_OPS_MINT_FLIP_GRACE_MS: z.coerce.number().int().default(90_000),
 
     // Session replication (Postgres → ClickHouse sessions_v1). Shares Redis
     // with the runs replicator for leader locking but has its own slot and

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

@@ -1,4 +1,5 @@
 import type { BatchTaskRunExecutionResult } from "@trigger.dev/core/v3";
+import { ownerEngine } from "@trigger.dev/core/v3/isomorphic";
 import {
   $replica,
   type PrismaClientOrTransaction,
@@ -8,7 +9,6 @@ import {
 import type { TaskRunWithAttempts } from "~/models/taskRun.server";
 import { executionResultForTaskRun } from "~/models/taskRun.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import { readThroughRun } from "~/v3/runOpsMigration/readThrough.server";
 import { runStore as defaultRunStore } from "~/v3/runStore.server";
 import { BasePresenter } from "./basePresenter.server";
 
@@ -43,11 +43,14 @@ const memberRunSelect = {
 } as const;
 
 /**
- * Split on: the batch row + its item rows resolve new-run-ops first, then the LEGACY RUN-OPS
- * READ REPLICA ONLY (never the legacy primary — there is no such handle); each member run is
- * hydrated independently via readThroughRun keyed on the member runId, so a batch whose members
- * span migrated + abandoned runs returns the complete reachable set (the batch-spanning-the-line
- * read; the dangling-reference termination gate is a separate, adjacent unit).
+ * Split on: the batch row + its members resolve new-run-ops first, then the LEGACY RUN-OPS READ
+ * REPLICA ONLY (never the legacy primary — there is no such handle). Members hydrate via ONE
+ * grouped read against `newClient` for the whole id set, then ONE grouped read against
+ * `legacyReplica` for just the misses that could still be legacy-resident — the same
+ * residency-partitioned shape as the old per-member read-through, but batched instead of fanned
+ * out one query per member. A batch whose members span migrated + abandoned runs returns the
+ * complete reachable set (the batch-spanning-the-line read; the dangling-reference termination
+ * gate is a separate, adjacent unit).
  *
  * Split off (single-DB / self-host): one passthrough read for the batch row + a single store
  * id-set hydrate for the members — no legacy read, no known-migrated probe, no second connection.
@@ -133,7 +136,7 @@ export class ApiBatchResultsPresenter extends BasePresenter {
   // Split: resolve the batch row new-first then off the legacy READ REPLICA only (a batch id may
   // be cuid or run-ops id, and a cuid-shaped id can still have been backfilled onto NEW, so id-shape
   // residency is not authoritative for the row — the new-first-then-legacy probe is), then
-  // hydrate every member run independently via the per-run read-through primitive.
+  // hydrate every member run in ONE grouped new-then-legacy read.
   async #callSplit(
     friendlyId: string,
     env: AuthenticatedEnvironment
@@ -175,41 +178,35 @@ export class ApiBatchResultsPresenter extends BasePresenter {
       };
     }
 
-    const readMemberRun = (client: PrismaClientOrTransaction, taskRunId: string) =>
-      client.taskRun.findFirst({
-        where: { id: taskRunId },
-        select: memberRunSelect,
-      }) as Promise<TaskRunWithAttempts | null>;
-
-    // Per-member fan-out: each member may live on a different DB, so a single nested include cannot
-    // cross the seam. Promise.all preserves batchRun.items order, unchanged from today.
-    const memberResults = await Promise.all(
-      batchRun.items.map(async (item) => {
-        const result = await readThroughRun<TaskRunWithAttempts>({
-          runId: item.taskRunId,
-          environmentId: env.id,
-          readNew: (client) => readMemberRun(client, item.taskRunId),
-          readLegacy: (replica) => readMemberRun(replica, item.taskRunId),
-          deps: {
-            splitEnabled: true,
-            // Pass the SAME resolved handles the batch row used, so the batch row and its members
-            // never resolve against different DBs. (Letting these fall through to readThroughRun's
-            // own module-level defaults would diverge from the batch read's `?? this._replica`.)
-            newClient,
-            legacyReplica,
-            isPastRetention: this.readThrough?.isPastRetention,
-          },
-        });
+    const taskRunIds = batchRun.items.map((item) => item.taskRunId);
 
-        // not-found / past-retention members are omitted (matches today's drop-undefined behavior);
-        // the dangling-reference termination gate (separate unit) governs whether that's permitted.
-        if (result.source === "not-found" || result.source === "past-retention") {
-          return undefined;
-        }
+    const newRows = (await newClient.taskRun.findMany({
+      where: { id: { in: taskRunIds } },
+      select: memberRunSelect,
+    })) as TaskRunWithAttempts[];
+    const runsById = new Map(newRows.map((run) => [run.id, run]));
 
-        return executionResultForTaskRun(result.value);
-      })
+    // A run-ops id can only live on NEW, so only misses that AREN'T run-ops-shaped are candidates
+    // for the legacy probe — mirrors readThroughRun's per-id "NEW residency skips legacy" rule.
+    const legacyCandidateIds = taskRunIds.filter(
+      (id) => !runsById.has(id) && ownerEngine(id) !== "NEW"
     );
+    if (legacyCandidateIds.length > 0) {
+      const legacyRows = (await legacyReplica.taskRun.findMany({
+        where: { id: { in: legacyCandidateIds } },
+        select: memberRunSelect,
+      })) as TaskRunWithAttempts[];
+      for (const run of legacyRows) {
+        runsById.set(run.id, run);
+      }
+    }
+
+    // not-found members are omitted (matches today's drop-undefined behavior); the
+    // dangling-reference termination gate (separate unit) governs whether that's permitted.
+    const memberResults = batchRun.items.map((item) => {
+      const run = runsById.get(item.taskRunId);
+      return run ? executionResultForTaskRun(run) : undefined;
+    });
 
     return {
       id: batchRun.friendlyId,

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

@@ -164,17 +164,13 @@ export class ApiRetrieveRunPresenter {
         collect(child.lockedToVersionId);
       }
 
+      const lockedWorkersByVersionId =
+        await controlPlaneResolver.resolveRunLockedWorkersByVersionIds([...distinctVersionIds]);
       const versionById = new Map<string, string | null>(
-        await Promise.all(
-          [...distinctVersionIds].map(
-            async (id) =>
-              [
-                id,
-                (await controlPlaneResolver.resolveRunLockedWorker({ lockedToVersionId: id }))
-                  ?.lockedToVersion?.version ?? null,
-              ] as const
-          )
-        )
+        [...distinctVersionIds].map((id) => [
+          id,
+          lockedWorkersByVersionId.get(id)?.lockedToVersion?.version ?? null,
+        ])
       );
 
       const resolveVersion = (id: string | null): { version: string } | null => {

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

@@ -11,6 +11,12 @@ import { waitpointStatusToApiStatus } from "./WaitpointListPresenter.server";
 
 export type WaitpointDetail = NonNullable<Awaited<ReturnType<WaitpointPresenter["call"]>>>;
 
+// Single-sourced display bound for a waitpoint's connected run friendlyIds.
+export const CONNECTED_RUNS_DISPLAY_LIMIT = 5;
+
+// Over-read connection rows on the FK-free dedicated join so danglers don't cost a display slot.
+export const CONNECTED_RUNS_CONNECTION_SCAN_LIMIT = CONNECTED_RUNS_DISPLAY_LIMIT * 5;
+
 export class WaitpointPresenter extends BasePresenter {
   constructor(
     prisma?: PrismaClientOrTransaction,
@@ -76,8 +82,7 @@ export class WaitpointPresenter extends BasePresenter {
 
   // Connected-run friendlyIds gathered across BOTH stores. The run<->waitpoint join co-locates with
   // the RUN (written on the run's DB), so the waitpoint's own store misses a cross-DB connection; we
-  // read the join on each client and resolve the run's friendlyId on that same client, then union.
-  // We never relation-select `connectedRuns`: it is not a field on the dedicated subset `Waitpoint`.
+  // read the join on each client, resolve the run's friendlyId on that same client, and union.
   async #connectedRunFriendlyIds(waitpointId: string): Promise<string[]> {
     const replica = this._replica as unknown as PrismaReplicaClient;
     const rawClients: PrismaReplicaClient[] =
@@ -91,47 +96,66 @@ export class WaitpointPresenter extends BasePresenter {
 
     const friendlyIds = new Set<string>();
     for (const client of clients) {
-      const runIds = await this.#connectedRunIdsOn(client, waitpointId);
-      if (runIds.length === 0) {
-        continue;
+      for (const friendlyId of await this.#connectedRunFriendlyIdsOn(client, waitpointId)) {
+        friendlyIds.add(friendlyId);
       }
-      const runs = await client.taskRun.findMany({
-        where: { id: { in: runIds } },
-        select: { friendlyId: true },
-        take: 5,
-      });
-      for (const run of runs) {
-        friendlyIds.add(run.friendlyId);
-      }
-      if (friendlyIds.size >= 5) {
+      if (friendlyIds.size >= CONNECTED_RUNS_DISPLAY_LIMIT) {
         break;
       }
     }
-    return Array.from(friendlyIds).slice(0, 5);
+    return Array.from(friendlyIds).slice(0, CONNECTED_RUNS_DISPLAY_LIMIT);
   }
 
-  // Schema-aware read of the run ids linked to a waitpoint: the dedicated subset uses the explicit
-  // `WaitpointRunConnection` model, the control-plane full schema the implicit `_WaitpointRunConnections`
-  // M2M (A = TaskRun.id, B = Waitpoint.id). The dedicated join delegate is absent on the full client.
-  async #connectedRunIdsOn(client: PrismaReplicaClient, waitpointId: string): Promise<string[]> {
-    const joinDelegate = (
+  // Connected-run friendlyIds for one store, via the ORM. Two indexed reads joined in memory instead
+  // of a SQL JOIN onto the (very large) TaskRun table: `id IN (...)` can only plan as a PK lookup, so
+  // the planner can never scan TaskRun.
+  //
+  // Dedicated subset: the explicit `WaitpointRunConnection` is scalar (`taskRunId`, no FK), so a
+  // connection can outlive a deleted run. We over-read connection ids, resolve runs by id (a missing
+  // id just drops out -- danglers cost no display slot), and cap at the display limit.
+  //
+  // Control-plane full schema: no queryable join delegate (implicit M2M), so we traverse the
+  // `connectedRuns` relation; it cascade-deletes with the run, so no dangler can exist.
+  async #connectedRunFriendlyIdsOn(
+    client: PrismaReplicaClient,
+    waitpointId: string
+  ): Promise<string[]> {
+    const dedicated = (
       client as unknown as {
         waitpointRunConnection?: {
           findMany: (args: unknown) => Promise<{ taskRunId: string }[]>;
         };
       }
     ).waitpointRunConnection;
-    if (joinDelegate && typeof joinDelegate.findMany === "function") {
-      const links = await joinDelegate.findMany({
+
+    if (dedicated) {
+      const connections = await dedicated.findMany({
         where: { waitpointId },
         select: { taskRunId: true },
+        take: CONNECTED_RUNS_CONNECTION_SCAN_LIMIT,
       });
-      return links.map((link) => link.taskRunId);
+      if (connections.length === 0) {
+        return [];
+      }
+      const runs = await client.taskRun.findMany({
+        where: { id: { in: connections.map((connection) => connection.taskRunId) } },
+        select: { friendlyId: true },
+        take: CONNECTED_RUNS_DISPLAY_LIMIT,
+      });
+      return runs.map((run) => run.friendlyId);
     }
-    const rows = await client.$queryRaw<{ A: string }[]>`
-      SELECT "A" FROM "_WaitpointRunConnections" WHERE "B" = ${waitpointId}
-    `;
-    return rows.map((row) => row.A);
+
+    const waitpoint = (await (
+      client.waitpoint.findFirst as (
+        args: unknown
+      ) => Promise<{ connectedRuns: { friendlyId: string }[] } | null>
+    )({
+      where: { id: waitpointId },
+      select: {
+        connectedRuns: { select: { friendlyId: true }, take: CONNECTED_RUNS_DISPLAY_LIMIT },
+      },
+    })) as { connectedRuns: { friendlyId: string }[] } | null;
+    return (waitpoint?.connectedRuns ?? []).map((run) => run.friendlyId);
   }
 
   public async call({

apps/webapp/app/routes/admin.api.v1.feature-flags.ts

@@ -1,8 +1,9 @@
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { prisma } from "~/db.server";
+import { env } from "~/env.server";
 import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
-import { makeSetMultipleFlags } from "~/v3/featureFlags.server";
+import { applyGlobalMintKindFlip, makeSetMultipleFlags } from "~/v3/featureFlags.server";
 import { validatePartialFeatureFlags } from "~/v3/featureFlags";
 
 export async function action({ request }: ActionFunctionArgs) {
@@ -24,9 +25,19 @@ export async function action({ request }: ActionFunctionArgs) {
       );
     }
 
-    const featureFlags = validationResult.data;
-    const setMultipleFlags = makeSetMultipleFlags(prisma);
-    const updatedFlags = await setMultipleFlags(featureFlags);
+    // Derived grace-stamp fields are computed server-side; never trust them from the body.
+    const {
+      runOpsMintKindPrev: _ignoredPrev,
+      runOpsMintKindFlippedAt: _ignoredFlippedAt,
+      ...requestedFlags
+    } = validationResult.data;
+
+    // A global mint-kind flip stamps its grace window under a lock (applyGlobalMintKindFlip);
+    // any other flag save writes directly.
+    const updatedFlags =
+      requestedFlags.runOpsMintKind !== undefined
+        ? await applyGlobalMintKindFlip(prisma, requestedFlags, env.RUN_OPS_MINT_FLIP_GRACE_MS)
+        : await makeSetMultipleFlags(prisma)(requestedFlags);
 
     return json({
       success: true,