Skip to content

Commit f10bc23

Browse files
ericallamericallam
authored
perf(run-engine,run-store): one execution snapshot per triggered run (#4419)
A non-delayed run used to get two execution snapshots the moment it was triggered: `RUN_CREATED` nested in the run-create transaction, immediately followed by `QUEUED` from its own `BEGIN`/`INSERT`/`COMMIT`. It now gets a single `QUEUED` snapshot written inside the create, and the trigger path only publishes to the queue. One fewer row per run on `TaskRunExecutionSnapshot`, and one fewer round trip on the trigger hot path. `EnqueueSystem` gains a `publishRun` seam that enqueues without writing a snapshot. Every re-enqueue path (waitpoint resume, checkpoint restore, delayed enqueue, pending version, retry requeue) still calls `enqueueRun` and writes its own `QUEUED`, so only the first enqueue changes. The `QUEUED` snapshot still commits before the queue message, so a dequeue sees a dequeueable status exactly as before. Two things for reviewers. Nesting the write skips `createExecutionSnapshot`, which is what emits `executionSnapshotCreated` and therefore the run timeline's `[engine] QUEUED` entry, so the trigger path now emits it directly, the same way the dequeue and attempt-start paths already do for their nested creates. And `RUN_CREATED` is still written when a dequeued run has no background worker yet, so the status and both `statuses.ts` helpers stay live and existing rows keep reading correctly. Delayed runs are untouched: `DELAYED` then `QUEUED` are two genuinely different moments and stay two snapshots. Rollback is a revert. Create-and-enqueue happen in one request in one process, so no in-flight run needs both code paths to agree during a rollout. One note for whoever debugs this path later. The `QUEUED` snapshot now commits before the queue publish, so a failed publish leaves the run recorded as `QUEUED` with no queue message. That state was already reachable, since the publish was never part of the snapshot transaction, but it used to be recorded as `RUN_CREATED`, which was distinctive because it never otherwise persisted. `QUEUED` with no message is indistinguishable from a run waiting on a concurrency slot, so trigger-time publish failure is now one more cause of an apparently stuck queued run.
1 parent a91c08c commit f10bc23

11 files changed

Lines changed: 522 additions & 55 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 fewer database write, so runs reach the queue slightly faster.

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1 +1,11 @@
11
export const MAX_TASK_RUN_ATTEMPTS = 250;
2+
3+
/**
4+
* The status and description a run's default entry into the queue is written with. Shared because
5+
* the trigger path writes this snapshot nested in the run-create transaction and then emits its own
6+
* `executionSnapshotCreated`, while every re-enqueue writes it through `enqueueRun`. Three places
7+
* have to agree, or the persisted row and the run timeline's `[engine]` entry drift apart.
8+
* Re-enqueues that describe why they requeued pass their own description instead.
9+
*/
10+
export const QUEUED_SNAPSHOT_STATUS = "QUEUED" as const;
11+
export const QUEUED_SNAPSHOT_DESCRIPTION = "Run was QUEUED";

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

Lines changed: 27 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,
@@ -53,6 +54,7 @@ import { RunQueue } from "../run-queue/index.js";
5354
import { RunQueueFullKeyProducer } from "../run-queue/keyProducer.js";
5455
import type { AuthenticatedEnvironment, MinimalAuthenticatedEnvironment } from "../shared/index.js";
5556
import { BillingCache } from "./billingCache.js";
57+
import { QUEUED_SNAPSHOT_DESCRIPTION, QUEUED_SNAPSHOT_STATUS } from "./consts.js";
5658
import {
5759
ExecutionSnapshotNotFoundError,
5860
NotImplementedError,
@@ -950,6 +952,7 @@ export class RunEngine {
950952

951953
let taskRun: TaskRun & { associatedWaitpoint: Waitpoint | null };
952954
const taskRunId = RunId.fromFriendlyId(friendlyId);
955+
const initialSnapshotId = generateInternalId();
953956

954957
// App-level replacement for the dropped TaskRun env/project Cascade FKs.
955958
await this.controlPlaneResolver.assertEnvExists(environment.id);
@@ -1035,9 +1038,10 @@ export class RunEngine {
10351038
annotations,
10361039
},
10371040
snapshot: {
1041+
id: initialSnapshotId,
10381042
engine: "V2",
1039-
executionStatus: delayUntil ? "DELAYED" : "RUN_CREATED",
1040-
description: delayUntil ? "Run is delayed" : "Run was created",
1043+
executionStatus: delayUntil ? "DELAYED" : QUEUED_SNAPSHOT_STATUS,
1044+
description: delayUntil ? "Run is delayed" : QUEUED_SNAPSHOT_DESCRIPTION,
10411045
runStatus: status,
10421046
environmentId: environment.id,
10431047
environmentType: environment.type,
@@ -1164,13 +1168,29 @@ export class RunEngine {
11641168
await this.ttlSystem.scheduleExpireRun({ runId: taskRun.id, ttl: taskRun.ttl });
11651169
}
11661170

1167-
await this.enqueueSystem.enqueueRun({
1171+
this.eventBus.emit("executionSnapshotCreated", {
1172+
time: new Date(),
1173+
run: {
1174+
id: taskRun.id,
1175+
},
1176+
snapshot: {
1177+
id: initialSnapshotId,
1178+
executionStatus: QUEUED_SNAPSHOT_STATUS,
1179+
description: QUEUED_SNAPSHOT_DESCRIPTION,
1180+
runStatus: taskRun.status,
1181+
attemptNumber: taskRun.attemptNumber ?? null,
1182+
checkpointId: null,
1183+
workerId: workerId ?? null,
1184+
runnerId: runnerId ?? null,
1185+
isValid: true,
1186+
error: null,
1187+
completedWaitpointIds: [],
1188+
},
1189+
});
1190+
1191+
await this.enqueueSystem.publishRun({
11681192
run: taskRun,
11691193
env: environment,
1170-
workerId,
1171-
runnerId,
1172-
tx: prisma,
1173-
skipRunLock: true,
11741194
includeTtl: true,
11751195
anchorEligibilityAtQueuePosition: true,
11761196
enableFastPath,

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

Lines changed: 22 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -44,27 +44,37 @@ function createEngineOptions(redisOptions: any, prisma: any, store?: PostgresRun
4444
}
4545

4646
/**
47-
* A real PostgresRunStore subclass that counts the snapshot create method that enqueueRun's
48-
* snapshot write routes through (via executionSnapshotSystem.createExecutionSnapshot). super.*
49-
* runs the genuine store implementation, so the routing is observed over real containers without
50-
* ever mocking prisma or the store.
47+
* A real PostgresRunStore subclass that counts the store methods a run's QUEUED snapshot can be
48+
* written through: nested in `createRun` on the trigger path, or standalone via
49+
* `createExecutionSnapshot` on every re-enqueue. super.* runs the genuine store implementation, so
50+
* the routing is observed over real containers without ever mocking prisma or the store.
5151
*/
5252
class CountingPostgresRunStore extends PostgresRunStore {
53-
public snapshotCreates = 0;
53+
/** Snapshots written nested in a run create: the trigger path's QUEUED write. */
54+
public nestedSnapshotCreates = 0;
55+
/** Snapshots written standalone through `createExecutionSnapshot`: every re-enqueue. */
56+
public standaloneSnapshotCreates = 0;
5457

5558
override async createExecutionSnapshot(
5659
input: any,
5760
tx?: any
5861
): ReturnType<PostgresRunStore["createExecutionSnapshot"]> {
59-
this.snapshotCreates++;
62+
this.standaloneSnapshotCreates++;
6063
return super.createExecutionSnapshot(input, tx);
6164
}
65+
66+
override async createRun(
67+
params: Parameters<PostgresRunStore["createRun"]>[0],
68+
tx?: any
69+
): ReturnType<PostgresRunStore["createRun"]> {
70+
this.nestedSnapshotCreates++;
71+
return super.createRun(params, tx);
72+
}
6273
}
6374

6475
describe("RunEngine enqueueRun store routing", () => {
65-
// The QUEUED snapshot written while enqueuing a run routes through the injected store.
6676
containerTest(
67-
"enqueueRun snapshot routes through the store",
77+
"the QUEUED snapshot routes through the store as a single nested write",
6878
async ({ prisma, redisOptions }) => {
6979
const countingStore = new CountingPostgresRunStore({ prisma, readOnlyPrisma: prisma });
7080
const engine = new RunEngine(createEngineOptions(redisOptions, prisma, countingStore));
@@ -74,7 +84,8 @@ describe("RunEngine enqueueRun store routing", () => {
7484
const taskIdentifier = "test-task";
7585
await setupBackgroundWorker(engine, environment, taskIdentifier);
7686

77-
const before = countingStore.snapshotCreates;
87+
const nestedBefore = countingStore.nestedSnapshotCreates;
88+
const standaloneBefore = countingStore.standaloneSnapshotCreates;
7889

7990
const run = await engine.trigger(
8091
{
@@ -96,7 +107,8 @@ describe("RunEngine enqueueRun store routing", () => {
96107
prisma
97108
);
98109

99-
expect(countingStore.snapshotCreates).toBeGreaterThan(before);
110+
expect(countingStore.nestedSnapshotCreates).toBe(nestedBefore + 1);
111+
expect(countingStore.standaloneSnapshotCreates).toBe(standaloneBefore);
100112

101113
const latest = await getLatestExecutionSnapshot(prisma, run.id);
102114
assertNonNullable(latest);

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

Lines changed: 66 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import type {
77
import type { RunStore } from "@internal/run-store";
88
import { parseNaturalLanguageDuration } from "@trigger.dev/core/v3/isomorphic";
99
import type { MinimalAuthenticatedEnvironment } from "../../shared/index.js";
10+
import { QUEUED_SNAPSHOT_DESCRIPTION, QUEUED_SNAPSHOT_STATUS } from "../consts.js";
1011
import type { ExecutionSnapshotSystem } from "./executionSnapshotSystem.js";
1112
import type { SystemResources } from "./systems.js";
1213

@@ -95,8 +96,8 @@ export class EnqueueSystem {
9596
{
9697
run: run,
9798
snapshot: {
98-
executionStatus: snapshot?.status ?? "QUEUED",
99-
description: snapshot?.description ?? "Run was QUEUED",
99+
executionStatus: snapshot?.status ?? QUEUED_SNAPSHOT_STATUS,
100+
description: snapshot?.description ?? QUEUED_SNAPSHOT_DESCRIPTION,
100101
metadata: snapshot?.metadata ?? undefined,
101102
},
102103
previousSnapshotId,
@@ -113,42 +114,74 @@ export class EnqueueSystem {
113114
store
114115
);
115116

116-
// Force development runs to use the environment id as the worker queue.
117-
const workerQueue = env.type === "DEVELOPMENT" ? env.id : run.workerQueue;
118-
119-
const queuePositionMs = (run.queueTimestamp ?? run.createdAt).getTime();
120-
const timestamp = queuePositionMs - run.priorityMs;
121-
const eligibleAtMs = anchorEligibilityAtQueuePosition ? queuePositionMs : Date.now();
122-
123-
let ttlExpiresAt: number | undefined;
124-
if (includeTtl && run.ttl) {
125-
const expireAt = parseNaturalLanguageDuration(run.ttl);
126-
if (expireAt) {
127-
ttlExpiresAt = expireAt.getTime();
128-
}
129-
}
130-
131-
await this.$.runQueue.enqueueMessage({
117+
await this.publishRun({
118+
run,
132119
env,
133-
workerQueue,
120+
includeTtl,
121+
anchorEligibilityAtQueuePosition,
134122
enableFastPath,
135-
message: {
136-
runId: run.id,
137-
taskIdentifier: run.taskIdentifier,
138-
orgId: env.organization.id,
139-
projectId: env.project.id,
140-
environmentId: env.id,
141-
environmentType: env.type,
142-
queue: run.queue,
143-
concurrencyKey: run.concurrencyKey ?? undefined,
144-
timestamp,
145-
eligibleAtMs,
146-
attempt: 0,
147-
ttlExpiresAt,
148-
},
149123
});
150124

151125
return newSnapshot;
152126
});
153127
}
128+
129+
/**
130+
* Publishes the run to the RunQueue without writing an execution snapshot. Callers that already
131+
* hold a `QUEUED` snapshot (the trigger path writes one inside the run-create transaction) use
132+
* this so the run does not pay for a second snapshot write.
133+
*/
134+
public async publishRun({
135+
run,
136+
env,
137+
includeTtl = false,
138+
anchorEligibilityAtQueuePosition = false,
139+
enableFastPath = false,
140+
}: {
141+
run: TaskRun;
142+
env: MinimalAuthenticatedEnvironment;
143+
/** See `enqueueRun`. */
144+
includeTtl?: boolean;
145+
/** See `enqueueRun`. */
146+
anchorEligibilityAtQueuePosition?: boolean;
147+
/** When true, allow the queue to push directly to worker queue if concurrency is available. */
148+
enableFastPath?: boolean;
149+
}) {
150+
// Force development runs to use the environment id as the worker queue.
151+
const workerQueue = env.type === "DEVELOPMENT" ? env.id : run.workerQueue;
152+
153+
const queuePositionMs = (run.queueTimestamp ?? run.createdAt).getTime();
154+
const timestamp = queuePositionMs - run.priorityMs;
155+
const eligibleAtMs = anchorEligibilityAtQueuePosition ? queuePositionMs : Date.now();
156+
157+
// Include TTL only when explicitly requested (first enqueue from trigger).
158+
// Re-enqueues (waitpoint, checkpoint, delayed, pending version) must not add TTL.
159+
let ttlExpiresAt: number | undefined;
160+
if (includeTtl && run.ttl) {
161+
const expireAt = parseNaturalLanguageDuration(run.ttl);
162+
if (expireAt) {
163+
ttlExpiresAt = expireAt.getTime();
164+
}
165+
}
166+
167+
await this.$.runQueue.enqueueMessage({
168+
env,
169+
workerQueue,
170+
enableFastPath,
171+
message: {
172+
runId: run.id,
173+
taskIdentifier: run.taskIdentifier,
174+
orgId: env.organization.id,
175+
projectId: env.project.id,
176+
environmentId: env.id,
177+
environmentType: env.type,
178+
queue: run.queue,
179+
concurrencyKey: run.concurrencyKey ?? undefined,
180+
timestamp,
181+
eligibleAtMs,
182+
attempt: 0,
183+
ttlExpiresAt,
184+
},
185+
});
186+
}
154187
}

‎internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.test.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,14 @@ class CountingPostgresRunStore extends PostgresRunStore {
6565
return super.createExecutionSnapshot(input, tx);
6666
}
6767

68+
override async createRun(
69+
params: Parameters<PostgresRunStore["createRun"]>[0],
70+
tx?: any
71+
): ReturnType<PostgresRunStore["createRun"]> {
72+
this.creates++;
73+
return super.createRun(params, tx);
74+
}
75+
6876
override async findLatestExecutionSnapshot(
6977
runId: string,
7078
client?: any

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

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ import type { PrismaClient } from "@trigger.dev/database";
77
import { RunEngine } from "../index.js";
88
import { getExecutionSnapshotsSince } from "../systems/executionSnapshotSystem.js";
99
import { copySnapshotsToReplica, createTestMetricsMeter } from "./helpers/replicaTestHelpers.js";
10-
import { setupTestScenario } from "./helpers/snapshotTestHelpers.js";
10+
import { createTestSnapshot, setupTestScenario } from "./helpers/snapshotTestHelpers.js";
1111
import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js";
1212

1313
vi.setConfig({ testTimeout: 120_000 });
@@ -1154,6 +1154,15 @@ describe("RunEngine getSnapshotsSince", () => {
11541154
workerQueue: "main",
11551155
});
11561156

1157+
await createTestSnapshot(prisma, {
1158+
runId: run.id,
1159+
status: "EXECUTING",
1160+
environmentId: authenticatedEnvironment.id,
1161+
environmentType: authenticatedEnvironment.type,
1162+
projectId: authenticatedEnvironment.project.id,
1163+
organizationId: authenticatedEnvironment.organization.id,
1164+
});
1165+
11571166
const allSnapshots = await prisma.taskRunExecutionSnapshot.findMany({
11581167
where: { runId: run.id, isValid: true },
11591168
orderBy: { createdAt: "asc" },

‎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
}

0 commit comments

Comments
 (0)