fix(run-store,run-engine): fix run-ops split hangs from wrong-store reads on the resume path (#4163)

## Summary

On the run-ops split, NEW-residency runs could hang. Time-based waits
(`wait.for`, `wait.until`, `delay`, waitpoint tokens),
`batchTriggerAndWait`, and attempt starts stalled and never resumed.
Each was a run-ops read or update that hit the wrong database: either
the owning store's read replica when it needed read-your-writes, or the
wrong store entirely because it routed by an id that does not encode
residency.

## Fixes

**Waitpoint resume (the main hang).** The managed resume path reads a
run's completed waitpoints by snapshot id
(`findSnapshotCompletedWaitpointIds`). Snapshot ids are cuids, which
always classify to the legacy store, so a NEW run's join rows (which
live on the new store) were never found. The resumed run saw zero
completed waitpoints and hung. It now fans out across both stores and
merges, like its sibling readers.

**Batch completion.** Batch item completion
(`updateManyBatchTaskRunItems`) routed by the item id, which is also a
cuid, so a NEW batch's items were updated on the wrong store, matched
zero rows, and the batch was treated as already complete (its parent's
`batchTriggerAndWait` then hung). It now routes by the batch id, which
does encode residency, matching the sibling `countBatchTaskRunItems`.

**Read-your-writes on the resume path.** The block-time
pending-waitpoint check (`countPendingWaitpoints`) and the attempt-start
lock check (`findRun` in `startRunAttempt`) both read the owning store's
replica with no read-your-writes guarantee, so a just-committed
waitpoint completion or dequeue lock could be missed under replica lag
and strand the run. Both now read the owning primary.

Each fix ships with a two-database store or engine test that reproduces
the hang and passes with the fix.

Author: d-cs

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

@@ -10,6 +10,7 @@ import {
 } from "@trigger.dev/core/v3";
 
 import { AttemptId, getMaxDuration, parseTraceparent } from "@trigger.dev/core/v3/isomorphic";
