Skip to content

Commit 710175b

Browse files
ericallamericallam
committed
perf(run-engine,run-store): one execution snapshot per triggered run
A non-delayed run was getting RUN_CREATED nested in the run-create transaction followed by a separate QUEUED write. It now gets a single QUEUED snapshot inside the create, with the trigger path only publishing to the queue.
1 parent 8d321f8 commit 710175b

7 files changed

Lines changed: 262 additions & 42 deletions

File tree

‎.server-changes/collapse-trigger-queued-snapshot.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
---
2+
area: webapp
3+
type: improvement
4+
---
5+
6+
Triggering a task now does one less database write, so runs reach the queue marginally faster.

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

Lines changed: 26 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import {
2323
} from "@trigger.dev/core/v3";
2424
import type { TaskRunError } from "@trigger.dev/core/v3/schemas";
2525
import {
26+
generateInternalId,
2627
parseNaturalLanguageDurationInMs,
2728
RunId,
2829
WaitpointId,
@@ -946,6 +947,7 @@ export class RunEngine {
946947

947948
let taskRun: TaskRun & { associatedWaitpoint: Waitpoint | null };
948949
const taskRunId = RunId.fromFriendlyId(friendlyId);
950+
const initialSnapshotId = generateInternalId();
949951

950952
// App-level replacement for the dropped TaskRun env/project Cascade FKs.
951953
await this.controlPlaneResolver.assertEnvExists(environment.id);
@@ -1031,9 +1033,10 @@ export class RunEngine {
10311033
annotations,
10321034
},
10331035
snapshot: {
1036+
id: initialSnapshotId,
10341037
engine: "V2",
1035-
executionStatus: delayUntil ? "DELAYED" : "RUN_CREATED",
1036-
description: delayUntil ? "Run is delayed" : "Run was created",
1038+
executionStatus: delayUntil ? "DELAYED" : "QUEUED",
1039+
description: delayUntil ? "Run is delayed" : "Run was QUEUED",
10371040
runStatus: status,
10381041
environmentId: environment.id,
10391042
environmentType: environment.type,
@@ -1160,13 +1163,29 @@ export class RunEngine {
11601163
await this.ttlSystem.scheduleExpireRun({ runId: taskRun.id, ttl: taskRun.ttl });
11611164
}
11621165

1163-
await this.enqueueSystem.enqueueRun({
1166+
this.eventBus.emit("executionSnapshotCreated", {
1167+
time: taskRun.createdAt,
1168+
run: {
1169+
id: taskRun.id,
1170+
},
1171+
snapshot: {
1172+
id: initialSnapshotId,
1173+
executionStatus: "QUEUED",
1174+
description: "Run was QUEUED",
1175+
runStatus: taskRun.status,
1176+
attemptNumber: taskRun.attemptNumber ?? null,
1177+
checkpointId: null,
1178+
workerId: workerId ?? null,
1179+
runnerId: runnerId ?? null,
1180+
isValid: true,
1181+
error: null,
1182+
completedWaitpointIds: [],
1183+
},
1184+
});
1185+
1186+
await this.enqueueSystem.publishRun({
11641187
run: taskRun,
11651188
env: environment,
1166-
workerId,
1167-
runnerId,
1168-
tx: prisma,
1169-
skipRunLock: true,
11701189
includeTtl: true,
11711190
enableFastPath,
11721191
});

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

Lines changed: 52 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -95,41 +95,62 @@ export class EnqueueSystem {
9595
store
9696
);
9797

98-
// Force development runs to use the environment id as the worker queue.
99-
const workerQueue = env.type === "DEVELOPMENT" ? env.id : run.workerQueue;
98+
await this.publishRun({ run, env, includeTtl, enableFastPath });
10099

101-
const timestamp = (run.queueTimestamp ?? run.createdAt).getTime() - run.priorityMs;
100+
return newSnapshot;
101+
});
102+
}
102103

103-
// Include TTL only when explicitly requested (first enqueue from trigger).
104-
// Re-enqueues (waitpoint, checkpoint, delayed, pending version) must not add TTL.
105-
let ttlExpiresAt: number | undefined;
106-
if (includeTtl && run.ttl) {
107-
const expireAt = parseNaturalLanguageDuration(run.ttl);
108-
if (expireAt) {
109-
ttlExpiresAt = expireAt.getTime();
110-
}
111-
}
104+
/**
105+
* Publishes the run to the RunQueue without writing an execution snapshot. Callers that already
106+
* hold a `QUEUED` snapshot (the trigger path writes one inside the run-create transaction) use
107+
* this so the run does not pay for a second snapshot write.
108+
*/
109+
public async publishRun({
110+
run,
111+
env,
112+
includeTtl = false,
113+
enableFastPath = false,
114+
}: {
115+
run: TaskRun;
116+
env: MinimalAuthenticatedEnvironment;
117+
/** When true, include TTL in the queued message (only for first enqueue from trigger). Default false. */
118+
includeTtl?: boolean;
119+
/** When true, allow the queue to push directly to worker queue if concurrency is available. */
120+
enableFastPath?: boolean;
121+
}) {
122+
// Force development runs to use the environment id as the worker queue.
123+
const workerQueue = env.type === "DEVELOPMENT" ? env.id : run.workerQueue;
112124

113-
await this.$.runQueue.enqueueMessage({
114-
env,
115-
workerQueue,
116-
enableFastPath,
117-
message: {
118-
runId: run.id,
119-
taskIdentifier: run.taskIdentifier,
120-
orgId: env.organization.id,
121-
projectId: env.project.id,
122-
environmentId: env.id,
123-
environmentType: env.type,
124-
queue: run.queue,
125-
concurrencyKey: run.concurrencyKey ?? undefined,
126-
timestamp,
127-
attempt: 0,
128-
ttlExpiresAt,
129-
},
130-
});
125+
const timestamp = (run.queueTimestamp ?? run.createdAt).getTime() - run.priorityMs;
131126

132-
return newSnapshot;
127+
// Include TTL only when explicitly requested (first enqueue from trigger).
128+
// Re-enqueues (waitpoint, checkpoint, delayed, pending version) must not add TTL.
129+
let ttlExpiresAt: number | undefined;
130+
if (includeTtl && run.ttl) {
131+
const expireAt = parseNaturalLanguageDuration(run.ttl);
132+
if (expireAt) {
133+
ttlExpiresAt = expireAt.getTime();
134+
}
135+
}
136+
137+
await this.$.runQueue.enqueueMessage({
138+
env,
139+
workerQueue,
140+
enableFastPath,
141+
message: {
142+
runId: run.id,
143+
taskIdentifier: run.taskIdentifier,
144+
orgId: env.organization.id,
145+
projectId: env.project.id,
146+
environmentId: env.id,
147+
environmentType: env.type,
148+
queue: run.queue,
149+
concurrencyKey: run.concurrencyKey ?? undefined,
150+
timestamp,
151+
attempt: 0,
152+
ttlExpiresAt,
153+
},
133154
});
134155
}
135156
}

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

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -137,9 +137,6 @@ const cancelledSnapshot = (friendlyId: string, environment: any) => ({
137137
});
138138

139139
describe("RunEngine trigger/create routing", () => {
140-
// trigger create routes through runStore.createRun with the structured
141-
// DTO, and the persisted run + its nested first RUN_CREATED snapshot land via
142-
// the single create call.
143140
containerTest(
144141
"trigger routes createRun and lands run + first snapshot",
145142
async ({ prisma, redisOptions }) => {
@@ -169,7 +166,7 @@ describe("RunEngine trigger/create routing", () => {
169166
orderBy: { createdAt: "asc" },
170167
});
171168
expect(snapshot).not.toBeNull();
172-
expect(snapshot!.executionStatus).toBe("RUN_CREATED");
169+
expect(snapshot!.executionStatus).toBe("QUEUED");
173170
} finally {
174171
await engine.quit();
175172
}

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

Lines changed: 175 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,175 @@
1+
import { assertNonNullable, containerTest } from "@internal/testcontainers";
2+
import { trace } from "@internal/tracing";
3+
import type { PrismaClient, TaskRunExecutionStatus } from "@trigger.dev/database";
4+
import { setTimeout } from "node:timers/promises";
5+
import { expect } from "vitest";
6+
import { RunEngine } from "../index.js";
7+
import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js";
8+
9+
vi.setConfig({ testTimeout: 60_000 });
10+
11+
async function snapshotStatuses(
12+
prisma: PrismaClient,
13+
runId: string
14+
): Promise<TaskRunExecutionStatus[]> {
15+
const snapshots = await prisma.taskRunExecutionSnapshot.findMany({
16+
where: { runId },
17+
orderBy: { createdAt: "asc" },
18+
select: { executionStatus: true },
19+
});
20+
21+
return snapshots.map((snapshot) => snapshot.executionStatus);
22+
}
23+
24+
describe("RunEngine trigger() execution snapshots", () => {
25+
containerTest(
26+
"a non-delayed run is created with a single QUEUED snapshot",
27+
async ({ prisma, redisOptions }) => {
28+
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
29+
30+
const engine = new RunEngine({
31+
prisma,
32+
worker: {
33+
redis: redisOptions,
34+
workers: 1,
35+
tasksPerWorker: 10,
36+
pollIntervalMs: 100,
37+
},
38+
queue: {
39+
redis: redisOptions,
40+
masterQueueConsumersDisabled: true,
41+
processWorkerQueueDebounceMs: 50,
42+
},
43+
runLock: {
44+
redis: redisOptions,
45+
},
46+
machines: {
47+
defaultMachine: "small-1x",
48+
machines: {
49+
"small-1x": {
50+
name: "small-1x" as const,
51+
cpu: 0.5,
52+
memory: 0.5,
53+
centsPerMs: 0.0001,
54+
},
55+
},
56+
baseCostInCents: 0.0001,
57+
},
58+
tracer: trace.getTracer("test", "0.0.0"),
59+
});
60+
61+
try {
62+
const taskIdentifier = "test-task";
63+
64+
await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
65+
66+
const run = await engine.trigger(
67+
{
68+
number: 1,
69+
friendlyId: "run_1234",
70+
environment: authenticatedEnvironment,
71+
taskIdentifier,
72+
payload: "{}",
73+
payloadType: "application/json",
74+
context: {},
75+
traceContext: {},
76+
traceId: "t_collapse_1",
77+
spanId: "s_collapse_1",
78+
workerQueue: "main",
79+
queue: "task/test-task",
80+
isTest: false,
81+
tags: [],
82+
},
83+
prisma
84+
);
85+
86+
expect(await snapshotStatuses(prisma, run.id)).toEqual(["QUEUED"]);
87+
88+
const queueLength = await engine.runQueue.lengthOfQueue(
89+
authenticatedEnvironment,
90+
run.queue
91+
);
92+
expect(queueLength).toBe(1);
93+
94+
const executionData = await engine.getRunExecutionData({ runId: run.id });
95+
assertNonNullable(executionData);
96+
expect(executionData.snapshot.executionStatus).toBe("QUEUED");
97+
} finally {
98+
await engine.quit();
99+
}
100+
}
101+
);
102+
103+
containerTest(
104+
"a delayed run keeps DELAYED and QUEUED as separate snapshots",
105+
async ({ prisma, redisOptions }) => {
106+
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
107+
108+
const engine = new RunEngine({
109+
prisma,
110+
worker: {
111+
redis: redisOptions,
112+
workers: 1,
113+
tasksPerWorker: 10,
114+
pollIntervalMs: 100,
115+
},
116+
queue: {
117+
redis: redisOptions,
118+
masterQueueConsumersDisabled: true,
119+
processWorkerQueueDebounceMs: 50,
120+
},
121+
runLock: {
122+
redis: redisOptions,
123+
},
124+
machines: {
125+
defaultMachine: "small-1x",
126+
machines: {
127+
"small-1x": {
128+
name: "small-1x" as const,
129+
cpu: 0.5,
130+
memory: 0.5,
131+
centsPerMs: 0.0001,
132+
},
133+
},
134+
baseCostInCents: 0.0001,
135+
},
136+
tracer: trace.getTracer("test", "0.0.0"),
137+
});
138+
139+
try {
140+
const taskIdentifier = "test-task";
141+
142+
await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
143+
144+
const run = await engine.trigger(
145+
{
146+
number: 1,
147+
friendlyId: "run_1235",
148+
environment: authenticatedEnvironment,
149+
taskIdentifier,
150+
payload: "{}",
151+
payloadType: "application/json",
152+
context: {},
153+
traceContext: {},
154+
traceId: "t_collapse_2",
155+
spanId: "s_collapse_2",
156+
workerQueue: "main",
157+
queue: "task/test-task",
158+
isTest: false,
159+
tags: [],
160+
delayUntil: new Date(Date.now() + 500),
161+
},
162+
prisma
163+
);
164+
165+
expect(await snapshotStatuses(prisma, run.id)).toEqual(["DELAYED"]);
166+
167+
await setTimeout(1_500);
168+
169+
expect(await snapshotStatuses(prisma, run.id)).toEqual(["DELAYED", "QUEUED"]);
170+
} finally {
171+
await engine.quit();
172+
}
173+
}
174+
);
175+
});

‎internal-packages/run-store/src/PostgresRunStore.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -680,6 +680,7 @@ export class PostgresRunStore implements RunStore {
680680
const client = tx ?? this.prisma;
681681

682682
const snapshotCreate = {
683+
id: params.snapshot.id,
683684
engine: params.snapshot.engine,
684685
executionStatus: params.snapshot.executionStatus,
685686
description: params.snapshot.description,

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ export type IdempotencyKeyRunMatch = {
2929
};
3030

3131
export type CreateRunSnapshotInput = {
32+
id?: string;
3233
engine: "V2";
3334
executionStatus: TaskRunExecutionStatus;
3435
description: string;

0 commit comments

Comments
 (0)