fix(run-ops split): resolve public wait tokens across the split boundary

d-cs · claude · d-cs · commit 3d6d068e98cb · 2026-07-02T18:54:58.000+01:00
The public wait-token routes (complete, HTTP callback, retrieve) resolved the
waitpoint with a bare read-through that defaulted its new-side client to the
dedicated run-ops replica and gated on the async mint flag. A standalone wait
token has a cuid id and, having no owning run, is written to the control-plane
store, so under the split topology the run-ops replica does not hold it and the
routes returned 404 "Waitpoint not found".

Resolve these routes the same way the working waiter route does: fan out through
the run-ops replica first, then the control-plane replica, so both a co-located
(run-owned) waitpoint and a standalone control-plane token are found. Gate the
fan-out on the URL-presence read gate rather than the mint flag, so read
visibility spans both DBs whenever both are configured — including the window
where both database URLs are set but the mint flag is off. The retrieve route
hands the same fan-out clients and gate to ApiWaitpointPresenter.

Adds a two-database testcontainer regression proving a control-plane-resident
standalone token resolves via the legacy fallback under the read gate, and that
the passthrough branch (gate off) misses it.

Co-Authored-By: Claude Opus 4.8 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts b/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts
@@ -2,13 +2,18 @@ import { type ActionFunctionArgs, json } from "@remix-run/server-runtime";
 import { type CompleteWaitpointTokenResponseBody, stringifyIO } from "@trigger.dev/core/v3";
 import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import { z } from "zod";
-import type { PrismaReplicaClient } from "~/db.server";
+import {
+  $replica,
+  type PrismaReplicaClient,
+  runOpsNewReplica,
+  runOpsSplitReadEnabled,
+} from "~/db.server";
 import { env } from "~/env.server";
 import { processWaitpointCompletionPacket } from "~/runEngine/concerns/waitpointCompletionPacket.server";
+import { resolveWaitpointThroughReadThrough } from "~/runEngine/concerns/resolveWaitpointThroughReadThrough.server";
 import { verifyHttpCallbackHash } from "~/services/httpCallback.server";
 import { logger } from "~/services/logger.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
