Skip to content

Commit e18cc63

Browse files
d-csd-cs
committed
test(run-engine): run triggerAndWait and batchTriggerAndWait on both coordinators
Extends the two-arm parameterization to the RUN and BATCH waitpoint suites, so triggerAndWait and batchTriggerAndWait both exercise the store path end to end rather than only the Postgres one. Same three adjustments as the waitpoint suite: run ids take a shape the anchor derive can work from, waitpoint reads go through the arm-aware helper, and block edge reads union both systems.
1 parent bd7d2d1 commit e18cc63

2 files changed

Lines changed: 104 additions & 105 deletions

File tree

‎internal-packages/run-engine/src/engine/tests/batchTriggerAndWait.test.ts‎

Lines changed: 59 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -4,20 +4,27 @@ import {
44
} from "@internal/testcontainers";
55
import { trace } from "@internal/tracing";
66
import { expect, describe } from "vitest";
7-
import { createTestEngine } from "./helpers/engineFactory.js";
7+
import {
8+
createTestEngine,
9+
freshRunFriendlyId,
10+
readRunBlockEdgesForArm,
11+
readWaitpointForArm,
12+
type WaitpointArm,
13+
} from "./helpers/engineFactory.js";
814
import { setTimeout } from "node:timers/promises";
915
import { generateFriendlyId, BatchId } from "@trigger.dev/core/v3/isomorphic";
1016
import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js";
1117
import type { CompleteBatchResult, BatchItem } from "../../batch-queue/types.js";
1218

1319
vi.setConfig({ testTimeout: 60_000 });
1420

