Skip to content

Commit 717e7f0

Browse files
d-csclaude
andcommitted
perf(run-engine): keep the record gate allocation-free, pin the resolver's absence rules
Two reviews of the resolver rewrite raised four things worth acting on. The resume-path gate built a mapped array, a filtered array and a Set before discovering there was nothing to ask for, which is the universal case today and stays the common one through a partial rollout. It now short-circuits on .some first, so the gate allocates nothing until a blocking waitpoint really is store-format, and stops scanning at the first one that is. The run-output reader no longer accepts a read client. The router reads the owning store's replica when none is passed and forces its primary for any client that is not replica-branded, so accepting one invited a wide output read onto the writer by reflex. This read cannot need read-your-writes: the child committed its output before completing the waitpoint that unblocked the parent. Three absence rules had no test holding them. An empty-string output is a value, and the reader's explicit null check is what keeps it from being reported as a lost output on a run that completed normally. Two waitpoints completed by the same run must read it once and both still receive it. And the differential suite never compared a cycle with two deferring records, so the batched read had no oracle covering the shape it exists for. Also throws the missing-reader error where the deferring record is found, so the message names a record that really defers and needs no optional chaining. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1 parent 0308288 commit 717e7f0

4 files changed

Lines changed: 118 additions & 21 deletions