+import { $replica, runOpsNewReplica, runOpsSplitReadEnabled } from "~/db.server";
 import {
   extractIdempotencyKeyScope,
   getUserProvidedIdempotencyKey,
@@ -722,7 +723,11 @@ export class SpanPresenter extends BasePresenter {
           return { ...data, entity: null };
         }
 
-        const presenter = new WaitpointPresenter(undefined, undefined, {});
+        const presenter = new WaitpointPresenter(undefined, undefined, {
+          newClient: runOpsNewReplica,
+          legacyReplica: $replica,
+          splitEnabled: runOpsSplitReadEnabled,
+        });
         const waitpoint = await presenter.call({
           friendlyId: span.entity.id,
           environmentId,

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

@@ -45,12 +45,6 @@ export class WaitpointPresenter extends BasePresenter {
       completedAfter: true,
       completedAt: true,
       createdAt: true,
-      connectedRuns: {
-        select: {
-          friendlyId: true,
-        },
-        take: 5,
-      },
       tags: true,
       environmentId: true,
     } as const;
@@ -80,6 +74,66 @@ export class WaitpointPresenter extends BasePresenter {
     return result.source === "new" || result.source === "legacy-replica" ? result.value : null;
   }
 
+  // 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`.
+  async #connectedRunFriendlyIds(waitpointId: string): Promise<string[]> {
+    const replica = this._replica as unknown as PrismaReplicaClient;
+    const rawClients: PrismaReplicaClient[] =
+      this.readThroughDeps?.splitEnabled === true
+        ? [
+            (this.readThroughDeps.newClient as PrismaReplicaClient | undefined) ?? replica,
+            (this.readThroughDeps.legacyReplica as PrismaReplicaClient | undefined) ?? replica,
+          ]
+        : [replica];
+    const clients = [...new Set(rawClients)];
+
+    const friendlyIds = new Set<string>();
+    for (const client of clients) {
+      const runIds = await this.#connectedRunIdsOn(client, waitpointId);
+      if (runIds.length === 0) {
+        continue;
+      }
+      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) {
+        break;
+      }
+    }
+    return Array.from(friendlyIds).slice(0, 5);
+  }
+
+  // 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 = (
+      client as unknown as {
+        waitpointRunConnection?: {
+          findMany: (args: unknown) => Promise<{ taskRunId: string }[]>;
+        };
+      }
+    ).waitpointRunConnection;
+    if (joinDelegate && typeof joinDelegate.findMany === "function") {
+      const links = await joinDelegate.findMany({
+        where: { waitpointId },
+        select: { taskRunId: true },
+      });
+      return links.map((link) => link.taskRunId);
+    }
+    const rows = await client.$queryRaw<{ A: string }[]>`
+      SELECT "A" FROM "_WaitpointRunConnections" WHERE "B" = ${waitpointId}
+    `;
+    return rows.map((row) => row.A);
+  }
+
   public async call({
     friendlyId,
     environmentId,
@@ -119,7 +173,7 @@ export class WaitpointPresenter extends BasePresenter {
       }
     }
 
-    const connectedRunIds = waitpoint.connectedRuns.map((run) => run.friendlyId);
+    const connectedRunIds = await this.#connectedRunFriendlyIds(waitpoint.id);
     const connectedRuns: NextRunListItem[] = [];
 
     if (connectedRunIds.length > 0) {

apps/webapp/app/routes/api.v1.runs.$runParam.result.ts

@@ -2,6 +2,7 @@ import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { ApiRunResultPresenter } from "~/presenters/v3/ApiRunResultPresenter.server";
+import { $replica, runOpsNewReplica, runOpsSplitReadEnabled } from "~/db.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 
@@ -27,7 +28,11 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
   const { runParam } = parsed.data;
 
   try {
-    const presenter = new ApiRunResultPresenter();
+    const presenter = new ApiRunResultPresenter(undefined, undefined, {
+      newClient: runOpsNewReplica,
+      legacyReplica: $replica,
+      splitEnabled: runOpsSplitReadEnabled,
+    });
     const result = await presenter.call(runParam, authenticationResult.environment);
 
     if (!result) {

apps/webapp/test/waitpointPresenter.dedicatedConnectedRuns.readthrough.test.ts

@@ -0,0 +1,253 @@
+import { describe, expect, vi } from "vitest";
+
+// Uses the REAL dedicated run-ops client (RunOpsPrismaClient / SUBSET schema) as the new-DB handle,
+// whose `Waitpoint` model has NO `connectedRuns` relation — so a relation-select of it throws rather
+// than missing. The existing suite can't catch that: it wires a full-schema PG17 as the "new" client.
+// NextRunListPresenter is stubbed to echo its `runId` set back as `runs`, so `result.connectedRuns` is
+// exactly the friendlyIds the presenter gathered cross-DB (isolating the gather from the CH hydrate).
+const legacyReplicaHolder = vi.hoisted(() => ({ client: undefined as any }));
+const newClientHolder = vi.hoisted(() => ({ client: undefined as any }));
+
+vi.mock("~/db.server", async () => {
+  const { Prisma } = await import("@trigger.dev/database");
+  const lazyProxy = (holder: { client: any }, label: string) =>
+    new Proxy(
+      {},
+      {
+        get(_t, prop) {
+          if (!holder.client) {
+            throw new Error(`${label} not set for this test`);
+          }
+          return holder.client[prop];
+        },
+      }
+    );
+  const replicaProxy = lazyProxy(legacyReplicaHolder, "legacyReplicaHolder.client");
+  return {
+    prisma: replicaProxy,
+    $replica: replicaProxy,
+    runOpsNewPrisma: lazyProxy(newClientHolder, "newClientHolder.client"),
+    runOpsNewReplica: lazyProxy(newClientHolder, "newClientHolder.client"),
+    runOpsLegacyPrisma: replicaProxy,
+    runOpsLegacyReplica: replicaProxy,
+    sqlDatabaseSchema: Prisma.sql([`public`]),
+  };
+});
+
+vi.mock("~/services/clickhouse/clickhouseFactoryInstance.server", () => ({
+  clickhouseFactory: {
+    getClickhouseForOrganization: async () => ({}),
+  },
+}));
+
+// Echo the runId set back as runs so `result.connectedRuns` == the friendlyIds the presenter gathered.
+vi.mock("~/presenters/v3/NextRunListPresenter.server", () => ({
+  NextRunListPresenter: class {
+    // eslint-disable-next-line @typescript-eslint/no-unused-vars
+    constructor(..._args: unknown[]) {}
+    async call(_organizationId: string, _environmentId: string, opts: { runId?: string[] }) {
+      return {
+        runs: (opts.runId ?? []).map((friendlyId) => ({
+          friendlyId,
+          taskIdentifier: "echoed",
+        })),
+      };
+    }
+  },
+}));
+
+import { heteroRunOpsPostgresTest } from "@internal/testcontainers";
+import type { PrismaClient } from "@trigger.dev/database";
+import type { RunOpsPrismaClient } from "@internal/run-ops-database";
+import { WaitpointPresenter } from "~/presenters/v3/WaitpointPresenter.server";
+
+vi.setConfig({ testTimeout: 90_000 });
+
+type SeedContext = {
+  organizationId: string;
+  projectId: string;
+  environmentId: string;
+};
+
+// Parents (org/project/env) only exist on the full control-plane schema; the dedicated subset has no
+// such models, so we always seed them on the legacy (PG14) client and let the resolver read them there.
+async function seedParents(prisma: PrismaClient, slug: string): Promise<SeedContext> {
+  const organization = await prisma.organization.create({
+    data: { title: `org-${slug}`, slug: `org-${slug}` },
+  });
+  const project = await prisma.project.create({
+    data: {
+      name: `proj-${slug}`,
+      slug: `proj-${slug}`,
+      organizationId: organization.id,
+      externalRef: `proj-${slug}`,
+    },
+  });
+  const runtimeEnvironment = await prisma.runtimeEnvironment.create({
+    data: {
+      slug: `env-${slug}`,
+      type: "DEVELOPMENT",
+      projectId: project.id,
+      organizationId: organization.id,
+      apiKey: `tr_dev_${slug}`,
+      pkApiKey: `pk_dev_${slug}`,
+      shortcode: `sc-${slug}`,
+    },
+  });
+  return {
+    organizationId: organization.id,
+    projectId: project.id,
+    environmentId: runtimeEnvironment.id,
+  };
+}
+
+async function seedWaitpoint(
+  prisma: PrismaClient | RunOpsPrismaClient,
+  ctx: SeedContext,
+  friendlyId: string
+) {
+  return (prisma as PrismaClient).waitpoint.create({
+    data: {
+      friendlyId,
+      type: "MANUAL",
+      status: "COMPLETED",
+      idempotencyKey: `idem-${friendlyId}`,
+      userProvidedIdempotencyKey: false,
+      output: JSON.stringify({ hello: "world" }),
+      outputType: "application/json",
+      outputIsError: false,
+      completedAt: new Date(),
+      tags: ["a", "b"],
+      projectId: ctx.projectId,
+      environmentId: ctx.environmentId,
+    },
+  });
+}
+
+async function seedRun(
+  prisma: PrismaClient | RunOpsPrismaClient,
+  ctx: SeedContext,
+  friendlyId: string
+) {
+  return (prisma as PrismaClient).taskRun.create({
+    data: {
+      friendlyId,
+      taskIdentifier: "my-task",
+      status: "PENDING",
+      payload: JSON.stringify({ foo: friendlyId }),
+      payloadType: "application/json",
+      traceId: friendlyId,
+      spanId: friendlyId,
+      queue: "test",
+      runtimeEnvironmentId: ctx.environmentId,
+      projectId: ctx.projectId,
+      organizationId: ctx.organizationId,
+      environmentType: "DEVELOPMENT",
+      engine: "V2",
+    },
+  });
+}
+
+const callArgs = (ctx: SeedContext, friendlyId: string) => ({
+  friendlyId,
+  environmentId: ctx.environmentId,
+  projectId: ctx.projectId,
+});
+
+describe("WaitpointPresenter against the REAL dedicated run-ops client", () => {
+  // A NEW-resident waitpoint (on the dedicated subset schema) with no connected runs. The current
+  // relation-select of `connectedRuns` is invalid on the dedicated Waitpoint model, so the read
+  // throws PrismaClientValidationError. Desired: resolves the waitpoint, connectedRuns empty.
+  heteroRunOpsPostgresTest(
+    "resolves a new-resident waitpoint without a connectedRuns relation-select (no throw)",
+    async ({ prisma14, prisma17 }) => {
+      const ctx = await seedParents(prisma14, "dedself");
+      const seeded = await seedWaitpoint(prisma17, ctx, "waitpoint_dedself");
+
+      legacyReplicaHolder.client = prisma14;
+      newClientHolder.client = prisma17;
+
+      const presenter = new WaitpointPresenter(undefined, undefined, {
+        splitEnabled: true,
+        newClient: prisma17 as unknown as PrismaClient,
+        legacyReplica: prisma14,
+      });
+
+      const result = await presenter.call(callArgs(ctx, seeded.friendlyId));
+
+      expect(result?.id).toBe(seeded.friendlyId);
+      expect(result?.connectedRuns).toEqual([]);
+    }
+  );
+
+  // Cross-DB connection: waitpoint on LEGACY (PG14), the connected run + its WaitpointRunConnection
+  // join on the NEW dedicated DB (PG17). A single-DB gather off the waitpoint's own store misses the
+  // run entirely; the fix reads the join from BOTH stores (dedicated `waitpointRunConnection`
+  // delegate + legacy raw `_WaitpointRunConnections`) and unions the friendlyIds.
+  heteroRunOpsPostgresTest(
+    "gathers a cross-DB connected run whose join lives on the other database",
+    async ({ prisma14, prisma17 }) => {
+      const ctx = await seedParents(prisma14, "crossdb");
+      const waitpoint = await seedWaitpoint(prisma14, ctx, "waitpoint_crossdb");
+
+      // The connected run + join live only on the NEW dedicated DB (co-resident with the run).
+      const run = await seedRun(prisma17, ctx, "run_crossnew");
+      await prisma17.waitpointRunConnection.create({
+        data: { taskRunId: run.id, waitpointId: waitpoint.id },
+      });
+
+      legacyReplicaHolder.client = prisma14;
+      newClientHolder.client = prisma17;
+
+      const presenter = new WaitpointPresenter(undefined, undefined, {
+        splitEnabled: true,
+        newClient: prisma17 as unknown as PrismaClient,
+        legacyReplica: prisma14,
+      });
+
+      const result = await presenter.call(callArgs(ctx, waitpoint.friendlyId));
+
+      expect(result?.id).toBe(waitpoint.friendlyId);
+      expect(result?.connectedRuns.map((r) => r.friendlyId)).toEqual(["run_crossnew"]);
+    }
+  );
+
+  // Same-DB legacy connection (no regression): waitpoint + connected run both on LEGACY, joined via
+  // the implicit `_WaitpointRunConnections` M2M. The gather must read the legacy raw join path.
+  heteroRunOpsPostgresTest(
+    "still gathers a same-DB legacy connected run via the implicit M2M",
+    async ({ prisma14, prisma17 }) => {
+      const ctx = await seedParents(prisma14, "legsame");
+      const run = await seedRun(prisma14, ctx, "run_legsame");
+      const waitpoint = await prisma14.waitpoint.create({
+        data: {
+          friendlyId: "waitpoint_legsame",
+          type: "MANUAL",
+          status: "COMPLETED",
+          idempotencyKey: "idem-waitpoint_legsame",
+          userProvidedIdempotencyKey: false,
+          outputType: "application/json",
+          outputIsError: false,
+          completedAt: new Date(),
+          tags: [],
+          projectId: ctx.projectId,
+          environmentId: ctx.environmentId,
+          connectedRuns: { connect: [{ id: run.id }] },
+        },
+      });
+
+      legacyReplicaHolder.client = prisma14;
+      newClientHolder.client = prisma17;
+
+      const presenter = new WaitpointPresenter(undefined, undefined, {
+        splitEnabled: true,
+        newClient: prisma17 as unknown as PrismaClient,
+        legacyReplica: prisma14,
+      });
+
+      const result = await presenter.call(callArgs(ctx, waitpoint.friendlyId));
+
+      expect(result?.connectedRuns.map((r) => r.friendlyId)).toEqual(["run_legsame"]);
+    }
+  );
+});

internal-packages/database/prisma/migrations/20260705210000_drop_waitpoint_run_connections_waitpoint_fk/migration.sql

@@ -0,0 +1,7 @@
+-- Run-ops split: allow a cross-DB token connection (a LEGACY run blocking on a NEW-resident token,
+-- whose Waitpoint row lives on the other database). Matches the split's control-plane FK-removal pattern.
+
+-- Fail fast instead of queueing behind a long txn/VACUUM for the ACCESS EXCLUSIVE lock.
+SET lock_timeout = '5s';
+
+ALTER TABLE "_WaitpointRunConnections" DROP CONSTRAINT IF EXISTS "_WaitpointRunConnections_B_fkey";