feat(run-engine,run-store): completed-waitpoint envelope, read-time resolver, and the fail-loud coverage check

internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts

@@ -40,7 +40,7 @@ const _pointerKeys: Exact<keyof CompletedWaitpointsPointer, "cycleSeq" | "count"
 
 const _argsKeys: Exact<
   keyof ResolveCompletedWaitpointsArgs,
-  "runId" | "batchId" | "pointer" | "order" | "records"
+  "runId" | "batchId" | "pointer" | "order" | "distinctIds" | "records"
 > = true;
 import { enhanceExecutionSnapshotWithWaitpoints } from "./executionSnapshotSystem.js";
 
@@ -193,6 +193,7 @@ async function assertParity(
     batchId: batchId ?? undefined,
     pointer: { cycleSeq: 1, count: order.length },
     order,
+    distinctIds: [...new Set(waitpoints.map((w) => w.id))],
     records: waitpoints.map(toRecord),
   };
   // count-carried-forward behaviour (order.length, not the record count) is covered by
@@ -406,6 +407,7 @@ describe("the completed-waitpoints freeze", () => {
       batchId: undefined,
       pointer: { cycleSeq: 1, count: 1 },
       order: ["wp_hook"],
+      distinctIds: ["wp_hook"],
       records: [toRecord(w)],
     });
     expect(resolved).toHaveLength(1);
@@ -611,6 +613,7 @@ describe("the exhaustive parity grid", () => {
                           batchId: readingBatchId ?? undefined,
                           pointer: { cycleSeq: 1, count: order.length },
                           order,
+                          distinctIds: [w.id],
                           records: [toRecord(w)],
                         };
                         const resolved = await referenceResolver(

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

@@ -4,7 +4,7 @@ import type {
   TaskRun,
   TaskRunExecutionStatus,
 } from "@trigger.dev/database";
-import type { RunStore } from "@internal/run-store";
+import type { CompletedWaitpointRecord, RunStore } from "@internal/run-store";
 import { parseNaturalLanguageDuration } from "@trigger.dev/core/v3/isomorphic";
 import type { MinimalAuthenticatedEnvironment } from "../../shared/index.js";
 import { QUEUED_SNAPSHOT_DESCRIPTION, QUEUED_SNAPSHOT_STATUS } from "../consts.js";
@@ -34,6 +34,7 @@ export class EnqueueSystem {
     batchId,
     checkpointId,
     completedWaitpoints,
+    completedWaitpointRecords,
     workerId,
     runnerId,
     skipRunLock,
@@ -57,6 +58,7 @@ export class EnqueueSystem {
       id: string;
       index?: number;
     }[];
+    completedWaitpointRecords?: CompletedWaitpointRecord[];
     workerId?: string;
     runnerId?: string;
     skipRunLock?: boolean;
@@ -108,6 +110,7 @@ export class EnqueueSystem {
           organizationId: env.organization.id,
           checkpointId,
           completedWaitpoints,
+          completedWaitpointRecords,
           workerId,
           runnerId,
         },

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

@@ -10,7 +10,7 @@ import type {
   TaskRunStatus,
   Waitpoint,
 } from "@trigger.dev/database";
-import type { RunStore } from "@internal/run-store";
+import type { CompletedWaitpointRecord, RunStore } from "@internal/run-store";
 import { ExecutionSnapshotNotFoundError, ServiceValidationError } from "../errors.js";
 import type { HeartbeatTimeouts } from "../types.js";
 import type { SystemResources } from "./systems.js";
@@ -200,6 +200,67 @@ 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
@@ -449,6 +510,7 @@ export class ExecutionSnapshotSystem {
       workerId,
       runnerId,
       completedWaitpoints,
+      completedWaitpointRecords,
       error,
     }: {
       run: { id: string; status: TaskRunStatus; attemptNumber?: number | null };
@@ -470,6 +532,7 @@ export class ExecutionSnapshotSystem {
         id: string;
         index?: number;
       }[];
+      completedWaitpointRecords?: CompletedWaitpointRecord[];
       error?: string;
     },
     // When set (inside runStore.runInTransaction), the snapshot write goes through the owning store
@@ -492,6 +555,7 @@ export class ExecutionSnapshotSystem {
         workerId,
         runnerId,
         completedWaitpoints,
+        completedWaitpointRecords,
         error,
       },
       prisma

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

@@ -1,5 +1,7 @@
 import { timeoutError } from "@trigger.dev/core/v3";
+import { parseWaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import type { ShardKey } from "@trigger.dev/core/v3/isomorphic";
+import type { CompletedWaitpointRecord } from "@internal/run-store";
 import type {
   PrismaClientOrTransaction,
   TaskRun,
@@ -11,7 +13,8 @@ import { assertNever } from "assert-never";
 import { sendNotificationToWorker } from "../eventBus.js";
 import { isFinalRunStatus } from "../statuses.js";
 import { LegacyPostgresWaitpointCoordinator } from "../waitpointCoordinator/legacyPostgresCoordinator.js";
-import type { WaitpointCoordinator } from "../waitpointCoordinator/types.js";
+import { buildCompletedWaitpointRecords } from "../waitpointCoordinator/completedWaitpointRecords.js";
+import type { RunBlockEdge, WaitpointCoordinator } from "../waitpointCoordinator/types.js";
 import type { EnqueueSystem } from "./enqueueSystem.js";
 import type { ExecutionSnapshotSystem } from "./executionSnapshotSystem.js";
 import { getLatestExecutionSnapshot } from "./executionSnapshotSystem.js";
@@ -605,6 +608,13 @@ export class WaitpointSystem {
           };
         }
         case "EXECUTING_WITH_WAITPOINTS": {
+          // Built inside the branch, not before the switch: the statuses above return without
+          // appending, and they must not pay an envelope read to do it.
+          const completedWaitpointRecords = await this.#completedWaitpointRecordsFor(
+            runId,
+            blockingWaitpoints
+          );
+
           const newSnapshot = await this.executionSnapshotSystem.createExecutionSnapshot(
             this.$.prisma,
             {
@@ -627,6 +637,7 @@ export class WaitpointSystem {
                 id: b.waitpoint.id,
                 index: b.batchIndex ?? undefined,
               })),
+              ...(completedWaitpointRecords && { completedWaitpointRecords }),
             }
           );
 
@@ -672,6 +683,11 @@ export class WaitpointSystem {
             );
           }
 
+          const completedWaitpointRecords = await this.#completedWaitpointRecordsFor(
+            runId,
+            blockingWaitpoints
+          );
+
           //put it back in the queue, with the original timestamp (w/ priority)
           //this prioritizes dequeuing waiting runs over new runs
           const newSnapshot = await this.enqueueSystem.enqueueRun({
@@ -686,6 +702,7 @@ export class WaitpointSystem {
               id: b.waitpoint.id,
               index: b.batchIndex ?? undefined,
             })),