File tree

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

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -778,6 +778,15 @@ export class WaitpointSystem {
778778
runId: string,
779779
blockingWaitpoints: RunBlockEdge[]
780780
): Promise<CompletedWaitpointRecord[] | undefined> {
781+
// `.some` before the dedup, so the gate allocates nothing in the case that is universal
782+
// today and stays common through a partial rollout: no blocking waitpoint is store-format.
783+
// Building the mapped array, the filtered array and the Set first meant three allocations
784+
// per resume -- for a 1000-wide batch fan-in too -- to discover there was nothing to ask for.
785+
if (!blockingWaitpoints.some((b) => parseWaitpointId(b.waitpoint.id).format === "b32hexW")) {
786+
return undefined;
787+
}
788+
789+
// Only a set that really holds one pays for the dedup.
781790
const storeFormatIds = [
782791
...new Set(
783792
blockingWaitpoints
@@ -786,10 +795,6 @@ export class WaitpointSystem {
786795
),
787796
];
788797

789-
if (storeFormatIds.length === 0) {
790-
return undefined;
791-
}
792-
793798
const sources = await this.coordinator.readCompletionEnvelopes({
794799
runId,
795800
waitpointIds: storeFormatIds,

‎internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointEquivalence.test.ts‎

Lines changed: 41 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ import { postgresTest } from "@internal/testcontainers";
55
import { PostgresRunStore } from "@internal/run-store";
66
import type { PrismaClient, Waitpoint } from "@trigger.dev/database";
77
import { describe, expect } from "vitest";
8-
import { seedChildRunWithOutput } from "./testFixtures/childRun.js";
8+
import { seedChildRunsWithOutputs, seedChildRunWithOutput } from "./testFixtures/childRun.js";
99
import { enhanceExecutionSnapshotWithWaitpoints } from "../systems/executionSnapshotSystem.js";
1010
import { buildCompletedWaitpointRecords } from "./completedWaitpointRecords.js";
1111
import {
@@ -192,6 +192,46 @@ describe("the resolver reproduces the existing hydration", () => {
192192
expect(actual).toEqual(expected);
193193
});
194194

195+
// Two deferring records in ONE cycle, which is the shape a batch fan-in produces and the one
196+
// the batched read introduced. Every other case here defers at most once, so a mis-keyed map
197+
// or a swapped output would compare equal against the oracle in all of them.
198+
//
199+
// The outputs differ deliberately, and one run sits at two interleaved positions, so a swap
200+
// between the two runs and a lost second position both show up as a diff.
201+
postgresTest("for two RUN waitpoints deferring to different runs", async ({ prisma }) => {
202+
const [runA, runB] = await seedChildRunsWithOutputs(prisma, ['{"child":"a"}', '{"child":"b"}']);
203+
204+
const { expected, actual } = await bothPaths(
205+
prisma,
206+
[
207+
pair({
208+
id: "wp_run_a",
209+
type: "RUN",
210+
output: '{"child":"a"}',
211+
completedByTaskRunId: runA,
212+
}),
213+
pair({
214+
id: "wp_run_b",
215+
type: "RUN",
216+
output: '{"child":"b"}',
217+
completedByTaskRunId: runB,
218+
}),
219+
],
220+
["wp_run_a", "wp_run_b", "wp_run_a"],
221+
BATCH_ID
222+
);
223+
224+
expect(actual).toEqual(expected);
225+
// Stated as well as compared: the oracle agreeing is the assertion, but a reader should not
226+
// have to run it to see that three entries come back and each output landed on its own id.
227+
expect(actual).toHaveLength(3);
228+
expect(actual.filter((w) => w.id === "wp_run_a").map((w) => w.output)).toEqual([
229+
'{"child":"a"}',
230+
'{"child":"a"}',
231+
]);
232+
expect(actual.find((w) => w.id === "wp_run_b")?.output).toBe('{"child":"b"}');
233+
});
234+
195235
postgresTest("for a RUN waitpoint read under a batch", async ({ prisma }) => {
196236
const childRunId = await seedChildRunWithOutput(prisma, '{"ok":true}');
197237
const { expected, actual } = await bothPaths(

‎internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.runOutput.test.ts‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -175,6 +175,46 @@ describe("the deriveFromRun branch", () => {
175175
}
176176
});
177177

178+
// An empty string is a VALUE, not an absence. The reader's `row.output !== null` is what keeps
179+
// it: narrowed to a truthy test it would report the run as output-less and throw
180+
// lost-run-output on a run that completed perfectly well. Nothing else pins that, and
181+
// `chooseOutput` already treats empty as a value on the write side.
182+
postgresTest("keeps an empty-string output rather than refusing", async ({ prisma }) => {
183+
const runId = await seedChildRunWithOutput(prisma, "");
184+
185+
const [entry] = await resolverFor(prisma)({
186+
runId: "run_parent",
187+
pointer: { cycleSeq: 1, count: 0 },
188+
order: [],
189+
distinctIds: ["wp_run"],
190+
records: [deriveRecord(runId)],
191+
});
192+
193+
expect(entry?.output).toBe("");
194+
});
195+
196+
// Two waitpoints completed by the SAME run. One id in one batch, and both entries carry it:
197+
// the dedup must not cost the second waitpoint its output.
198+
postgresTest("reads a shared run once and hydrates both waitpoints", async ({ prisma }) => {
199+
const runId = await seedChildRunWithOutput(prisma, '{"shared":true}');
200+
const { resolve, batches } = countingResolver(prisma);
201+
202+
const result = await resolve({
203+
runId: "run_parent",
204+
pointer: { cycleSeq: 1, count: 2 },
205+
order: ["wp_first", "wp_second"],
206+
distinctIds: ["wp_first", "wp_second"],
207+
records: [
208+
{ ...deriveRecord(runId), id: "wp_first", friendlyId: "waitpoint_wp_first" },
209+
{ ...deriveRecord(runId), id: "wp_second", friendlyId: "waitpoint_wp_second" },
210+
],
211+
});
212+
213+
expect(batches).toEqual([[runId]]);
214+
expect(result).toHaveLength(2);
215+
expect(result.map((w) => w.output)).toEqual(['{"shared":true}', '{"shared":true}']);
216+
});
217+
178218
// A cycle that defers nothing reads nothing, so an all-inline resume pays no Postgres round
179219
// trip at all.
180220
postgresTest("reads nothing when no record defers", async ({ prisma }) => {

‎internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.ts‎

Lines changed: 28 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
import type {
22
CompletedWaitpointRecord,
3-
ReadClient,
43
ResolveCompletedWaitpointsArgs,
54
RunStore,
65
} from "@internal/run-store";
@@ -64,17 +63,23 @@ const RUN_OUTPUT_CHUNK_SIZE = 100;
6463
* `findRunsByIds` is the store's own grouped replacement for `Promise.all(ids.map(findRun))`,
6564
* and it forces `id` into the projection so the map keys correctly even though this select
6665
* names only `output`.
66+
*
67+
* Takes no read client on purpose. The router reads the owning store's REPLICA when no client is
68+
* passed, and forces its PRIMARY for any client that is not replica-branded -- so accepting one
69+
* would let a caller put a wide `TaskRun.output` read on the writer by reflex. This read does not
70+
* need read-your-writes: the child run committed its output before it completed the waitpoint that
71+
* unblocked this parent, so replica lag cannot hide it. That is the opposite of the envelope read
72+
* in the legacy arm, which reads a waitpoint completed moments earlier and must use the writer.
6773
*/
6874
export function createRunOutputsReader(
69-
runStore: Pick<RunStore, "findRunsByIds">,
70-
client?: ReadClient
75+
runStore: Pick<RunStore, "findRunsByIds">
7176
): (taskRunIds: string[]) => Promise<Map<string, string>> {
7277
return async (taskRunIds) => {
7378
const outputs = new Map<string, string>();
7479

7580
for (let i = 0; i < taskRunIds.length; i += RUN_OUTPUT_CHUNK_SIZE) {
7681
const chunk = taskRunIds.slice(i, i + RUN_OUTPUT_CHUNK_SIZE);
77-
const rows = await runStore.findRunsByIds(chunk, { select: { output: true } }, client);
82+
const rows = await runStore.findRunsByIds(chunk, { select: { output: true } });
7883

7984
for (const [id, row] of rows) {
8085
// A row present with a null output is the same absence as a missing row: either way the
@@ -218,27 +223,34 @@ async function readDeferredOutputs(
218223
deps: CompletedWaitpointResolverDeps
219224
): Promise<Map<string, string>> {
220225
const runIds = new Set<string>();
221-
let deferring: CompletedWaitpointRecord | undefined;
226+
let read: NonNullable<CompletedWaitpointResolverDeps["readRunOutputs"]> | undefined;
222227

223228
for (const record of records) {
224229
const runId = deferredRunIdOf(record);
225-
if (runId !== undefined) {
226-
runIds.add(runId);
227-
deferring ??= record;
230+
if (runId === undefined) {
231+
continue;
228232
}
229-
}
230233

231-
if (runIds.size === 0) {
232-
return new Map();
234+
// Checked here rather than after the pass, so the message names a record that really does
235+
// defer rather than whichever one happened to be first. A missing reader is a wiring fault,
236+
// so every deferring record is equally implicated and the first is as informative as any.
237+
if (!deps.readRunOutputs) {
238+
throw new Error(
239+
`Waitpoint ${record.id} defers its output to run ${runId}, but the resolver was built with no run-output reader.`
240+
);
241+
}
242+
243+
read = deps.readRunOutputs;
244+
runIds.add(runId);
233245
}
234246

235-
if (!deps.readRunOutputs) {
236-
throw new Error(
237-
`Waitpoint ${deferring?.id} defers its output to run ${deferring?.completedByTaskRunId}, but the resolver was built with no run-output reader.`
238-
);
247+
// Set exactly when a record deferred, so this is the same condition as an empty `runIds` --
248+
// and unlike `runIds.size`, it carries the reader with it.
249+
if (read === undefined) {
250+
return new Map();
239251
}
240252

241-
return deps.readRunOutputs([...runIds]);
253+
return read([...runIds]);
242254
}
243255

244256
// The run a record defers its output to, or undefined when it carries its own output or has

0 commit comments

Comments
 (0)