15-
describe("RunEngine batchTriggerAndWait", () => {
21+
describe.each<WaitpointArm>(["legacy", "store"])("RunEngine batchTriggerAndWait (%s)", (arm) => {
1622
containerTest("batchTriggerAndWait (no idempotency)", async ({ prisma, redisOptions }) => {
1723
//create environment
1824
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
1925

2026
const engine = createTestEngine({
27+
waitpointArm: arm,
2128
prisma,
2229
worker: {
2330
redis: redisOptions,
@@ -67,7 +74,7 @@ describe("RunEngine batchTriggerAndWait", () => {
6774
const parentRun = await engine.trigger(
6875
{
6976
number: 1,
70-
friendlyId: "run_p1234",
77+
friendlyId: freshRunFriendlyId(arm),
7178
environment: authenticatedEnvironment,
7279
taskIdentifier: parentTask,
7380
payload: "{}",
@@ -115,7 +122,7 @@ describe("RunEngine batchTriggerAndWait", () => {
115122
const child1 = await engine.trigger(
116123
{
117124
number: 1,
118-
friendlyId: "run_c1234",
125+
friendlyId: freshRunFriendlyId(arm),
119126
environment: authenticatedEnvironment,
120127
taskIdentifier: childTask,
121128
payload: "{}",
@@ -142,7 +149,7 @@ describe("RunEngine batchTriggerAndWait", () => {
142149
const child2 = await engine.trigger(
143150
{
144151
number: 2,
145-
friendlyId: "run_c12345",
152+
friendlyId: freshRunFriendlyId(arm),
146153
environment: authenticatedEnvironment,
147154
taskIdentifier: childTask,
148155
payload: "{}",
@@ -167,16 +174,11 @@ describe("RunEngine batchTriggerAndWait", () => {
167174
expect(parentAfterChild2.snapshot.executionStatus).toBe("EXECUTING_WITH_WAITPOINTS");
168175

169176
//check the waitpoint blocking the parent run
170-
const runWaitpoints = await prisma.taskRunWaitpoint.findMany({
171-
where: {
172-
taskRunId: parentRun.id,
173-
},
174-
include: {
175-
waitpoint: true,
176-
},
177-
orderBy: {
178-
createdAt: "asc",
179-
},
177+
const runWaitpoints = await readRunBlockEdgesForArm({
178+
arm,
179+
prisma,
180+
redisOptions,
181+
runId: parentRun.id,
180182
});
181183
expect(runWaitpoints.length).toBe(3);
182184
const child1Waitpoint = runWaitpoints.find(
@@ -230,24 +232,23 @@ describe("RunEngine batchTriggerAndWait", () => {
230232
assertNonNullable(childExecutionDataAfter);
231233
expect(childExecutionDataAfter.snapshot.executionStatus).toBe("FINISHED");
232234

233-
const child1WaitpointAfter = await prisma.waitpoint.findFirst({
234-
where: {
235-
id: child1Waitpoint?.waitpointId,
236-
},
235+
const child1WaitpointAfter = await readWaitpointForArm({
236+
arm,
237+
prisma,
238+
redisOptions,
239+
waitpointId: child1Waitpoint!.waitpointId,
237240
});
238241
expect(child1WaitpointAfter?.completedAt).not.toBeNull();
239242
expect(child1WaitpointAfter?.status).toBe("COMPLETED");
240243
expect(child1WaitpointAfter?.output).toBe('{"foo":"bar"}');
241244

242245
await setTimeout(500);
243246

244-
const runWaitpointsAfterFirstChild = await prisma.taskRunWaitpoint.findMany({
245-
where: {
246-
taskRunId: parentRun.id,
247-
},
248-
include: {
249-
waitpoint: true,
250-
},
247+
const runWaitpointsAfterFirstChild = await readRunBlockEdgesForArm({
248+
arm,
249+
prisma,
250+
redisOptions,
251+
runId: parentRun.id,
251252
});
252253
expect(runWaitpointsAfterFirstChild.length).toBe(3);
253254

@@ -291,24 +292,23 @@ describe("RunEngine batchTriggerAndWait", () => {
291292
assertNonNullable(child2ExecutionDataAfter);
292293
expect(child2ExecutionDataAfter.snapshot.executionStatus).toBe("FINISHED");
293294

294-
const child2WaitpointAfter = await prisma.waitpoint.findFirst({
295-
where: {
296-
id: child2Waitpoint?.waitpointId,
297-
},
295+
const child2WaitpointAfter = await readWaitpointForArm({
296+
arm,
297+
prisma,
298+
redisOptions,
299+
waitpointId: child2Waitpoint!.waitpointId,
298300
});
299301
expect(child2WaitpointAfter?.completedAt).not.toBeNull();
300302
expect(child2WaitpointAfter?.status).toBe("COMPLETED");
301303
expect(child2WaitpointAfter?.output).toBe('{"baz":"qux"}');
302304

303305
await setTimeout(1_000);
304306

305-
const runWaitpointsAfterSecondChild = await prisma.taskRunWaitpoint.findMany({
306-
where: {
307-
taskRunId: parentRun.id,
308-
},
309-
include: {
310-
waitpoint: true,
311-
},
307+
const runWaitpointsAfterSecondChild = await readRunBlockEdgesForArm({
308+
arm,
309+
prisma,
310+
redisOptions,
311+
runId: parentRun.id,
312312
});
313313
expect(runWaitpointsAfterSecondChild.length).toBe(0);
314314

@@ -367,6 +367,7 @@ describe("RunEngine batchTriggerAndWait", () => {
367367
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
368368

369369
const engine = createTestEngine({
370+
waitpointArm: arm,
370371
prisma,
371372
worker: {
372373
redis: redisOptions,
@@ -421,7 +422,7 @@ describe("RunEngine batchTriggerAndWait", () => {
421422
const parentRun = await engine.trigger(
422423
{
423424
number: 1,
424-
friendlyId: "run_p1234",
425+
friendlyId: freshRunFriendlyId(arm),
425426
environment: authenticatedEnvironment,
426427
taskIdentifier: parentTask,
427428
payload: "{}",
@@ -471,7 +472,7 @@ describe("RunEngine batchTriggerAndWait", () => {
471472
const batchChild = await engine.trigger(
472473
{
473474
number: 1,
474-
friendlyId: "run_c1234",
475+
friendlyId: freshRunFriendlyId(arm),
475476
environment: authenticatedEnvironment,
476477
taskIdentifier: batchChildTask,
477478
payload: "{}",
@@ -524,13 +525,11 @@ describe("RunEngine batchTriggerAndWait", () => {
524525

525526
await setTimeout(500);
526527

527-
const runWaitpointsAfterBatchChild = await prisma.taskRunWaitpoint.findMany({
528-
where: {
529-
taskRunId: parentRun.id,
530-
},
531-
include: {
532-
waitpoint: true,
533-
},
528+
const runWaitpointsAfterBatchChild = await readRunBlockEdgesForArm({
529+
arm,
530+
prisma,
531+
redisOptions,
532+
runId: parentRun.id,
534533
});
535534
expect(runWaitpointsAfterBatchChild.length).toBe(0);
536535

@@ -549,7 +548,7 @@ describe("RunEngine batchTriggerAndWait", () => {
549548
const _triggerAndWaitChildRun = await engine.trigger(
550549
{
551550
number: 1,
552-
friendlyId: "run_c123456",
551+
friendlyId: freshRunFriendlyId(arm),
553552
environment: authenticatedEnvironment,
554553
taskIdentifier: triggerAndWaitChildTask,
555554
payload: "{}",
@@ -588,6 +587,7 @@ describe("RunEngine batchTriggerAndWait", () => {
588587
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
589588

590589
const engine = createTestEngine({
590+
waitpointArm: arm,
591591
prisma,
592592
worker: {
593593
redis: redisOptions,
@@ -713,7 +713,7 @@ describe("RunEngine batchTriggerAndWait", () => {
713713
const parentRun = await engine.trigger(
714714
{
715715
number: 1,
716-
friendlyId: "run_parent",
716+
friendlyId: freshRunFriendlyId(arm),
717717
environment: authenticatedEnvironment,
718718
taskIdentifier: parentTask,
719719
payload: "{}",
@@ -847,8 +847,11 @@ describe("RunEngine batchTriggerAndWait", () => {
847847
// Wait for parent to be unblocked (use waitFor since tryCompleteBatch runs as background job)
848848
await vi.waitFor(
849849
async () => {
850-
const waitpoints = await prisma.taskRunWaitpoint.findMany({
851-
where: { taskRunId: parentRun.id },
850+
const waitpoints = await readRunBlockEdgesForArm({
851+
arm,
852+
prisma,
853+
redisOptions,
854+
runId: parentRun.id,
852855
});
853856
expect(waitpoints.length).toBe(0);
854857
},
@@ -884,6 +887,7 @@ describe("RunEngine batchTriggerAndWait", () => {
884887
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
885888

886889
const engine = createTestEngine({
890+
waitpointArm: arm,
887891
prisma,
888892
worker: {
889893
redis: redisOptions,
@@ -1190,8 +1194,11 @@ describe("RunEngine batchTriggerAndWait", () => {
11901194
// Wait for parent to be unblocked (use waitFor since tryCompleteBatch runs as background job)
11911195
await vi.waitFor(
11921196
async () => {
1193-
const waitpoints = await prisma.taskRunWaitpoint.findMany({
1194-
where: { taskRunId: parentRun.id },
1197+
const waitpoints = await readRunBlockEdgesForArm({
1198+
arm,
1199+
prisma,
1200+
redisOptions,
1201+
runId: parentRun.id,
11951202
});
11961203
expect(waitpoints.length).toBe(0);
11971204
},

0 commit comments

Comments
 (0)