+            ...(completedWaitpointRecords && { completedWaitpointRecords }),
             checkpointId: snapshot.checkpointId ?? undefined,
           });
 
@@ -738,6 +755,54 @@ export class WaitpointSystem {
     });
   }
 
+  /**
+   * The record set for one resume, or undefined when no blocking waitpoint carries a store-format
+   * id.
+   *
+   * Gated on id FORMAT, not residency. The two are not the same during a migration: a
+   * store-format id can still be served by the Postgres arm, exactly as run-ops ids were for
+   * runs. Whichever arm owns it answers, so the gate only decides whether to ask at all.
+   *
+   * 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,
+    blockingWaitpoints: RunBlockEdge[]
+  ): Promise<CompletedWaitpointRecord[] | undefined> {
+    // `.some` before the dedup, so the gate allocates nothing in the case that is universal
+    // today and stays common through a partial rollout: no blocking waitpoint is store-format.
+    // Building the mapped array, the filtered array and the Set first meant three allocations
+    // per resume -- for a 1000-wide batch fan-in too -- to discover there was nothing to ask for.
+    if (!blockingWaitpoints.some((b) => parseWaitpointId(b.waitpoint.id).format === "b32hexW")) {
+      return undefined;
+    }
+
+    // Only a set that really holds one pays for the dedup.
+    const storeFormatIds = [
+      ...new Set(
+        blockingWaitpoints
+          .map((b) => b.waitpoint.id)
+          .filter((id) => parseWaitpointId(id).format === "b32hexW")
+      ),
+    ];
+
+    const sources = await this.coordinator.readCompletionEnvelopes({
+      runId,
+      waitpointIds: storeFormatIds,
+    });
+
+    return buildCompletedWaitpointRecords(sources);
+  }
+
   /**
    * Builds the waitpoint output payload from a completed run's stored output/error.
    */