triggerdotdev · d-cs · Jul 6, 2026 · Jul 6, 2026 · Jul 6, 2026 · Jul 6, 2026
diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts
@@ -2097,13 +2097,24 @@ export class RunEngine {
       this.readOnlyPrisma !== this.prisma;
     const prisma = tx ?? (useReplica ? this.readOnlyPrisma : this.prisma);
 
-    const query = async (client: PrismaClientOrTransaction) => {
-      const snapshots = await getExecutionSnapshotsSince(client, runId, snapshotId, this.runStore);
+    const query = async (
+      client: PrismaClientOrTransaction,
+      repairClient?: PrismaClientOrTransaction
+    ) => {
+      const snapshots = await getExecutionSnapshotsSince(
+        client,
+        runId,
+        snapshotId,
+        this.runStore,
+        repairClient
+      );
       return snapshots.map(executionDataFromSnapshot);
     };
 
     try {
-      return await query(prisma);
+      // When reading the replica, pass the primary so a snapshot whose completed-waitpoint join rows
+      // have not replicated yet is repaired from the primary instead of resuming the run waitpoint-less.
+      return await query(prisma, useReplica ? this.prisma : undefined);
     } catch (e) {
       if (useReplica && e instanceof ExecutionSnapshotNotFoundError) {
         // Replica lag: the runner learned this snapshot id from the writer before the
@@ -2115,7 +2126,7 @@ export class RunEngine {
         if (maxMs > 0) {
           await setTimeout(minMs + Math.random() * Math.max(0, maxMs - minMs));
           try {
-            const result = await query(this.readOnlyPrisma);
+            const result = await query(this.readOnlyPrisma, this.prisma);
             this.snapshotsSinceReplicaMissCounter.add(1, { outcome: "replica_retry" });
             return result;
           } catch (replicaRetryError) {

diff --git a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts
@@ -260,7 +260,10 @@ export async function getExecutionSnapshotsSince(
   prisma: PrismaClientOrTransaction,
   runId: string,
   sinceSnapshotId: string,
-  runStore?: RunStore
+  runStore?: RunStore,
+  // The primary, for read-repair when `prisma` is a lagging read replica (see Step 3). Omit when
+  // `prisma` is already the primary.
+  repairClient?: PrismaClientOrTransaction
 ): Promise<EnhancedExecutionSnapshot[]> {
   // Step 1: Find the createdAt of the sinceSnapshotId
   const sinceSnapshot = runStore
@@ -316,10 +319,27 @@ export async function getExecutionSnapshotsSince(
 
   // Step 3: Get waitpoint IDs for the LATEST snapshot only (first in desc order)
   const latestSnapshot = snapshots[0];
-  const waitpointIds = await getSnapshotWaitpointIds(prisma, latestSnapshot.id, runStore);
+  let readClient = prisma;
+  let waitpointIds = await getSnapshotWaitpointIds(prisma, latestSnapshot.id, runStore);
+
+  // Read-repair: completedWaitpointOrder is written on the snapshot row in the same statement as the
+  // snapshot, so it is authoritative. If it lists more waitpoints than the join read returned, the
+  // _completedWaitpoints rows are stale on this client (a lagging read replica) - re-read them from the
+  // primary so the runner is not handed a waitpoint-less continue, which it drops and the run hangs.
+  // Distinct: completedWaitpointOrder can list the same id twice (a run batched under one idempotency
+  // key) while the _completedWaitpoints join is deduped, so compare unique ids or the repair fires every
+  // poll even against a caught-up replica.
+  const expectedCount = new Set(latestSnapshot.completedWaitpointOrder ?? []).size;
+  if (repairClient && repairClient !== prisma && waitpointIds.length < expectedCount) {
+    const repaired = await getSnapshotWaitpointIds(repairClient, latestSnapshot.id, runStore);
+    if (repaired.length > waitpointIds.length) {
+      waitpointIds = repaired;
+      readClient = repairClient;
+    }
+  }
 
   // Step 4: Fetch waitpoints in chunks to avoid NAPI string conversion limits
-  const waitpoints = await fetchWaitpointsInChunks(prisma, waitpointIds, runStore);
+  const waitpoints = await fetchWaitpointsInChunks(readClient, waitpointIds, runStore);
 
   // Step 5: Build enhanced snapshots - only latest gets waitpoints, others get empty arrays
   // The runner only uses completedWaitpoints from the latest snapshot anyway

diff --git a/internal-packages/run-engine/src/engine/tests/getSnapshotsSince.test.ts b/internal-packages/run-engine/src/engine/tests/getSnapshotsSince.test.ts
@@ -3,7 +3,9 @@ import { trace } from "@internal/tracing";
 import { generateFriendlyId } from "@trigger.dev/core/v3/isomorphic";
 import { setTimeout } from "node:timers/promises";
 import { describe, expect } from "vitest";
+import type { PrismaClient } from "@trigger.dev/database";
 import { RunEngine } from "../index.js";
+import { getExecutionSnapshotsSince } from "../systems/executionSnapshotSystem.js";
 import { copySnapshotsToReplica, createTestMetricsMeter } from "./helpers/replicaTestHelpers.js";
 import { setupTestScenario } from "./helpers/snapshotTestHelpers.js";
 import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js";
@@ -1398,4 +1400,131 @@ describe("RunEngine getSnapshotsSince", () => {
       }
     }
   );
+
+  // This models a BATCH resume: production only populates completedWaitpointOrder for waitpoints that
+  // carry a batch index, so the read-repair only ever engages for batch resumes. Single-waitpoint
+  // continues have an empty order (repair is a no-op) and are covered instead by the atomic write
+  // (snapshot + join commit in one tx, which a replica applies atomically).
+  containerTest(
+    "repairs a batch resume's completed-waitpoints from the primary when the replica lags the join rows",
+    async ({ prisma, schemaOnlyPrisma, redisOptions }) => {
+      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+
+      const engine = new RunEngine({
+        prisma,
+        // Replica has the snapshot rows but lags on the _completedWaitpoints join (see below).
+        readOnlyPrisma: schemaOnlyPrisma,
+        readReplicaSnapshotsSinceEnabled: true,
+        readReplicaSnapshotsSinceRetryDelay: { minMs: 1, maxMs: 2 },
+        worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
+        queue: { redis: redisOptions },
+        runLock: { redis: redisOptions },
+        machines: {
+          defaultMachine: "small-1x",
+          machines: {
+            "small-1x": {
+              name: "small-1x" as const,
+              cpu: 0.5,
+              memory: 0.5,
+              centsPerMs: 0.0001,
+            },
+          },
+          baseCostInCents: 0.0001,
+        },
+        tracer: trace.getTracer("test", "0.0.0"),
+      });
+
+      try {
+        // Primary: a run whose latest snapshot is a warm-continue carrying one completed waitpoint -
+        // completedWaitpointOrder is set on the row AND the _completedWaitpoints join + waitpoint exist.
+        const scenario = await setupTestScenario(prisma, authenticatedEnvironment, {
+          totalWaitpoints: 1,
+          outputSizeKB: 1,
+          snapshotConfigs: [
+            { status: "RUN_CREATED", completedWaitpointCount: 0 },
+            { status: "EXECUTING", completedWaitpointCount: 0 },
+            { status: "EXECUTING_WITH_WAITPOINTS", completedWaitpointCount: 0 },
+            { status: "EXECUTING", completedWaitpointCount: 1 },
+          ],
+        });
+        const waitpointId = scenario.waitpoints[0].id;
+
+        // Replica lag: it receives the snapshot ROWS (incl. completedWaitpointOrder) but NOT the
+        // _completedWaitpoints join rows or the waitpoint - the exact state that drops the resume.
+        await copySnapshotsToReplica(prisma, schemaOnlyPrisma, scenario.run.id);
+
+        const result = await engine.getSnapshotsSince({
+          runId: scenario.run.id,
+          snapshotId: scenario.snapshots[0].id,
+        });
+
+        expect(result).not.toBeNull();
+        const latest = result![result!.length - 1];
+        // The runner must receive the completed waitpoint (repaired from the primary), not an empty set.
+        expect(latest.completedWaitpoints.map((w) => w.id)).toEqual([waitpointId]);
+      } finally {
+        await engine.quit();
+      }
+    }
+  );
+
+  // completedWaitpointOrder can contain the same waitpoint id twice (a run batched under one idempotency
+  // key), while the _completedWaitpoints join is deduped. The read-repair must compare DISTINCT ids, or it
+  // fires on every poll for such a snapshot even against a caught-up replica - silently defeating offload.
+  containerTest(
+    "does not repair from the primary when completedWaitpointOrder has duplicates but the join is complete",
+    async ({ prisma }) => {
+      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      // Drives getExecutionSnapshotsSince directly, so it needs no RunEngine (Redis/worker) at all.
+      {
+        const scenario = await setupTestScenario(prisma, authenticatedEnvironment, {
+          totalWaitpoints: 1,
+          outputSizeKB: 1,
+          snapshotConfigs: [
+            { status: "RUN_CREATED", completedWaitpointCount: 0 },
+            { status: "EXECUTING", completedWaitpointCount: 1 },
+          ],
+        });
+        const waitpointId = scenario.waitpoints[0].id;
+        const latestSnapshot = scenario.snapshots[scenario.snapshots.length - 1];
+        // The join stays a single row; the order column lists the same id twice (the batch-dup case).
+        await prisma.taskRunExecutionSnapshot.update({
+          where: { id: latestSnapshot.id },
+          data: { completedWaitpointOrder: [waitpointId, waitpointId] },
+        });
+
+        // Count join reads on the repair client; the replica read (`prisma`) is already complete, so a
+        // correct repair must NEVER touch the repair client for this snapshot.
+        const repairCalls = { queryRaw: 0 };
+        const repairClient = new Proxy(prisma, {
+          get(target, prop) {
+            if (prop === "$queryRaw") {
+              return (...args: unknown[]) => {
+                repairCalls.queryRaw++;
+                return (target as unknown as { $queryRaw: (...a: unknown[]) => unknown }).$queryRaw(
+                  ...args
+                );
+              };
+            }
+            const value = (target as Record<string | symbol, unknown>)[prop];
+            return typeof value === "function" ? value.bind(target) : value;
+          },
+        }) as unknown as PrismaClient;
+
+        const result = await getExecutionSnapshotsSince(
+          prisma,
+          scenario.run.id,
+          scenario.snapshots[0].id,
+          undefined,
+          repairClient
+        );
+
+        const latest = result[result.length - 1];
+        // The duplicate is intentional (batch dup) - enhance expands by completedWaitpointOrder.
+        expect(latest.completedWaitpoints.map((w) => w.id)).toEqual([waitpointId, waitpointId]);
+        // No spurious primary read: distinct(order) === join count, so the repair must not fire.
+        expect(repairCalls.queryRaw).toBe(0);
+      }
+    }
+  );
 });
diff --git a/internal-packages/run-store/src/PostgresRunStore.test.ts b/internal-packages/run-store/src/PostgresRunStore.test.ts
@@ -888,6 +888,124 @@ describe("PostgresRunStore", () => {
     }
   );
 
+  // lockRunToWorker creates a PENDING_EXECUTING snapshot (which can inherit completed waitpoints) and
+  // then connects the _completedWaitpoints join as a separate statement. These must commit together, or
+  // a replica-served /snapshots/since read can return the snapshot waitpoint-less and drop the resume.
+  postgresTest(
+    "lockRunToWorker writes the snapshot and its completed-waitpoint links atomically",
+    async ({ prisma }) => {
+      const { organization, project, environment } = await seedEnvironment(prisma);
+      const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });
+      const runId = "run_lock_atomic";
+
+      await store.createRun(
+        buildCreateRunInput({
+          runId,
+          organizationId: organization.id,
+          projectId: project.id,
+          runtimeEnvironmentId: environment.id,
+        })
+      );
+
+      const backgroundWorker = await prisma.backgroundWorker.create({
+        data: {
+          friendlyId: "worker_atomic",
+          version: "20260601.1",
+          runtimeEnvironmentId: environment.id,
+          projectId: project.id,
+          contentHash: "abc",
+          sdkVersion: "3.0.0",
+          cliVersion: "3.0.0",
+          metadata: {},
+        },
+      });
+      const workerTask = await prisma.backgroundWorkerTask.create({
+        data: {
+          friendlyId: "task_atomic",
+          slug: "my-task",
+          filePath: "src/my-task.ts",
+          exportName: "myTask",
+          workerId: backgroundWorker.id,
+          runtimeEnvironmentId: environment.id,
+          projectId: project.id,
+        },
+      });
+      const queue = await prisma.taskQueue.create({
+        data: {
+          friendlyId: "queue_atomic",
+          name: "task/my-task",
+          runtimeEnvironmentId: environment.id,
+          projectId: project.id,
+        },
+      });
+      const priorSnapshot = await prisma.taskRunExecutionSnapshot.create({
+        data: {
+          engine: "V2",
+          executionStatus: "RUN_CREATED",
+          description: "prior",
+          runStatus: "PENDING",
+          environmentId: environment.id,
+          environmentType: "DEVELOPMENT",
+          projectId: project.id,
+          organizationId: organization.id,
+          runId,
+        },
+      });
+      const waitpoint = await prisma.waitpoint.create({
+        data: {
+          friendlyId: "wp_lock_atomic",
+          type: "MANUAL",
+          status: "COMPLETED",
+          idempotencyKey: "idem-lock-atomic",
+          userProvidedIdempotencyKey: false,
+          projectId: project.id,
+          environmentId: environment.id,
+        },
+      });
+
+      // Force the completed-waitpoint join insert to fail mid-write.
+      await prisma.$executeRawUnsafe('DROP TABLE "_completedWaitpoints"');
+
+      const snapshotId = "snap_lock_atomic";
+      await expect(
+        // Base client as `tx` = how the dequeue path calls it; the store must still open its own tx.
+        store.lockRunToWorker(
+          runId,
+          {
+            lockedAt: new Date(),
+            lockedById: workerTask.id,
+            lockedToVersionId: backgroundWorker.id,
+            lockedQueueId: queue.id,
+            startedAt: new Date(),
+            baseCostInCents: 5,
+            machinePreset: "small-1x",
+            taskVersion: "20260601.1",
+            sdkVersion: "3.0.0",
+            cliVersion: "3.0.0",
+            maxDurationInSeconds: null,
+            snapshot: {
+              id: snapshotId,
+              previousSnapshotId: priorSnapshot.id,
+              environmentId: environment.id,
+              environmentType: "DEVELOPMENT",
+              projectId: project.id,
+              organizationId: organization.id,
+              completedWaitpointIds: [waitpoint.id],
+              completedWaitpointOrder: [waitpoint.id],
+            },
+          },
+          prisma
+        )
+      ).rejects.toThrow();
+
+      // Atomic: neither the snapshot nor the run lock persists without the links.
+      const snap = await prisma.taskRunExecutionSnapshot.findUnique({ where: { id: snapshotId } });
+      expect(snap).toBeNull();
+      const run = await prisma.taskRun.findUniqueOrThrow({ where: { id: runId } });
+      expect(run.status).not.toBe("DEQUEUED");
+    }
+  );
+
   postgresTest(
     "parkPendingVersion sets status to PENDING_VERSION and stores statusReason",
     async ({ prisma }) => {