perf(run-engine,run-store): one execution snapshot per triggered run

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

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Triggering a task now does one fewer database write, so runs reach the queue slightly faster.

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";

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,
@@ -53,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,
@@ -950,6 +952,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 +1038,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_SNAPSHOT_STATUS,
+                description: delayUntil ? "Run is delayed" : QUEUED_SNAPSHOT_DESCRIPTION,
                 runStatus: status,
                 environmentId: environment.id,
                 environmentType: environment.type,
@@ -1164,13 +1168,29 @@ export class RunEngine {
               await this.ttlSystem.scheduleExpireRun({ runId: taskRun.id, ttl: taskRun.ttl });
             }
 
-            await this.enqueueSystem.enqueueRun({
+            this.eventBus.emit("executionSnapshotCreated", {
+              time: new Date(),
+              run: {
+                id: taskRun.id,
+              },
+              snapshot: {
+                id: initialSnapshotId,
+                executionStatus: QUEUED_SNAPSHOT_STATUS,
+                description: QUEUED_SNAPSHOT_DESCRIPTION,
+                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,

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

@@ -44,27 +44,37 @@ 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;
+  /** 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<PostgresRunStore["createExecutionSnapshot"]> {
-    this.snapshotCreates++;
+    this.standaloneSnapshotCreates++;
     return super.createExecutionSnapshot(input, tx);
   }
+
+  override async createRun(
+    params: Parameters<PostgresRunStore["createRun"]>[0],
+    tx?: any
+  ): ReturnType<PostgresRunStore["createRun"]> {
+    this.nestedSnapshotCreates++;
+    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 as a single nested write",
     async ({ prisma, redisOptions }) => {
       const countingStore = new CountingPostgresRunStore({ prisma, readOnlyPrisma: prisma });
       const engine = new RunEngine(createEngineOptions(redisOptions, prisma, countingStore));
@@ -74,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(
           {
@@ -96,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);

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,
@@ -113,42 +114,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,
+      },
+    });
+  }
 }

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<PostgresRunStore["createRun"]>[0],
+    tx?: any
+  ): ReturnType<PostgresRunStore["createRun"]> {
+    this.creates++;
+    return super.createRun(params, tx);
+  }
+
   override async findLatestExecutionSnapshot(
     runId: string,
     client?: any

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 });
@@ -1154,6 +1154,15 @@ describe("RunEngine getSnapshotsSince", () => {
           workerQueue: "main",
         });
 
+        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({
           where: { runId: run.id, isValid: true },
           orderBy: { createdAt: "asc" },

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();
       }