perf(run-store,run-engine): take the records read off the copy-forward path

The append script now sources a refused carry-forward's record set from the
cycle it replaces, rather than the caller pre-reading it and passing it in.
The pre-read cost one HGET plus a parse of the whole record blob on every
copy-forward append -- attempt start, dequeue, checkpoint -- to serve a branch
that needs partial eviction to reach. Reading it in the branch that uses it is
also atomic with the mint, so no reader can observe a replacement cycle whose
ids have no records.

Projects the legacy arm's envelope read to the columns the envelope is built
from, so it stops reading tags and the ownership and timestamp columns it does
not use on a read that lands on the writer inside the run lock.

Corrects a docstring that described the extra read as happening on the resume
path: the resume path supplies the records and was the one path that never
read. Records the rollout-ordering constraint that keeps the format gate cheap.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: d-cs

internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts

@@ -173,7 +173,7 @@ async function getSnapshotWaitpointIdsWithPresence(
  * This is necessary because waitpoints can have large outputs (100KB+),
  * and fetching many at once can exceed Node.js string limits.
  */
-export async function fetchWaitpointsInChunks(
+async function fetchWaitpointsInChunks(
   prisma: PrismaClientOrTransaction,
   waitpointIds: string[],
   runStore?: RunStore,
@@ -200,6 +200,67 @@ export async function fetchWaitpointsInChunks(
   return allWaitpoints;
 }
 
+/**
+ * The columns the completion envelope is built from, and only those.
+ *
+ * `fetchWaitpointsInChunks` reads whole rows because the snapshot hydration builds a full
+ * executor waitpoint from them. The envelope needs fewer, so projecting drops `tags`,
+ * `projectId`, `environmentId`, `createdAt`, `updatedAt` and `idempotencyKeyExpiresAt` from a
+ * read that happens on the writer inside the run lock. It cannot drop `output`, which is the
+ * 100KB+ column and also the payload the envelope exists to carry.
+ *
+ * `status` is here for the arm's COMPLETED filter, not for the mapper.
+ */
+const WAITPOINT_ENVELOPE_SELECT = {
+  id: true,
+  friendlyId: true,
+  type: true,
+  status: true,
+  completedAt: true,
+  output: true,
+  outputType: true,
+  outputIsError: true,
+  completedByTaskRunId: true,
+  completedByBatchId: true,
+  completedAfter: true,
+  idempotencyKey: true,
+  userProvidedIdempotencyKey: true,
+  inactiveIdempotencyKey: true,
+} satisfies Prisma.WaitpointSelect;
+
+export type WaitpointEnvelopeRow = Pick<Waitpoint, keyof typeof WAITPOINT_ENVELOPE_SELECT>;
+
+/**
+ * The projected sibling of `fetchWaitpointsInChunks`, for the envelope read.
+ *
+ * Chunked identically, and for the same reason: a waitpoint output can be 100KB+, so a large
+ * fan-in read whole can exceed Node's string limits. `boundedIn` pads for plan-cache stability,
+ * it does not bound the set. `runId` is the routing hint the router needs to read the run's own
+ * store instead of fanning every chunk across both run-ops databases.
+ */
+export async function fetchWaitpointEnvelopeRowsInChunks(
+  prisma: PrismaClientOrTransaction,
+  waitpointIds: string[],
+  runStore?: RunStore,
+  runId?: string
+): Promise<WaitpointEnvelopeRow[]> {
+  if (waitpointIds.length === 0) return [];
+
+  const rows: WaitpointEnvelopeRow[] = [];
+  for (let i = 0; i < waitpointIds.length; i += WAITPOINT_CHUNK_SIZE) {
+    const chunk = waitpointIds.slice(i, i + WAITPOINT_CHUNK_SIZE);
+    const args = {
+      where: { id: { in: boundedIn(chunk) } },
+      select: WAITPOINT_ENVELOPE_SELECT,
+    };
+    const found = runStore
+      ? await runStore.findManyWaitpoints(args, prisma, runId)
+      : await prisma.waitpoint.findMany(args);
+    rows.push(...found);
+  }
+  return rows;
+}
+
 /**
  * Gets the most recent valid snapshot for a run. When `environmentId` is provided the read is scoped
  * to that environment (tenant boundary): a run in another environment reads as not-found and rejects

internal-packages/run-engine/src/engine/systems/waitpointSystem.ts

@@ -766,6 +766,13 @@ export class WaitpointSystem {
    * That gate is what keeps this inert. `parseWaitpointId` reports legacy for every id minted
    * today, so no live resume reads an envelope or writes a record until a waitpoint mints in
    * store format.
+   *
+   * ROLLOUT ORDER. Because the gate reads the id and not the organisation, the first store-format
+   * mint is what starts the cost, for every organisation that then holds one -- not the first
+   * snapshot-store flip. The arm that answers here is the Postgres one until a store arm is
+   * wired, so a mint enabled ahead of the store means a projected `Waitpoint` read on the
+   * WRITER, inside the run lock, once per resume. That is bounded and correct, but it is not
+   * free, so store-format minting should follow the snapshot store rather than lead it.
    */
   async #completedWaitpointRecordsFor(
     runId: string,

internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts

@@ -6,7 +6,7 @@ import type { PrismaClient, Waitpoint } from "@trigger.dev/database";
 import { boundedIn, Prisma } from "@trigger.dev/database";
 import { nanoid } from "nanoid";
 import { UnclassifiableWaitpointId } from "../errors.js";
-import { fetchWaitpointsInChunks } from "../systems/executionSnapshotSystem.js";
+import { fetchWaitpointEnvelopeRowsInChunks } from "../systems/executionSnapshotSystem.js";
 import { envelopeSourceFromWaitpointRow } from "./completionEnvelopeSource.js";
 import type {
   AssociatedWaitpointData,
@@ -96,6 +96,10 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator
    * Chunked for the same reason the snapshot hydration chunks: a waitpoint output can be
    * 100KB+, and a large fan-in read whole can exceed Node's string limits. `boundedIn` pads
    * for plan-cache stability, it does not bound the set.
+   *
+   * Projected to the columns the envelope is built from. This read lands on the writer, inside
+   * the run lock, and the hydration reads the same rows again afterwards for a store-format
+   * resume that Postgres still serves -- so it takes no more than it uses.
    */
   async readCompletionEnvelopes({
     runId,
@@ -105,7 +109,12 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator
       return [];
     }
 
-    const rows = await fetchWaitpointsInChunks(this.prisma, waitpointIds, this.runStore, runId);
+    const rows = await fetchWaitpointEnvelopeRowsInChunks(
+      this.prisma,
+      waitpointIds,
+      this.runStore,
+      runId
+    );
 
     // COMPLETED only, so both arms honour one omission contract. The store arm cannot return a
     // pending waitpoint because a pending one has no completion to read; this arm reads rows by

internal-packages/run-store/src/redisSnapshotStore.recordsOnCarryRefusal.test.ts

@@ -97,9 +97,9 @@ describe("a refused carry-forward", () => {
   });
 
   // The reachable production shape. Every copy-forward append (dequeue, checkpoint, attempt)
-  // re-passes the same refs and carries no records of its own, so this is the case a refusal
-  // actually meets. Before the decorator read the surviving cycle's records, this minted a
-  // replacement holding ids with no records, permanently.
+  // re-passes the same refs and carries NO records of its own, and no longer pre-reads them: the
+  // append script sources the record set from the cycle it is replacing. So a refusal preserves
+  // the records without the caller having paid a read on every copy-forward that did not refuse.
   redisTest("keeps the records when the caller carried refs but none", async ({ redisOptions }) => {
     const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 });
     const raw = createRedisClient(redisOptions, { onError: () => {} });
@@ -115,35 +115,85 @@ describe("a refused carry-forward", () => {
         },
       });
 
-      // What the decorator now does for a records-less carry: read the surviving cycle's
-      // records and carry those into the refusal branch.
-      const carried = await store.getCycleRecords("run_1", 1);
-      expect(carried).toHaveLength(1);
-
+      // Lose the four core keys, keeping the cycle key. This is the shape that makes the store
+      // refuse the pointer: the seq counter is behind cycleSeqIn while wp:1 still lives.
       await raw.del("snap:{run_1}:e", "snap:{run_1}:idx", "snap:{run_1}:cur", "snap:{run_1}:seq");
 
-      await store.append({
+      const carried = await store.append({
         entry: entry({ id: "snap_2" }),
         kind: "birth",
         isTerminal: false,
+        // No `records`, exactly as a copy-forward append passes it.
         cycle: {
           kind: "carryForward",
           cycleSeq: 1,
           completedWaitpoints: [{ id: "w_a", index: 0 }],
-          records: carried,
         },
       });
 
+      expect(carried).toMatchObject({ outcome: "written", cycleMismatch: true });
+
+      // The replacement holds the records the refused cycle held, copied inside the script.
       const read = await store.getLatest("run_1");
       const records = await recordsAt(raw, read!.cycle!.cycleSeq);
 
       expect(records).toHaveLength(1);
       expect(records?.[0]?.id).toBe("w_a");
+      expect(records?.[0]?.output).toEqual({ inline: "first" });
     } finally {
       await Promise.all([store.quit(), raw.quit().catch(() => {})]);
     }
   });
 
+  // The other half of the refusal: the cycle key itself is gone, so there are no records anywhere
+  // and the replacement legitimately holds none. It must not inherit a stale set from a re-minted
+  // cycleSeq whose key survived.
+  redisTest(
+    "mints a records-less replacement when the cycle key is gone",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 });
+      const raw = createRedisClient(redisOptions, { onError: () => {} });
+      try {
+        await store.append({
+          entry: entry({ id: "snap_1" }),
+          kind: "birth",
+          isTerminal: false,
+          cycle: {
+            kind: "new",
+            completedWaitpoints: [{ id: "w_a", index: 0 }],
+            records: [record("w_a", "first")],
+          },
+        });
+
+        await raw.del(
+          "snap:{run_1}:e",
+          "snap:{run_1}:idx",
+          "snap:{run_1}:cur",
+          "snap:{run_1}:seq",
+          "snap:{run_1}:wp:1"
+        );
+
+        const carried = await store.append({
+          entry: entry({ id: "snap_2" }),
+          kind: "birth",
+          isTerminal: false,
+          cycle: {
+            kind: "carryForward",
+            cycleSeq: 1,
+            completedWaitpoints: [{ id: "w_b", index: 0 }],
+          },
+        });
+
+        expect(carried).toMatchObject({ outcome: "written", cycleMismatch: true });
+
+        const read = await store.getLatest("run_1");
+        expect(await recordsAt(raw, read!.cycle!.cycleSeq)).toBeUndefined();
+      } finally {
+        await Promise.all([store.quit(), raw.quit().catch(() => {})]);
+      }
+    }
+  );
+
   // Without refs there is nothing to mint from, so the entry is written with no pointer. That is
   // the older behaviour and it stays: no pointer is safe, a pointer with no records is not.
   redisTest("writes no pointer when the caller carried no refs", async ({ redisOptions }) => {

internal-packages/run-store/src/redisSnapshotStore.test.ts

@@ -4,9 +4,7 @@ import { expect, describe, vi } from "vitest";
 import { redisTest, slotOf } from "@internal/testcontainers";
 import { createRedisClient } from "@internal/redis";
 import { Logger } from "@trigger.dev/core/logger";
-import { generateWaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import {
-  mayHoldRecords,
   snapshotKeys,
   deriveOrder,
   isValidFor,
@@ -26,46 +24,6 @@ describe("snapshotKeys", () => {
   });
 });
 
-describe("mayHoldRecords", () => {
-  it("is false for an empty set", () => {
-    expect(mayHoldRecords([])).toBe(false);
-  });
-
-  it("is false when every id is legacy", () => {
-    expect(mayHoldRecords([{ id: "w_cuid_a", index: 0 }, { id: "w_cuid_b" }])).toBe(false);
-  });
-
-  it("is true when one id is store-format", () => {
-    expect(mayHoldRecords([{ id: generateWaitpointId("MANUAL"), index: 0 }])).toBe(true);
-  });
-
-  // The mixed case is the one a per-organisation rollout produces: a run holding waitpoints
-  // minted either side of the flip. One store-format id is enough to require the read.
-  it("is true when a store-format id sits among legacy ones", () => {
-    expect(
-      mayHoldRecords([
-        { id: "w_cuid_a", index: 0 },
-        { id: generateWaitpointId("MANUAL"), index: 1 },
-        { id: "w_cuid_c" },
-      ])
-    ).toBe(true);
-  });
-
-  it("reads the id, not the index, so an index-less store id still counts", () => {
-    expect(mayHoldRecords([{ id: generateWaitpointId("DATETIME") }])).toBe(true);
-  });
-
-  it("counts every store waitpoint type", () => {
-    for (const t of ["RUN", "BATCH", "DATETIME", "MANUAL"] as const) {
-      expect(mayHoldRecords([{ id: generateWaitpointId(t) }])).toBe(true);
-    }
-  });
-
-  it("is false for a foreign prefix that is not a waitpoint id", () => {
-    expect(mayHoldRecords([{ id: "run_0123456789abcdefghijklm" }])).toBe(false);
-  });
-});
-
 describe("deriveOrder", () => {
   it("drops entries with no index, sorts by index, and maps to id", () => {
     expect(