Skip to content

Commit 8e28ab4

Browse files
d-csTrigger.dev RepoOps
authored andcommitted
feat(run-engine): resolve completed waitpoint records
Mono-RevId: 54db6e11f2826446a1cafc90c782df783f471ce5
1 parent dc5b188 commit 8e28ab4

22 files changed

Lines changed: 1534 additions & 24 deletions

‎apps/webapp/app/v3/featureFlags.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ import { z } from "zod";
22
import { DeployBuildPath } from "@trigger.dev/core/v3";
33

44
export const FEATURE_FLAG = {
5+
waitpointStoreEnabled: "waitpointStoreEnabled",
56
snapshotStoreMode: "snapshotStoreMode",
67
allowEmptyEnvironmentVariableValues: "allowEmptyEnvironmentVariableValues",
78
defaultWorkerInstanceGroupId: "defaultWorkerInstanceGroupId",
@@ -92,6 +93,7 @@ const WorkerQueueEnvironmentOverrideKey = z
9293
);
9394

9495
export const FeatureFlagCatalog = {
96+
[FEATURE_FLAG.waitpointStoreEnabled]: z.boolean(),
9597
// Organization override, then global default, then off.
9698
[FEATURE_FLAG.snapshotStoreMode]: z.enum(["off", "dual-write", "redis-read", "redis-only"]),
9799
[FEATURE_FLAG.allowEmptyEnvironmentVariableValues]: z.boolean(),

‎apps/webapp/app/v3/runEngine.server.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
import { createWaitpointStoreEnabledResolver } from "./utils/waitpointStoreEnabled.server";
2+
import { globalFlagsRegistry } from "./globalFlagsRegistry.server";
13
import { RunEngine } from "@internal/run-engine";
24
import { $replica, prisma } from "~/db.server";
35
import { env } from "~/env.server";
@@ -20,6 +22,10 @@ export type { RunEngine };
2022

2123
function createRunEngine() {
2224
const engine = new RunEngine({
25+
waitpointStoreEnabled: createWaitpointStoreEnabledResolver(
26+
() => globalFlagsRegistry.current(),
27+
prisma
28+
),
2329
prisma,
2430
readOnlyPrisma: $replica,
2531
crossSeamGuard: pickRunOpsStoreForCompletion,
Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
import type { PrismaClient } from "@trigger.dev/database";
2+
import { FEATURE_FLAG, FeatureFlagCatalog } from "../featureFlags";
3+
4+
/** The normal organization override, then global default, then off. No rollout history. */
5+
export function createWaitpointStoreEnabledResolver(
6+
readFlags: () => { waitpointStoreEnabled?: boolean } | undefined,
7+
prisma: Pick<PrismaClient, "organization">
8+
) {
9+
return async (organizationId: string, organizationFlags?: unknown): Promise<boolean> => {
10+
const overrides =
11+
organizationFlags === undefined
12+
? (
13+
await prisma.organization.findFirst({
14+
where: { id: organizationId },
15+
select: { featureFlags: true },
16+
})
17+
)?.featureFlags
18+
: organizationFlags;
19+
const value =
20+
typeof overrides === "object" && overrides !== null && "waitpointStoreEnabled" in overrides
21+
? overrides.waitpointStoreEnabled
22+
: undefined;
23+
const parsed = FeatureFlagCatalog[FEATURE_FLAG.waitpointStoreEnabled].safeParse(value);
24+
return parsed.success ? parsed.data : (readFlags()?.waitpointStoreEnabled ?? false);
25+
};
26+
}

‎apps/webapp/test/api-auth.e2e.test.ts‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -538,3 +538,37 @@ describe("API bearer auth — environment/project edge cases", () => {
538538
expect(res.status).toBe(401);
539539
});
540540
});
541+
542+
describe("admin waitpoint rollout controls", () => {
543+
it("saves the normal org/global flag without writing derived global state", async () => {
544+
const user = await seedTestUser(server.prisma, { admin: true });
545+
const pat = await seedTestPAT(server.prisma, user.id);
546+
const organization = await server.prisma.organization.create({
547+
data: { title: "Waitpoint rollout", slug: `waitpoint-rollout-${user.id}` },
548+
});
549+
const post = async (path: string, body: unknown) => {
550+
const response = await server.webapp.fetch(path, {
551+
method: "POST",
552+
headers: { Authorization: `Bearer ${pat.token}`, "Content-Type": "application/json" },
553+
body: JSON.stringify(body),
554+
});
555+
expect(response.status, await response.text()).toBe(200);
556+
};
557+
const globalPath = "/admin/api/v1/feature-flags";
558+
try {
559+
await post(`/admin/api/v1/orgs/${organization.id}/feature-flags`, {
560+
waitpointStoreEnabled: true,
561+
});
562+
await post(globalPath, { waitpointStoreEnabled: false });
563+
expect(
564+
(await server.prisma.organization.findFirstOrThrow({ where: { id: organization.id } }))
565+
.featureFlags
566+
).toMatchObject({ waitpointStoreEnabled: true });
567+
expect(
568+
await server.prisma.featureFlag.findFirst({ where: { key: "waitpointRolloutState" } })
569+
).toBeNull();
570+
} finally {
571+
await post(globalPath, { waitpointStoreEnabled: false });
572+
}
573+
});
574+
});
Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
import { describe, expect } from "vitest";
2+
import { postgresTest } from "@internal/testcontainers";
3+
import { createWaitpointStoreEnabledResolver } from "~/v3/utils/waitpointStoreEnabled.server";
4+
5+
describe("waitpoint rollout", () => {
6+
postgresTest(
7+
"waitpoint enrollment uses ordinary org flags, global fallback, and immediate rollback",
8+
async ({ prisma }) => {
9+
const org = await prisma.organization.create({
10+
data: { title: "Waitpoints", slug: "waitpoints" },
11+
});
12+
let global: { waitpointStoreEnabled?: boolean } | undefined;
13+
const enabled = createWaitpointStoreEnabledResolver(() => global, prisma);
14+
expect(await enabled(org.id)).toBe(false);
15+
await prisma.organization.update({
16+
where: { id: org.id },
17+
data: { featureFlags: { waitpointStoreEnabled: true } },
18+
});
19+
expect(await enabled(org.id)).toBe(true);
20+
global = { waitpointStoreEnabled: false };
21+
expect(await enabled(org.id)).toBe(true);
22+
await prisma.organization.update({
23+
where: { id: org.id },
24+
data: { featureFlags: { waitpointStoreEnabled: false } },
25+
});
26+
global = { waitpointStoreEnabled: true };
27+
expect(await enabled(org.id)).toBe(false);
28+
expect(await enabled(org.id, {})).toBe(true);
29+
expect(await enabled(org.id, { waitpointStoreEnabled: false })).toBe(false);
30+
global = { waitpointStoreEnabled: false };
31+
expect(await enabled(org.id)).toBe(false);
32+
expect(await prisma.featureFlag.count()).toBe(0);
33+
}
34+
);
35+
});

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -466,6 +466,7 @@ export class RunEngine {
466466
});
467467

468468
this.waitpointSystem = new WaitpointSystem({
469+
waitpointStoreEnabled: options.waitpointStoreEnabled,
469470
resources,
470471
executionSnapshotSystem: this.executionSnapshotSystem,
471472
enqueueSystem: this.enqueueSystem,
Lines changed: 209 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,209 @@
1+
// Pure mapping coverage; storage routing and failures use real stores in the integration suite.
2+
import { describe, expect, it } from "vitest";
3+
import { generateWaitpointId } from "@trigger.dev/core/v3/isomorphic";
4+
import type {
5+
WaitpointCompletion,
6+
WaitpointRecordInput,
7+
WaitpointStoredRecord,
8+
} from "../waitpointCoordinator/storeCoordinator.js";
9+
import { storeRecordToCompletedWaitpointRecord } from "./completedWaitpointRecordAdapter.js";
10+
11+
function storeRecordInput(
12+
id: string,
13+
overrides: Partial<WaitpointRecordInput> = {}
14+
): WaitpointRecordInput {
15+
return {
16+
id,
17+
friendlyId: `waitpoint_${id}`,
18+
type: "MANUAL",
19+
environmentId: "env_1",
20+
projectId: "proj_1",
21+
createdAt: "2026-01-01T00:00:00.000Z",
22+
updatedAt: "2026-01-01T00:00:00.000Z",
23+
userProvidedIdempotencyKey: false,
24+
tags: [],
25+
...overrides,
26+
};
27+
}
28+
29+
function storeCompletion(overrides: Partial<WaitpointCompletion> = {}): WaitpointCompletion {
30+
return {
31+
completedAt: "2026-01-01T00:00:00.000Z",
32+
outputType: "application/json",
33+
outputIsError: false,
34+
output: null,
35+
...overrides,
36+
};
37+
}
38+
39+
function storeEntry(
40+
id: string,
41+
recordOverrides: Partial<WaitpointRecordInput> = {},
42+
completion: WaitpointCompletion | undefined = storeCompletion(),
43+
status: WaitpointStoredRecord["status"] = "COMPLETED"
44+
): WaitpointStoredRecord {
45+
return { status, record: storeRecordInput(id, recordOverrides), completion };
46+
}
47+
48+
// A fixed effective time for the mapping tests that do not exercise expiry.
49+
const NOW_MS = Date.parse("2026-08-21T12:00:00.000Z");
50+
const mapStoreRecord = (entry: WaitpointStoredRecord, nowMs: number = NOW_MS) =>
51+
storeRecordToCompletedWaitpointRecord(entry, nowMs);
52+
53+
describe("storeRecordToCompletedWaitpointRecord — envelope fidelity", () => {
54+
it("preserves an inline output", () => {
55+
const id = generateWaitpointId("MANUAL");
56+
const record = mapStoreRecord(
57+
storeEntry(id, {}, storeCompletion({ output: { inline: '{"value":42}' } }))
58+
);
59+
expect(record.output).toEqual({ inline: '{"value":42}' });
60+
});
61+
62+
it("preserves a referenced (offloaded) output", () => {
63+
const id = generateWaitpointId("MANUAL");
64+
const record = mapStoreRecord(
65+
storeEntry(
66+
id,
67+
{},
68+
storeCompletion({ outputType: "application/store", output: { ref: "s3://bucket/key" } })
69+
)
70+
);
71+
expect(record.output).toEqual({ ref: "s3://bucket/key" });
72+
expect(record.outputType).toBe("application/store");
73+
});
74+
75+
it("preserves an error output and its flag", () => {
76+
const id = generateWaitpointId("MANUAL");
77+
const record = mapStoreRecord(
78+
storeEntry(
79+
id,
80+
{},
81+
storeCompletion({ outputIsError: true, output: { inline: '{"type":"STRING_ERROR"}' } })
82+
)
83+
);
84+
expect(record.outputIsError).toBe(true);
85+
expect(record.output).toEqual({ inline: '{"type":"STRING_ERROR"}' });
86+
});
87+
88+
it("keeps a RUN success on deriveFromRun, not a copy of its output", () => {
89+
const id = generateWaitpointId("RUN");
90+
const record = mapStoreRecord(
91+
storeEntry(
92+
id,
93+
{ type: "RUN", completedByTaskRunId: "run_child" },
94+
storeCompletion({ output: { inline: '{"value":42}' } })
95+
)
96+
);
97+
// The stored inline value is dropped: the read side re-derives it from TaskRun.output.
98+
expect(record.output).toEqual({ deriveFromRun: true });
99+
expect(record.completedByTaskRunId).toBe("run_child");
100+
});
101+
102+
it("keeps a RUN success on deriveFromRun even when the stored output is null", () => {
103+
// The store may keep a re-derivable RUN success's output null rather than copy it. That null must
104+
// still resolve to deriveFromRun, NOT to no output — the completion is re-read from TaskRun.output.
105+
const id = generateWaitpointId("RUN");
106+
const record = mapStoreRecord(
107+
storeEntry(
108+
id,
109+
{ type: "RUN", completedByTaskRunId: "run_child" },
110+
storeCompletion({ output: null })
111+
)
112+
);
113+
expect(record.output).toEqual({ deriveFromRun: true });
114+
expect(record.completedByTaskRunId).toBe("run_child");
115+
});
116+
117+
it("carries a RUN ERROR inline, never deriveFromRun", () => {
118+
const id = generateWaitpointId("RUN");
119+
const record = mapStoreRecord(
120+
storeEntry(
121+
id,
122+
{ type: "RUN", completedByTaskRunId: "run_child" },
123+
storeCompletion({ outputIsError: true, output: { inline: '{"type":"BUILT_IN_ERROR"}' } })
124+
)
125+
);
126+
expect(record.output).toEqual({ inline: '{"type":"BUILT_IN_ERROR"}' });
127+
});
128+
129+
it("preserves timestamps and type-specific metadata", () => {
130+
const id = generateWaitpointId("DATETIME");
131+
const record = mapStoreRecord(
132+
storeEntry(
133+
id,
134+
{ type: "DATETIME", completedAfter: "2026-02-02T00:00:00.000Z" },
135+
storeCompletion({ completedAt: "2026-03-03T04:05:06.000Z" })
136+
)
137+
);
138+
expect(record.type).toBe("DATETIME");
139+
expect(record.completedAt).toBe("2026-03-03T04:05:06.000Z");
140+
expect(record.completedAfter).toBe("2026-02-02T00:00:00.000Z");
141+
});
142+
143+
it("exposes a user-provided idempotency key and hides a system one", () => {
144+
const id = generateWaitpointId("MANUAL");
145+
const withUserKey = mapStoreRecord(
146+
storeEntry(id, { userProvidedIdempotencyKey: true, idempotencyKey: "idem_user" })
147+
);
148+
expect(withUserKey.idempotencyKey).toBe("idem_user");
149+
150+
const withSystemKey = mapStoreRecord(
151+
storeEntry(id, { userProvidedIdempotencyKey: false, idempotencyKey: "idem_system" })
152+
);
153+
expect(withSystemKey.idempotencyKey).toBeUndefined();
154+
});
155+
});
156+
157+
describe("storeRecordToCompletedWaitpointRecord — idempotency key expiry", () => {
158+
const iso = (ms: number) => new Date(ms).toISOString();
159+
const userKey = (overrides: Partial<WaitpointRecordInput>) =>
160+
storeEntry(generateWaitpointId("MANUAL"), {
161+
userProvidedIdempotencyKey: true,
162+
idempotencyKey: "idem_user",
163+
...overrides,
164+
});
165+
166+
it("keeps a user key with no expiry visible", () => {
167+
expect(mapStoreRecord(userKey({}), NOW_MS).idempotencyKey).toBe("idem_user");
168+
});
169+
170+
it("keeps a user key with a future expiry visible", () => {
171+
const record = mapStoreRecord(
172+
userKey({ idempotencyKeyExpiresAt: iso(NOW_MS + 60_000) }),
173+
NOW_MS
174+
);
175+
expect(record.idempotencyKey).toBe("idem_user");
176+
});
177+
178+
it("omits a user key with a past expiry", () => {
179+
const record = mapStoreRecord(
180+
userKey({ idempotencyKeyExpiresAt: iso(NOW_MS - 60_000) }),
181+
NOW_MS
182+
);
183+
expect(record.idempotencyKey).toBeUndefined();
184+
});
185+
186+
it("omits a user key expiring exactly at the effective time (PEXPIREAT semantics)", () => {
187+
const record = mapStoreRecord(userKey({ idempotencyKeyExpiresAt: iso(NOW_MS) }), NOW_MS);
188+
expect(record.idempotencyKey).toBeUndefined();
189+
});
190+
191+
it("fails loud on a malformed stored expiry", () => {
192+
expect(() =>
193+
mapStoreRecord(userKey({ idempotencyKeyExpiresAt: "not-a-date" }), NOW_MS)
194+
).toThrow(/malformed idempotencyKeyExpiresAt/);
195+
});
196+
197+
it("keeps a system-generated key hidden regardless of expiry (never throws on its expiry)", () => {
198+
const system = (expiry: string | undefined) =>
199+
storeEntry(generateWaitpointId("MANUAL"), {
200+
userProvidedIdempotencyKey: false,
201+
idempotencyKey: "idem_system",
202+
idempotencyKeyExpiresAt: expiry,
203+
});
204+
expect(mapStoreRecord(system(iso(NOW_MS - 60_000)), NOW_MS).idempotencyKey).toBeUndefined();
205+
expect(mapStoreRecord(system(iso(NOW_MS + 60_000)), NOW_MS).idempotencyKey).toBeUndefined();
206+
// A malformed expiry on a hidden key is never consulted, so it does not fail loud.
207+
expect(mapStoreRecord(system("not-a-date"), NOW_MS).idempotencyKey).toBeUndefined();
208+
});
209+
});

0 commit comments

Comments
 (0)