Skip to content

Commit 8140d17

Browse files
d-csclaude
andcommitted
perf(run-engine): gate the record build on an O(1) predicate, not a fan-in scan
Deciding whether to build a completed-waitpoint record set meant walking every blocking waitpoint and classifying its id. Because no id is store-format today, every resume for every organisation walked its whole blocker list to reach the same answer: nothing to build. No allocation and no I/O, but not nothing, and proportional to fan-in on exactly the wide waits this feature exists to serve. An injected predicate now answers that in constant time, ahead of the scan. It has to be injected rather than derived in the engine: the answer is per-organisation, following the snapshot store's own rollout, and the store keeps that state private. It defaults to never, so no resume does any of this work until the ticket that wires the store supplies it. The id-format scan stays behind the predicate, because it is what keeps a mixed cycle working once records are enabled: an organisation mid-rollout holds waitpoints minted either side of the flip, and the format is the only thing that says which half an id belongs to. Threaded through the engine options so it is settable and testable rather than unreachable configuration. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1 parent 5bc039e commit 8140d17

4 files changed

Lines changed: 229 additions & 5 deletions

File tree

‎internal-packages/run-engine/src/engine/index.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -423,6 +423,9 @@ export class RunEngine {
423423
resources,
424424
executionSnapshotSystem: this.executionSnapshotSystem,
425425
enqueueSystem: this.enqueueSystem,
426+
...(options.completedWaitpointRecordsEnabled && {
427+
completedWaitpointRecordsEnabled: options.completedWaitpointRecordsEnabled,
428+
}),
426429
});
427430