-import { readThroughRun } from "~/v3/runOpsMigration/readThrough.server";
 import { engine } from "~/v3/runEngine.server";
 
 const paramsSchema = z.object({
@@ -34,28 +39,27 @@ export async function action({ request, params }: ActionFunctionArgs) {
   const waitpointId = WaitpointId.toId(waitpointFriendlyId);
 
   try {
-    // Read through the split-aware run-ops read-through (passthrough in single-DB). The env is
-    // resolved below from the row; residency is classified off the waitpoint id, so env "" is fine.
-    const findWaitpoint = (client: PrismaReplicaClient) =>
-      client.waitpoint.findFirst({
-        where: {
-          id: waitpointId,
-        },
-        select: { id: true, status: true, environmentId: true },
-      });
-
-    const waitpointResult = await readThroughRun({
-      runId: waitpointId,
+    // Resolve wherever the waitpoint resides. The env is resolved below from the row; residency
+    // is classified off the waitpoint id, so env "" is fine. Fan-out reads the run-ops replica
+    // first, then the control-plane replica so both a co-located and a standalone token resolve,
+    // gated on the URL-presence read gate so the fan-out spans both DBs independent of the mint flag.
+    const waitpoint = await resolveWaitpointThroughReadThrough({
+      waitpointId,
       environmentId: "",
-      readNew: (client) => findWaitpoint(client),
-      readLegacy: (replica) => findWaitpoint(replica),
+      read: (client: PrismaReplicaClient) =>
+        client.waitpoint.findFirst({
+          where: {
+            id: waitpointId,
+          },
+          select: { id: true, status: true, environmentId: true },
+        }),
+      deps: {
+        newClient: runOpsNewReplica,
+        legacyReplica: $replica,
+        splitEnabled: runOpsSplitReadEnabled,
+      },
     });
 
-    const waitpoint =
-      waitpointResult.source === "new" || waitpointResult.source === "legacy-replica"
-        ? waitpointResult.value
-        : null;
-
     if (!waitpoint) {
       return json({ error: "Waitpoint not found" }, { status: 404 });
     }
diff --git a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts b/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts
@@ -6,12 +6,17 @@ import {
 } from "@trigger.dev/core/v3";
 import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import { z } from "zod";
-import type { PrismaReplicaClient } from "~/db.server";
+import {
+  $replica,
+  type PrismaReplicaClient,
+  runOpsNewReplica,
+  runOpsSplitReadEnabled,
+} from "~/db.server";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { processWaitpointCompletionPacket } from "~/runEngine/concerns/waitpointCompletionPacket.server";
+import { resolveWaitpointThroughReadThrough } from "~/runEngine/concerns/resolveWaitpointThroughReadThrough.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
-import { readThroughRun } from "~/v3/runOpsMigration/readThrough.server";
 import { engine } from "~/v3/runEngine.server";
 
 const { action, loader } = createActionApiRoute(
@@ -34,27 +39,27 @@ const { action, loader } = createActionApiRoute(
 
     try {
       //check permissions
-      // Read through the split-aware run-ops read-through (passthrough in single-DB).
-      const findWaitpoint = (client: PrismaReplicaClient) =>
-        client.waitpoint.findFirst({
-          where: {
-            id: waitpointId,
-            environmentId: authentication.environment.id,
-          },
-        });
-
-      const waitpointResult = await readThroughRun({
-        runId: waitpointId,
+      // Resolve wherever the waitpoint resides: a standalone token lives on the control-plane
+      // store, while a run-owned waitpoint co-locates with its run. Fan-out reads the run-ops
+      // replica first, then the control-plane replica so both residencies resolve, gated on the
+      // URL-presence read gate so the fan-out spans both DBs independent of the mint flag.
+      const waitpoint = await resolveWaitpointThroughReadThrough({
+        waitpointId,
         environmentId: authentication.environment.id,
-        readNew: (client) => findWaitpoint(client),
-        readLegacy: (replica) => findWaitpoint(replica),
+        read: (client: PrismaReplicaClient) =>
+          client.waitpoint.findFirst({
+            where: {
+              id: waitpointId,
+              environmentId: authentication.environment.id,
+            },
+          }),
+        deps: {
+          newClient: runOpsNewReplica,
+          legacyReplica: $replica,
+          splitEnabled: runOpsSplitReadEnabled,
+        },
       });
 
-      const waitpoint =
-        waitpointResult.source === "new" || waitpointResult.source === "legacy-replica"
-          ? waitpointResult.value
-          : null;
-
       if (!waitpoint) {
         throw json({ error: "Waitpoint not found" }, { status: 404 });
       }
diff --git a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.ts b/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.ts
@@ -2,6 +2,7 @@ import { json } from "@remix-run/server-runtime";
 import { type WaitpointRetrieveTokenResponse } from "@trigger.dev/core/v3";
 import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import { z } from "zod";
+import { $replica, runOpsNewReplica, runOpsSplitReadEnabled } from "~/db.server";
 import { ApiWaitpointPresenter } from "~/presenters/v3/ApiWaitpointPresenter.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 
@@ -13,7 +14,11 @@ export const loader = createLoaderApiRoute(
     findResource: async () => 1, // This is a dummy function, we don't need to find a resource
   },
   async ({ params, authentication }) => {
-    const presenter = new ApiWaitpointPresenter();
+    const presenter = new ApiWaitpointPresenter(undefined, undefined, {
+      newClient: runOpsNewReplica,
+      legacyReplica: $replica,
+      splitEnabled: runOpsSplitReadEnabled,
+    });
     const result: WaitpointRetrieveTokenResponse = await presenter.call(
       authentication.environment,
       WaitpointId.toId(params.waitpointFriendlyId)
diff --git a/apps/webapp/app/v3/runOpsMigration/waitpointTokenResolve.server.test.ts b/apps/webapp/app/v3/runOpsMigration/waitpointTokenResolve.server.test.ts
@@ -0,0 +1,148 @@
+// A standalone wait token is minted with a cuid id and, having no owning run, is written to the
+// control-plane store. Under the split topology the run-ops read replica is a distinct database
+// that does not hold it, so the public token routes must fan out to the control-plane replica or
+// the token is reported missing. These reads run as real queries against two containers.
+import { heteroPostgresTest } from "@internal/testcontainers";
+import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
+import { describe, expect, vi } from "vitest";
+import type { PrismaClient } from "@trigger.dev/database";
+import type { PrismaReplicaClient } from "~/db.server";
+import { resolveWaitpointThroughReadThrough } from "~/runEngine/concerns/resolveWaitpointThroughReadThrough.server";
+import { readThroughRun } from "./readThrough.server";
+
+vi.setConfig({ testTimeout: 60_000 });
+
+async function seedControlPlaneEnv(prisma: PrismaClient, suffix: string) {
+  const organization = await prisma.organization.create({
+    data: { title: `Org ${suffix}`, slug: `org-${suffix}` },
+  });
+  const project = await prisma.project.create({
+    data: {
+      name: `Project ${suffix}`,
+      slug: `project-${suffix}`,
+      externalRef: `proj_${suffix}`,
+      organizationId: organization.id,
+    },
+  });
+  const environment = await prisma.runtimeEnvironment.create({
+    data: {
+      type: "PRODUCTION",
+      slug: `prod-${suffix}`,
+      projectId: project.id,
+      organizationId: organization.id,
+      apiKey: `tr_prod_${suffix}`,
+      pkApiKey: `pk_prod_${suffix}`,
+      shortcode: `short_${suffix}`,
+      maximumConcurrencyLimit: 10,
+    },
+  });
+  return { organization, project, environment };
+}
+
+async function seedStandaloneToken(prisma: PrismaClient, environmentId: string, projectId: string) {
+  const { id, friendlyId } = WaitpointId.generate();
+  await prisma.waitpoint.create({
+    data: {
+      id,
+      friendlyId,
+      type: "MANUAL",
+      status: "PENDING",
+      idempotencyKey: id,
+      userProvidedIdempotencyKey: false,
+      environmentId,
+      projectId,
+    },
+  });
+  return { id, friendlyId };
+}
+
+describe("public wait-token resolution across the split boundary", () => {
+  heteroPostgresTest(
+    "a control-plane-resident standalone token is found when the run-ops replica does not hold it",
+    async ({ prisma14, prisma17 }) => {
+      const { project, environment } = await seedControlPlaneEnv(prisma14, "token_cp");
+      const { id: waitpointId } = await seedStandaloneToken(prisma14, environment.id, project.id);
+
+      const waitpoint = await resolveWaitpointThroughReadThrough({
+        waitpointId,
+        environmentId: environment.id,
+        read: (client: PrismaReplicaClient) =>
+          client.waitpoint.findFirst({
+            where: { id: waitpointId, environmentId: environment.id },
+          }),
+        deps: {
+          newClient: prisma17 as unknown as PrismaReplicaClient,
+          legacyReplica: prisma14 as unknown as PrismaReplicaClient,
+        },
+      });
+
+      expect(waitpoint).not.toBeNull();
+      expect(waitpoint?.id).toBe(waitpointId);
+    }
+  );
+
+  heteroPostgresTest(
+    "pinning both reads at the run-ops replica misses the control-plane token",
+    async ({ prisma14, prisma17 }) => {
+      const { project, environment } = await seedControlPlaneEnv(prisma14, "token_miss");
+      const { id: waitpointId } = await seedStandaloneToken(prisma14, environment.id, project.id);
+
+      const waitpoint = await resolveWaitpointThroughReadThrough({
+        waitpointId,
+        environmentId: environment.id,
+        read: (client: PrismaReplicaClient) =>
+          client.waitpoint.findFirst({
+            where: { id: waitpointId, environmentId: environment.id },
+          }),
+        deps: {
+          newClient: prisma17 as unknown as PrismaReplicaClient,
+          legacyReplica: prisma17 as unknown as PrismaReplicaClient,
+        },
+      });
+
+      expect(waitpoint).toBeNull();
+    }
+  );
+
+  heteroPostgresTest(
+    "the read gate forces fan-out so a control-plane token resolves while the mint flag is off",
+    async ({ prisma14, prisma17 }) => {
+      const { project, environment } = await seedControlPlaneEnv(prisma14, "token_gate");
+      const { id: waitpointId } = await seedStandaloneToken(prisma14, environment.id, project.id);
+
+      const read = (client: PrismaReplicaClient) =>
+        client.waitpoint.findFirst({
+          where: { id: waitpointId, environmentId: environment.id },
+        });
+
+      const gated = await resolveWaitpointThroughReadThrough({
+        waitpointId,
+        environmentId: environment.id,
+        read,
+        deps: {
+          splitEnabled: true,
+          newClient: prisma17 as unknown as PrismaReplicaClient,
+          legacyReplica: prisma14 as unknown as PrismaReplicaClient,
+        },
+      });
+
+      expect(gated).not.toBeNull();
+      expect(gated?.id).toBe(waitpointId);
+
+      const passthrough = await readThroughRun({
+        runId: waitpointId,
+        environmentId: environment.id,
+        readNew: (c) => read(c),
+        readLegacy: (r) => read(r),
+        deps: {
+          splitEnabled: false,
+          newClient: prisma17 as unknown as PrismaReplicaClient,
+          legacyReplica: prisma14 as unknown as PrismaReplicaClient,
+        },
+      });
+
+      expect(gated).not.toBeNull();
+      expect(passthrough.source).toBe("not-found");
+    }
+  );
+});
