From 30d1303c8e548a4e8b2987a46fe157017cbcefea Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 29 Jul 2026 14:33:53 +0100 Subject: [PATCH 1/6] 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. --- .../collapse-trigger-queued-snapshot.md | 6 + .../run-engine/src/engine/index.ts | 33 +++- .../src/engine/systems/enqueueSystem.ts | 94 ++++++---- .../engine/tests/triggerCreateRouting.test.ts | 5 +- .../tests/triggerSnapshotCollapse.test.ts | 175 ++++++++++++++++++ .../run-store/src/PostgresRunStore.ts | 1 + internal-packages/run-store/src/types.ts | 1 + 7 files changed, 273 insertions(+), 42 deletions(-) create mode 100644 .server-changes/collapse-trigger-queued-snapshot.md create mode 100644 internal-packages/run-engine/src/engine/tests/triggerSnapshotCollapse.test.ts diff --git a/.server-changes/collapse-trigger-queued-snapshot.md b/.server-changes/collapse-trigger-queued-snapshot.md new file mode 100644 index 0000000000..ccc0169c6c --- /dev/null +++ b/.server-changes/collapse-trigger-queued-snapshot.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Triggering a task now does one less database write, so runs reach the queue marginally faster. diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index 07de80bed8..cbbbecbc8a 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -23,6 +23,7 @@ import { } from "@trigger.dev/core/v3"; import type { TaskRunError } from "@trigger.dev/core/v3/schemas"; import { + generateInternalId, parseNaturalLanguageDurationInMs, RunId, WaitpointId, @@ -950,6 +951,7 @@ export class RunEngine { let taskRun: TaskRun & { associatedWaitpoint: Waitpoint | null }; const taskRunId = RunId.fromFriendlyId(friendlyId); + const initialSnapshotId = generateInternalId(); // App-level replacement for the dropped TaskRun env/project Cascade FKs. await this.controlPlaneResolver.assertEnvExists(environment.id); @@ -1035,9 +1037,10 @@ export class RunEngine { annotations, }, snapshot: { + id: initialSnapshotId, engine: "V2", - executionStatus: delayUntil ? "DELAYED" : "RUN_CREATED", - description: delayUntil ? "Run is delayed" : "Run was created", + executionStatus: delayUntil ? "DELAYED" : "QUEUED", + description: delayUntil ? "Run is delayed" : "Run was QUEUED", runStatus: status, environmentId: environment.id, environmentType: environment.type, @@ -1164,13 +1167,29 @@ export class RunEngine { await this.ttlSystem.scheduleExpireRun({ runId: taskRun.id, ttl: taskRun.ttl }); } - await this.enqueueSystem.enqueueRun({ + this.eventBus.emit("executionSnapshotCreated", { + time: taskRun.createdAt, + run: { + id: taskRun.id, + }, + snapshot: { + id: initialSnapshotId, + executionStatus: "QUEUED", + description: "Run was QUEUED", + runStatus: taskRun.status, + attemptNumber: taskRun.attemptNumber ?? null, + checkpointId: null, + workerId: workerId ?? null, + runnerId: runnerId ?? null, + isValid: true, + error: null, + completedWaitpointIds: [], + }, + }); + + await this.enqueueSystem.publishRun({ run: taskRun, env: environment, - workerId, - runnerId, - tx: prisma, - skipRunLock: true, includeTtl: true, anchorEligibilityAtQueuePosition: true, enableFastPath, diff --git a/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts b/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts index 2b5aac6d37..75382ce82c 100644 --- a/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts @@ -113,42 +113,74 @@ export class EnqueueSystem { store ); - // Force development runs to use the environment id as the worker queue. - const workerQueue = env.type === "DEVELOPMENT" ? env.id : run.workerQueue; - - const queuePositionMs = (run.queueTimestamp ?? run.createdAt).getTime(); - const timestamp = queuePositionMs - run.priorityMs; - const eligibleAtMs = anchorEligibilityAtQueuePosition ? queuePositionMs : Date.now(); - - let ttlExpiresAt: number | undefined; - if (includeTtl && run.ttl) { - const expireAt = parseNaturalLanguageDuration(run.ttl); - if (expireAt) { - ttlExpiresAt = expireAt.getTime(); - } - } - - await this.$.runQueue.enqueueMessage({ + await this.publishRun({ + run, env, - workerQueue, + includeTtl, + anchorEligibilityAtQueuePosition, enableFastPath, - message: { - runId: run.id, - taskIdentifier: run.taskIdentifier, - orgId: env.organization.id, - projectId: env.project.id, - environmentId: env.id, - environmentType: env.type, - queue: run.queue, - concurrencyKey: run.concurrencyKey ?? undefined, - timestamp, - eligibleAtMs, - attempt: 0, - ttlExpiresAt, - }, }); return newSnapshot; }); } + + /** + * Publishes the run to the RunQueue without writing an execution snapshot. Callers that already + * hold a `QUEUED` snapshot (the trigger path writes one inside the run-create transaction) use + * this so the run does not pay for a second snapshot write. + */ + public async publishRun({ + run, + env, + includeTtl = false, + anchorEligibilityAtQueuePosition = false, + enableFastPath = false, + }: { + run: TaskRun; + env: MinimalAuthenticatedEnvironment; + /** See `enqueueRun`. */ + includeTtl?: boolean; + /** See `enqueueRun`. */ + anchorEligibilityAtQueuePosition?: boolean; + /** When true, allow the queue to push directly to worker queue if concurrency is available. */ + enableFastPath?: boolean; + }) { + // Force development runs to use the environment id as the worker queue. + const workerQueue = env.type === "DEVELOPMENT" ? env.id : run.workerQueue; + + const queuePositionMs = (run.queueTimestamp ?? run.createdAt).getTime(); + const timestamp = queuePositionMs - run.priorityMs; + const eligibleAtMs = anchorEligibilityAtQueuePosition ? queuePositionMs : Date.now(); + + // Include TTL only when explicitly requested (first enqueue from trigger). + // Re-enqueues (waitpoint, checkpoint, delayed, pending version) must not add TTL. + let ttlExpiresAt: number | undefined; + if (includeTtl && run.ttl) { + const expireAt = parseNaturalLanguageDuration(run.ttl); + if (expireAt) { + ttlExpiresAt = expireAt.getTime(); + } + } + + await this.$.runQueue.enqueueMessage({ + env, + workerQueue, + enableFastPath, + message: { + runId: run.id, + taskIdentifier: run.taskIdentifier, + orgId: env.organization.id, + projectId: env.project.id, + environmentId: env.id, + environmentType: env.type, + queue: run.queue, + concurrencyKey: run.concurrencyKey ?? undefined, + timestamp, + eligibleAtMs, + attempt: 0, + ttlExpiresAt, + }, + }); + } } diff --git a/internal-packages/run-engine/src/engine/tests/triggerCreateRouting.test.ts b/internal-packages/run-engine/src/engine/tests/triggerCreateRouting.test.ts index 34353cb44e..af213dc7a7 100644 --- a/internal-packages/run-engine/src/engine/tests/triggerCreateRouting.test.ts +++ b/internal-packages/run-engine/src/engine/tests/triggerCreateRouting.test.ts @@ -137,9 +137,6 @@ const cancelledSnapshot = (friendlyId: string, environment: any) => ({ }); describe("RunEngine trigger/create routing", () => { - // trigger create routes through runStore.createRun with the structured - // DTO, and the persisted run + its nested first RUN_CREATED snapshot land via - // the single create call. containerTest( "trigger routes createRun and lands run + first snapshot", async ({ prisma, redisOptions }) => { @@ -169,7 +166,7 @@ describe("RunEngine trigger/create routing", () => { orderBy: { createdAt: "asc" }, }); expect(snapshot).not.toBeNull(); - expect(snapshot!.executionStatus).toBe("RUN_CREATED"); + expect(snapshot!.executionStatus).toBe("QUEUED"); } finally { await engine.quit(); } diff --git a/internal-packages/run-engine/src/engine/tests/triggerSnapshotCollapse.test.ts b/internal-packages/run-engine/src/engine/tests/triggerSnapshotCollapse.test.ts new file mode 100644 index 0000000000..32372179ce --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/triggerSnapshotCollapse.test.ts @@ -0,0 +1,175 @@ +import { assertNonNullable, containerTest } from "@internal/testcontainers"; +import { trace } from "@internal/tracing"; +import type { PrismaClient, TaskRunExecutionStatus } from "@trigger.dev/database"; +import { setTimeout } from "node:timers/promises"; +import { expect } from "vitest"; +import { RunEngine } from "../index.js"; +import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js"; + +vi.setConfig({ testTimeout: 60_000 }); + +async function snapshotStatuses( + prisma: PrismaClient, + runId: string +): Promise { + const snapshots = await prisma.taskRunExecutionSnapshot.findMany({ + where: { runId }, + orderBy: { createdAt: "asc" }, + select: { executionStatus: true }, + }); + + return snapshots.map((snapshot) => snapshot.executionStatus); +} + +describe("RunEngine trigger() execution snapshots", () => { + containerTest( + "a non-delayed run is created with a single QUEUED snapshot", + async ({ prisma, redisOptions }) => { + const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + + const engine = new RunEngine({ + prisma, + worker: { + redis: redisOptions, + workers: 1, + tasksPerWorker: 10, + pollIntervalMs: 100, + }, + queue: { + redis: redisOptions, + masterQueueConsumersDisabled: true, + processWorkerQueueDebounceMs: 50, + }, + runLock: { + redis: redisOptions, + }, + machines: { + defaultMachine: "small-1x", + machines: { + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, + }, + }, + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier); + + const run = await engine.trigger( + { + number: 1, + friendlyId: "run_1234", + environment: authenticatedEnvironment, + taskIdentifier, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t_collapse_1", + spanId: "s_collapse_1", + workerQueue: "main", + queue: "task/test-task", + isTest: false, + tags: [], + }, + prisma + ); + + expect(await snapshotStatuses(prisma, run.id)).toEqual(["QUEUED"]); + + const queueLength = await engine.runQueue.lengthOfQueue( + authenticatedEnvironment, + run.queue + ); + expect(queueLength).toBe(1); + + const executionData = await engine.getRunExecutionData({ runId: run.id }); + assertNonNullable(executionData); + expect(executionData.snapshot.executionStatus).toBe("QUEUED"); + } finally { + await engine.quit(); + } + } + ); + + containerTest( + "a delayed run keeps DELAYED and QUEUED as separate snapshots", + async ({ prisma, redisOptions }) => { + const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + + const engine = new RunEngine({ + prisma, + worker: { + redis: redisOptions, + workers: 1, + tasksPerWorker: 10, + pollIntervalMs: 100, + }, + queue: { + redis: redisOptions, + masterQueueConsumersDisabled: true, + processWorkerQueueDebounceMs: 50, + }, + runLock: { + redis: redisOptions, + }, + machines: { + defaultMachine: "small-1x", + machines: { + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, + }, + }, + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier); + + const run = await engine.trigger( + { + number: 1, + friendlyId: "run_1235", + environment: authenticatedEnvironment, + taskIdentifier, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t_collapse_2", + spanId: "s_collapse_2", + workerQueue: "main", + queue: "task/test-task", + isTest: false, + tags: [], + delayUntil: new Date(Date.now() + 500), + }, + prisma + ); + + expect(await snapshotStatuses(prisma, run.id)).toEqual(["DELAYED"]); + + await setTimeout(1_500); + + expect(await snapshotStatuses(prisma, run.id)).toEqual(["DELAYED", "QUEUED"]); + } finally { + await engine.quit(); + } + } + ); +}); diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts index 1b0d2f89ce..676e345ad5 100644 --- a/internal-packages/run-store/src/PostgresRunStore.ts +++ b/internal-packages/run-store/src/PostgresRunStore.ts @@ -680,6 +680,7 @@ export class PostgresRunStore implements RunStore { const client = tx ?? this.prisma; const snapshotCreate = { + id: params.snapshot.id, engine: params.snapshot.engine, executionStatus: params.snapshot.executionStatus, description: params.snapshot.description, diff --git a/internal-packages/run-store/src/types.ts b/internal-packages/run-store/src/types.ts index 374a550e41..ac7cc35e3c 100644 --- a/internal-packages/run-store/src/types.ts +++ b/internal-packages/run-store/src/types.ts @@ -29,6 +29,7 @@ export type IdempotencyKeyRunMatch = { }; export type CreateRunSnapshotInput = { + id?: string; engine: "V2"; executionStatus: TaskRunExecutionStatus; description: string; From eee8cc51916e35d636a38435e0cd781e32408871 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 29 Jul 2026 15:00:50 +0100 Subject: [PATCH 2/6] test(run-engine): update snapshot-count expectations for the collapsed trigger write The store-routing counters now also count createRun, since the trigger path's QUEUED snapshot is written nested in the run create. The replica-lag test starts the run attempt so it still has a middle snapshot to exercise rather than degenerating to an empty comparison. --- .../collapse-trigger-queued-snapshot.md | 2 +- .../src/engine/systems/enqueueSystem.test.ts | 19 +++++++++++++------ .../systems/executionSnapshotSystem.test.ts | 8 ++++++++ .../engine/tests/getSnapshotsSince.test.ts | 7 ++++++- 4 files changed, 28 insertions(+), 8 deletions(-) diff --git a/.server-changes/collapse-trigger-queued-snapshot.md b/.server-changes/collapse-trigger-queued-snapshot.md index ccc0169c6c..541c5b6450 100644 --- a/.server-changes/collapse-trigger-queued-snapshot.md +++ b/.server-changes/collapse-trigger-queued-snapshot.md @@ -3,4 +3,4 @@ area: webapp type: improvement --- -Triggering a task now does one less database write, so runs reach the queue marginally faster. +Triggering a task now reaches the queue marginally faster. diff --git a/internal-packages/run-engine/src/engine/systems/enqueueSystem.test.ts b/internal-packages/run-engine/src/engine/systems/enqueueSystem.test.ts index a5730b306f..eca893712c 100644 --- a/internal-packages/run-engine/src/engine/systems/enqueueSystem.test.ts +++ b/internal-packages/run-engine/src/engine/systems/enqueueSystem.test.ts @@ -44,10 +44,10 @@ function createEngineOptions(redisOptions: any, prisma: any, store?: PostgresRun } /** - * A real PostgresRunStore subclass that counts the snapshot create method that enqueueRun's - * snapshot write routes through (via executionSnapshotSystem.createExecutionSnapshot). super.* - * runs the genuine store implementation, so the routing is observed over real containers without - * ever mocking prisma or the store. + * A real PostgresRunStore subclass that counts the store methods a run's QUEUED snapshot can be + * written through: nested in `createRun` on the trigger path, or standalone via + * `createExecutionSnapshot` on every re-enqueue. super.* runs the genuine store implementation, so + * the routing is observed over real containers without ever mocking prisma or the store. */ class CountingPostgresRunStore extends PostgresRunStore { public snapshotCreates = 0; @@ -59,12 +59,19 @@ class CountingPostgresRunStore extends PostgresRunStore { this.snapshotCreates++; return super.createExecutionSnapshot(input, tx); } + + override async createRun( + params: Parameters[0], + tx?: any + ): ReturnType { + this.snapshotCreates++; + return super.createRun(params, tx); + } } describe("RunEngine enqueueRun store routing", () => { - // The QUEUED snapshot written while enqueuing a run routes through the injected store. containerTest( - "enqueueRun snapshot routes through the store", + "the QUEUED snapshot routes through the store", async ({ prisma, redisOptions }) => { const countingStore = new CountingPostgresRunStore({ prisma, readOnlyPrisma: prisma }); const engine = new RunEngine(createEngineOptions(redisOptions, prisma, countingStore)); diff --git a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.test.ts b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.test.ts index aeb361019e..61d17a6b8b 100644 --- a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.test.ts +++ b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.test.ts @@ -65,6 +65,14 @@ class CountingPostgresRunStore extends PostgresRunStore { return super.createExecutionSnapshot(input, tx); } + override async createRun( + params: Parameters[0], + tx?: any + ): ReturnType { + this.creates++; + return super.createRun(params, tx); + } + override async findLatestExecutionSnapshot( runId: string, client?: any diff --git a/internal-packages/run-engine/src/engine/tests/getSnapshotsSince.test.ts b/internal-packages/run-engine/src/engine/tests/getSnapshotsSince.test.ts index 61ff3bff75..54059943fd 100644 --- a/internal-packages/run-engine/src/engine/tests/getSnapshotsSince.test.ts +++ b/internal-packages/run-engine/src/engine/tests/getSnapshotsSince.test.ts @@ -1149,11 +1149,16 @@ describe("RunEngine getSnapshotsSince", () => { ); await setTimeout(500); - await engine.dequeueFromWorkerQueue({ + const dequeued = await engine.dequeueFromWorkerQueue({ consumerId: "test_replica_stale_tail", workerQueue: "main", }); + await engine.startRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: dequeued[0].snapshot.id, + }); + const allSnapshots = await prisma.taskRunExecutionSnapshot.findMany({ where: { runId: run.id, isValid: true }, orderBy: { createdAt: "asc" }, From 29b372e687f6882887cf4bae0853e81227f7b115 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 29 Jul 2026 15:15:33 +0100 Subject: [PATCH 3/6] test(run-engine): build the replica-lag test's middle snapshot on the primary startRunAttempt resolves the background worker task through readOnlyPrisma, and this test configures a deliberately empty schema-only replica, so using it to add a snapshot failed on the replica read instead of exercising the lag scenario. Write the snapshot with the file's own primary-side helper. --- .../src/engine/tests/getSnapshotsSince.test.ts | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/internal-packages/run-engine/src/engine/tests/getSnapshotsSince.test.ts b/internal-packages/run-engine/src/engine/tests/getSnapshotsSince.test.ts index 54059943fd..4cf74753e0 100644 --- a/internal-packages/run-engine/src/engine/tests/getSnapshotsSince.test.ts +++ b/internal-packages/run-engine/src/engine/tests/getSnapshotsSince.test.ts @@ -7,7 +7,7 @@ import type { PrismaClient } from "@trigger.dev/database"; import { RunEngine } from "../index.js"; import { getExecutionSnapshotsSince } from "../systems/executionSnapshotSystem.js"; import { copySnapshotsToReplica, createTestMetricsMeter } from "./helpers/replicaTestHelpers.js"; -import { setupTestScenario } from "./helpers/snapshotTestHelpers.js"; +import { createTestSnapshot, setupTestScenario } from "./helpers/snapshotTestHelpers.js"; import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js"; vi.setConfig({ testTimeout: 120_000 }); @@ -1149,14 +1149,18 @@ describe("RunEngine getSnapshotsSince", () => { ); await setTimeout(500); - const dequeued = await engine.dequeueFromWorkerQueue({ + await engine.dequeueFromWorkerQueue({ consumerId: "test_replica_stale_tail", workerQueue: "main", }); - await engine.startRunAttempt({ - runId: dequeued[0].run.id, - snapshotId: dequeued[0].snapshot.id, + await createTestSnapshot(prisma, { + runId: run.id, + status: "EXECUTING", + environmentId: authenticatedEnvironment.id, + environmentType: authenticatedEnvironment.type, + projectId: authenticatedEnvironment.project.id, + organizationId: authenticatedEnvironment.organization.id, }); const allSnapshots = await prisma.taskRunExecutionSnapshot.findMany({ From c48c44defded5280216d0cc8faaa0062fcbfaf13 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 29 Jul 2026 15:33:48 +0100 Subject: [PATCH 4/6] fix(run-engine): stamp the queued timeline event at write time The run's createdAt is caller-supplied and the scheduler sets it to the exact schedule time, so using it for the emitted event backdated the queued entry on every scheduled run's timeline. Use the write moment, as the other nested-create paths do. --- .../run-engine/src/engine/index.ts | 2 +- .../tests/triggerSnapshotCollapse.test.ts | 81 +++++++++++++++++++ 2 files changed, 82 insertions(+), 1 deletion(-) diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index cbbbecbc8a..6f12b70ca6 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -1168,7 +1168,7 @@ export class RunEngine { } this.eventBus.emit("executionSnapshotCreated", { - time: taskRun.createdAt, + time: new Date(), run: { id: taskRun.id, }, diff --git a/internal-packages/run-engine/src/engine/tests/triggerSnapshotCollapse.test.ts b/internal-packages/run-engine/src/engine/tests/triggerSnapshotCollapse.test.ts index 32372179ce..cddb2756c3 100644 --- a/internal-packages/run-engine/src/engine/tests/triggerSnapshotCollapse.test.ts +++ b/internal-packages/run-engine/src/engine/tests/triggerSnapshotCollapse.test.ts @@ -100,6 +100,87 @@ describe("RunEngine trigger() execution snapshots", () => { } ); + containerTest( + "the QUEUED snapshot event is stamped at write time, not at an overridden run createdAt", + async ({ prisma, redisOptions }) => { + const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + + const engine = new RunEngine({ + prisma, + worker: { + redis: redisOptions, + workers: 1, + tasksPerWorker: 10, + pollIntervalMs: 100, + }, + queue: { + redis: redisOptions, + masterQueueConsumersDisabled: true, + processWorkerQueueDebounceMs: 50, + }, + runLock: { + redis: redisOptions, + }, + machines: { + defaultMachine: "small-1x", + machines: { + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, + }, + }, + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier); + + const stampedTimes: Date[] = []; + engine.eventBus.on("executionSnapshotCreated", ({ time }) => { + stampedTimes.push(time); + }); + + const backdatedCreatedAt = new Date(Date.now() - 60 * 60 * 1000); + const triggeredAt = Date.now(); + + const run = await engine.trigger( + { + number: 1, + friendlyId: "run_1236", + environment: authenticatedEnvironment, + taskIdentifier, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t_collapse_3", + spanId: "s_collapse_3", + workerQueue: "main", + queue: "task/test-task", + isTest: false, + tags: [], + createdAt: backdatedCreatedAt, + }, + prisma + ); + + const storedRun = await prisma.taskRun.findUnique({ where: { id: run.id } }); + expect(storedRun?.createdAt.getTime()).toBe(backdatedCreatedAt.getTime()); + + expect(stampedTimes.length).toBe(1); + expect(stampedTimes[0].getTime()).toBeGreaterThanOrEqual(triggeredAt); + } finally { + await engine.quit(); + } + } + ); + containerTest( "a delayed run keeps DELAYED and QUEUED as separate snapshots", async ({ prisma, redisOptions }) => { From 7e78ed8b13226c0e06ffdd48f9aaf0bacd971d9b Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 31 Jul 2026 14:28:27 +0100 Subject: [PATCH 5/6] refactor(run-engine): share the queued snapshot status and description The trigger path writes this snapshot nested in the run create and emits the timeline event itself, while every re-enqueue writes it through enqueueRun. Keep the status and the default description in one place so the persisted row and the emitted event cannot drift apart. --- .server-changes/collapse-trigger-queued-snapshot.md | 2 +- internal-packages/run-engine/src/engine/consts.ts | 10 ++++++++++ internal-packages/run-engine/src/engine/index.ts | 9 +++++---- .../run-engine/src/engine/systems/enqueueSystem.ts | 5 +++-- 4 files changed, 19 insertions(+), 7 deletions(-) diff --git a/.server-changes/collapse-trigger-queued-snapshot.md b/.server-changes/collapse-trigger-queued-snapshot.md index 541c5b6450..dffc9671e6 100644 --- a/.server-changes/collapse-trigger-queued-snapshot.md +++ b/.server-changes/collapse-trigger-queued-snapshot.md @@ -3,4 +3,4 @@ area: webapp type: improvement --- -Triggering a task now reaches the queue marginally faster. +Triggering a task now does one fewer database write, so runs reach the queue slightly faster. diff --git a/internal-packages/run-engine/src/engine/consts.ts b/internal-packages/run-engine/src/engine/consts.ts index 6ea6f54c38..3f0bb59eb4 100644 --- a/internal-packages/run-engine/src/engine/consts.ts +++ b/internal-packages/run-engine/src/engine/consts.ts @@ -1 +1,11 @@ export const MAX_TASK_RUN_ATTEMPTS = 250; + +/** + * The status and description a run's default entry into the queue is written with. Shared because + * the trigger path writes this snapshot nested in the run-create transaction and then emits its own + * `executionSnapshotCreated`, while every re-enqueue writes it through `enqueueRun`. Three places + * have to agree, or the persisted row and the run timeline's `[engine]` entry drift apart. + * Re-enqueues that describe why they requeued pass their own description instead. + */ +export const QUEUED_SNAPSHOT_STATUS = "QUEUED" as const; +export const QUEUED_SNAPSHOT_DESCRIPTION = "Run was QUEUED"; diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index 6f12b70ca6..b88f4f276e 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -54,6 +54,7 @@ import { RunQueue } from "../run-queue/index.js"; import { RunQueueFullKeyProducer } from "../run-queue/keyProducer.js"; import type { AuthenticatedEnvironment, MinimalAuthenticatedEnvironment } from "../shared/index.js"; import { BillingCache } from "./billingCache.js"; +import { QUEUED_SNAPSHOT_DESCRIPTION, QUEUED_SNAPSHOT_STATUS } from "./consts.js"; import { ExecutionSnapshotNotFoundError, NotImplementedError, @@ -1039,8 +1040,8 @@ export class RunEngine { snapshot: { id: initialSnapshotId, engine: "V2", - executionStatus: delayUntil ? "DELAYED" : "QUEUED", - description: delayUntil ? "Run is delayed" : "Run was QUEUED", + executionStatus: delayUntil ? "DELAYED" : QUEUED_SNAPSHOT_STATUS, + description: delayUntil ? "Run is delayed" : QUEUED_SNAPSHOT_DESCRIPTION, runStatus: status, environmentId: environment.id, environmentType: environment.type, @@ -1174,8 +1175,8 @@ export class RunEngine { }, snapshot: { id: initialSnapshotId, - executionStatus: "QUEUED", - description: "Run was QUEUED", + executionStatus: QUEUED_SNAPSHOT_STATUS, + description: QUEUED_SNAPSHOT_DESCRIPTION, runStatus: taskRun.status, attemptNumber: taskRun.attemptNumber ?? null, checkpointId: null, diff --git a/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts b/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts index 75382ce82c..38c681c511 100644 --- a/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts @@ -7,6 +7,7 @@ import type { import type { RunStore } from "@internal/run-store"; import { parseNaturalLanguageDuration } from "@trigger.dev/core/v3/isomorphic"; import type { MinimalAuthenticatedEnvironment } from "../../shared/index.js"; +import { QUEUED_SNAPSHOT_DESCRIPTION, QUEUED_SNAPSHOT_STATUS } from "../consts.js"; import type { ExecutionSnapshotSystem } from "./executionSnapshotSystem.js"; import type { SystemResources } from "./systems.js"; @@ -95,8 +96,8 @@ export class EnqueueSystem { { run: run, snapshot: { - executionStatus: snapshot?.status ?? "QUEUED", - description: snapshot?.description ?? "Run was QUEUED", + executionStatus: snapshot?.status ?? QUEUED_SNAPSHOT_STATUS, + description: snapshot?.description ?? QUEUED_SNAPSHOT_DESCRIPTION, metadata: snapshot?.metadata ?? undefined, }, previousSnapshotId, From ed4836ee901c4d9643b89a24c2b8219758abd2ad Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 31 Jul 2026 14:28:50 +0100 Subject: [PATCH 6/6] test(run-engine): pin the single nested write and the resumed QUEUED The store-routing counter separated nested creates from standalone ones, so it can assert the trigger path makes no standalone snapshot write at all rather than just more writes than before. A new case drives a run through suspend and resume to prove a re-enqueue still writes its own QUEUED snapshot. --- .../src/engine/systems/enqueueSystem.test.ts | 17 ++- .../tests/triggerSnapshotCollapse.test.ts | 114 ++++++++++++++++++ 2 files changed, 125 insertions(+), 6 deletions(-) diff --git a/internal-packages/run-engine/src/engine/systems/enqueueSystem.test.ts b/internal-packages/run-engine/src/engine/systems/enqueueSystem.test.ts index eca893712c..8903129fc8 100644 --- a/internal-packages/run-engine/src/engine/systems/enqueueSystem.test.ts +++ b/internal-packages/run-engine/src/engine/systems/enqueueSystem.test.ts @@ -50,13 +50,16 @@ function createEngineOptions(redisOptions: any, prisma: any, store?: PostgresRun * the routing is observed over real containers without ever mocking prisma or the store. */ class CountingPostgresRunStore extends PostgresRunStore { - public snapshotCreates = 0; + /** Snapshots written nested in a run create: the trigger path's QUEUED write. */ + public nestedSnapshotCreates = 0; + /** Snapshots written standalone through `createExecutionSnapshot`: every re-enqueue. */ + public standaloneSnapshotCreates = 0; override async createExecutionSnapshot( input: any, tx?: any ): ReturnType { - this.snapshotCreates++; + this.standaloneSnapshotCreates++; return super.createExecutionSnapshot(input, tx); } @@ -64,14 +67,14 @@ class CountingPostgresRunStore extends PostgresRunStore { params: Parameters[0], tx?: any ): ReturnType { - this.snapshotCreates++; + this.nestedSnapshotCreates++; return super.createRun(params, tx); } } describe("RunEngine enqueueRun store routing", () => { containerTest( - "the QUEUED snapshot routes through the store", + "the QUEUED snapshot routes through the store as a single nested write", async ({ prisma, redisOptions }) => { const countingStore = new CountingPostgresRunStore({ prisma, readOnlyPrisma: prisma }); const engine = new RunEngine(createEngineOptions(redisOptions, prisma, countingStore)); @@ -81,7 +84,8 @@ describe("RunEngine enqueueRun store routing", () => { const taskIdentifier = "test-task"; await setupBackgroundWorker(engine, environment, taskIdentifier); - const before = countingStore.snapshotCreates; + const nestedBefore = countingStore.nestedSnapshotCreates; + const standaloneBefore = countingStore.standaloneSnapshotCreates; const run = await engine.trigger( { @@ -103,7 +107,8 @@ describe("RunEngine enqueueRun store routing", () => { prisma ); - expect(countingStore.snapshotCreates).toBeGreaterThan(before); + expect(countingStore.nestedSnapshotCreates).toBe(nestedBefore + 1); + expect(countingStore.standaloneSnapshotCreates).toBe(standaloneBefore); const latest = await getLatestExecutionSnapshot(prisma, run.id); assertNonNullable(latest); diff --git a/internal-packages/run-engine/src/engine/tests/triggerSnapshotCollapse.test.ts b/internal-packages/run-engine/src/engine/tests/triggerSnapshotCollapse.test.ts index cddb2756c3..bdb49419c0 100644 --- a/internal-packages/run-engine/src/engine/tests/triggerSnapshotCollapse.test.ts +++ b/internal-packages/run-engine/src/engine/tests/triggerSnapshotCollapse.test.ts @@ -253,4 +253,118 @@ describe("RunEngine trigger() execution snapshots", () => { } } ); + + containerTest( + "a resumed run still writes its own QUEUED snapshot", + async ({ prisma, redisOptions }) => { + const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + + const engine = new RunEngine({ + prisma, + worker: { + redis: redisOptions, + workers: 1, + tasksPerWorker: 10, + pollIntervalMs: 100, + }, + queue: { + redis: redisOptions, + masterQueueConsumersDisabled: true, + processWorkerQueueDebounceMs: 50, + }, + runLock: { + redis: redisOptions, + }, + machines: { + defaultMachine: "small-1x", + machines: { + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, + }, + }, + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier); + + const run = await engine.trigger( + { + number: 1, + friendlyId: "run_1237", + environment: authenticatedEnvironment, + taskIdentifier, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t_collapse_4", + spanId: "s_collapse_4", + workerQueue: "main", + queue: "task/test-task", + isTest: false, + tags: [], + }, + prisma + ); + + await setTimeout(500); + const dequeued = await engine.dequeueFromWorkerQueue({ + consumerId: "test_collapse_4", + workerQueue: "main", + }); + assertNonNullable(dequeued[0]); + + await engine.startRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: dequeued[0].snapshot.id, + }); + + const waitpointResult = await engine.createManualWaitpoint({ + environmentId: authenticatedEnvironment.id, + projectId: authenticatedEnvironment.projectId, + }); + + const blockedResult = await engine.blockRunWithWaitpoint({ + runId: run.id, + waitpoints: waitpointResult.waitpoint.id, + projectId: authenticatedEnvironment.projectId, + organizationId: authenticatedEnvironment.organizationId, + }); + + const checkpointResult = await engine.createCheckpoint({ + runId: run.id, + snapshotId: blockedResult.id, + checkpoint: { + type: "DOCKER", + reason: "TEST_CHECKPOINT", + location: "test-location", + imageRef: "test-image-ref", + }, + }); + expect(checkpointResult.ok).toBe(true); + + await engine.completeWaitpoint({ id: waitpointResult.waitpoint.id }); + await setTimeout(500); + + expect(await snapshotStatuses(prisma, run.id)).toEqual([ + "QUEUED", + "PENDING_EXECUTING", + "EXECUTING", + "EXECUTING_WITH_WAITPOINTS", + "SUSPENDED", + "QUEUED", + ]); + } finally { + await engine.quit(); + } + } + ); });