428431
this.ttlSystem = new TtlSystem({

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

Lines changed: 29 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,19 @@ export type WaitpointSystemOptions = {
2424
resources: SystemResources;
2525
executionSnapshotSystem: ExecutionSnapshotSystem;
2626
enqueueSystem: EnqueueSystem;
27+
/**
28+
* Whether this run's snapshots can hold a completed-waitpoint record set. Must be O(1).
29+
*
30+
* Sits in FRONT of the id-format scan so a run that cannot hold records pays one predicate
31+
* call, not one `parseWaitpointId` per blocking waitpoint. It has to be injected rather than
32+
* derived here: the answer is per-organisation, since it follows the snapshot store's own
33+
* rollout, and the store keeps that state private to itself.
34+
*
35+
* Defaults to never. A record set is only reachable through the snapshot store, so until the
36+
* ticket that wires that store supplies this predicate there is no run for which one could
37+
* exist, and no resume does any of this work.
38+
*/
39+
completedWaitpointRecordsEnabled?: (runId: string) => boolean;
2740
};
2841

2942
type WaitpointContinuationWaitpoint = Pick<Waitpoint, "id" | "type" | "completedAfter" | "status">;
@@ -47,11 +60,13 @@ export class WaitpointSystem {
4760
private readonly executionSnapshotSystem: ExecutionSnapshotSystem;
4861
private readonly enqueueSystem: EnqueueSystem;
4962
private readonly coordinator: WaitpointCoordinator;
63+
private readonly recordsEnabled: (runId: string) => boolean;
5064

5165
constructor(private readonly options: WaitpointSystemOptions) {
5266
this.$ = options.resources;
5367
this.executionSnapshotSystem = options.executionSnapshotSystem;
5468
this.enqueueSystem = options.enqueueSystem;
69+
this.recordsEnabled = options.completedWaitpointRecordsEnabled ?? (() => false);
5570
this.coordinator = new LegacyPostgresWaitpointCoordinator({
5671
runStore: this.$.runStore,
5772
prisma: this.$.prisma,
@@ -759,7 +774,8 @@ export class WaitpointSystem {
759774
* The record set for one resume, or undefined when no blocking waitpoint carries a store-format
760775
* id.
761776
*
762-
* Gated on id FORMAT, not residency. The two are not the same during a migration: a
777+
* Gated on the record set being reachable at all, and only then on id FORMAT rather than
778+
* residency. Format and residency are not the same during a migration: a
763779
* store-format id can still be served by the Postgres arm, exactly as run-ops ids were for
764780
* runs. Whichever arm owns it answers, so the gate only decides whether to ask at all.
765781
*
@@ -778,10 +794,18 @@ export class WaitpointSystem {
778794
runId: string,
779795
blockingWaitpoints: RunBlockEdge[]
780796
): 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.
797+
// O(1) first, and unconditionally first: a run whose snapshots cannot hold a record set has
798+
// nothing to build, and deciding that by walking its blocking waitpoints made every resume
799+
// for every organisation pay a scan proportional to its fan-in to reach the same answer.
800+
// Defaults to never, so today this returns here for everyone.
801+
if (!this.recordsEnabled(runId)) {
802+
return undefined;
803+
}
804+
805+
// Only then the id-format scan, which is what keeps a MIXED cycle working once records are
806+
// enabled: an organisation mid-rollout holds waitpoints minted either side of the flip, and
807+
// the format is the only thing that says which half each id belongs to. `.some` before the
808+
// dedup so an enabled run holding no store-format id still allocates nothing.
785809
if (!blockingWaitpoints.some((b) => parseWaitpointId(b.waitpoint.id).format === "b32hexW")) {
786810
return undefined;
787811
}
Lines changed: 189 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,189 @@
1+
// The O(1) gate in front of the completed-waitpoint record build.
2+
//
3+
// Deciding whether to build a record set by scanning a run's blocking waitpoints made every
4+
// resume, for every organisation, pay work proportional to its fan-in to reach the same answer:
5+
// no. The predicate moves that decision in front of the scan, and defaults to never.
6+
import { containerTest } from "@internal/testcontainers";
7+
import { trace } from "@internal/tracing";
8+
import { expect } from "vitest";
9+
import { RunEngine } from "../index.js";
10+
import { setTimeout } from "node:timers/promises";
11+
import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js";
12+
13+
// The trigger has to reach the queue, and the completion resumes the run out of band, so both
14+
// need settling before the assertions read state back.
15+
const SETTLE_MS = 500;
16+
17+
async function waitForResume(
18+
engine: RunEngine,
19+
runId: string
20+
): Promise<{ completedWaitpointIds: string[] }> {
21+
for (let attempt = 0; attempt < 20; attempt++) {
22+
const data = await engine.getRunExecutionData({ runId });
23+
const ids = data?.completedWaitpoints?.map((w) => w.id) ?? [];
24+
if (ids.length > 0) {
25+
return { completedWaitpointIds: ids };
26+
}
27+
await setTimeout(100);
28+
}
29+
return { completedWaitpointIds: [] };
30+
}
31+
32+
vi.setConfig({ testTimeout: 60_000 });
33+
34+
function engineWith(
35+
prisma: never,
36+
redisOptions: never,
37+
completedWaitpointRecordsEnabled?: (runId: string) => boolean
38+
) {
39+
return new RunEngine({
40+
prisma,
41+
worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
42+
queue: { redis: redisOptions },
43+
runLock: { redis: redisOptions },
44+
machines: {
45+
defaultMachine: "small-1x",
46+
machines: {
47+
"small-1x": { name: "small-1x" as const, cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 },
48+
},
49+
baseCostInCents: 0.0001,
50+
},
51+
tracer: trace.getTracer("test", "0.0.0"),
52+
...(completedWaitpointRecordsEnabled && { completedWaitpointRecordsEnabled }),
53+
});
54+
}
55+
56+
describe("the completed-waitpoint records gate", () => {
57+
containerTest(
58+
"is consulted once per resume, and the resume is unaffected",
59+
async ({ prisma, redisOptions }) => {
60+
const consulted: string[] = [];
61+
const engine = engineWith(prisma as never, redisOptions as never, (runId) => {
62+
consulted.push(runId);
63+
return false;
64+
});
65+
66+
try {
67+
const env = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
68+
const taskIdentifier = "test-task";
69+
await setupBackgroundWorker(engine, env, taskIdentifier);
70+
71+
const run = await engine.trigger(
72+
{
73+
number: 1,
74+
friendlyId: "run_gate1",
75+
environment: env,
76+
taskIdentifier,
77+
payload: "{}",
78+
payloadType: "application/json",
79+
context: {},
80+
traceContext: {},
81+
traceId: "t12345",
82+
spanId: "s12345",
83+
workerQueue: "main",
84+
queue: `task/${taskIdentifier}`,
85+
isTest: false,
86+
tags: [],
87+
},
88+
prisma
89+
);
90+
91+
await setTimeout(SETTLE_MS);
92+
const dequeued = await engine.dequeueFromWorkerQueue({
93+
consumerId: "test_12345",
94+
workerQueue: "main",
95+
});
96+
const attempt = await engine.startRunAttempt({
97+
runId: dequeued[0]!.run.id,
98+
snapshotId: dequeued[0]!.snapshot.id,
99+
});
100+
101+
// Block on a manual waitpoint, then release it: the release is what resumes the run and
102+
// reaches the gate.
103+
const waitpoint = await engine.createManualWaitpoint({
104+
environmentId: env.id,
105+
projectId: env.projectId,
106+
});
107+
await engine.blockRunWithWaitpoint({
108+
runId: run.id,
109+
waitpoints: [waitpoint.waitpoint.id],
110+
projectId: env.project.id,
111+
organizationId: env.organization.id,
112+
});
113+
114+
await engine.completeWaitpoint({ id: waitpoint.waitpoint.id });
115+
116+
const { completedWaitpointIds } = await waitForResume(engine, run.id);
117+
118+
// The run resumed, because a disabled gate only skips the record build...
119+
expect(completedWaitpointIds).toContain(waitpoint.waitpoint.id);
120+
// ...and the gate was consulted for it, which is what puts it in the resume path.
121+
expect(consulted).toContain(run.id);
122+
expect(attempt.run.id).toBe(run.id);
123+
} finally {
124+
await engine.quit();
125+
}
126+
}
127+
);
128+
129+
containerTest("defaults to never, so no resume consults it", async ({ prisma, redisOptions }) => {
130+
// No predicate supplied, which is every deployment until the store that consumes a record set
131+
// is wired. The default has to be a constant false rather than a scan, or the cost this gate
132+
// exists to remove comes straight back for everyone.
133+
const engine = engineWith(prisma as never, redisOptions as never);
134+
135+
try {
136+
const env = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
137+
const taskIdentifier = "test-task";
138+
await setupBackgroundWorker(engine, env, taskIdentifier);
139+
140+
const run = await engine.trigger(
141+
{
142+
number: 1,
143+
friendlyId: "run_gate2",
144+
environment: env,
145+
taskIdentifier,
146+
payload: "{}",
147+
payloadType: "application/json",
148+
context: {},
149+
traceContext: {},
150+
traceId: "t22345",
151+
spanId: "s22345",
152+
workerQueue: "main",
153+
queue: `task/${taskIdentifier}`,
154+
isTest: false,
155+
tags: [],
156+
},
157+
prisma
158+
);
159+
160+
await setTimeout(SETTLE_MS);
161+
const dequeued = await engine.dequeueFromWorkerQueue({
162+
consumerId: "test_22345",
163+
workerQueue: "main",
164+
});
165+
await engine.startRunAttempt({
166+
runId: dequeued[0]!.run.id,
167+
snapshotId: dequeued[0]!.snapshot.id,
168+
});
169+
170+
const waitpoint = await engine.createManualWaitpoint({
171+
environmentId: env.id,
172+
projectId: env.projectId,
173+
});
174+
await engine.blockRunWithWaitpoint({
175+
runId: run.id,
176+
waitpoints: [waitpoint.waitpoint.id],
177+
projectId: env.project.id,
178+
organizationId: env.organization.id,
179+
});
180+
181+
await engine.completeWaitpoint({ id: waitpoint.waitpoint.id });
182+
183+
const { completedWaitpointIds } = await waitForResume(engine, run.id);
184+
expect(completedWaitpointIds).toContain(waitpoint.waitpoint.id);
185+
} finally {
186+
await engine.quit();
187+
}
188+
});
189+
});

‎internal-packages/run-engine/src/engine/types.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,14 @@ export type RunEngineOptions = {
7777
billing?: {
7878
getCurrentPlan: (orgId: string) => Promise<BillingPlan>;
7979
};
80+
/**
81+
* Whether a run's snapshots can hold a completed-waitpoint record set. Must be O(1): it is
82+
* consulted on every resume, ahead of any work proportional to the run's fan-in.
83+
*
84+
* Per-run rather than global because the answer follows the snapshot store's own
85+
* per-organisation rollout. Omitted means never, so no resume builds a record set.
86+
*/
87+
completedWaitpointRecordsEnabled?: (runId: string) => boolean;
8088
queue: {
8189
redis: RedisOptions;
8290
shardCount?: number;

0 commit comments

Comments
 (0)