perf(run-engine): merge dequeue snapshot creation into taskRun.update transaction

Nest the TaskRunExecutionSnapshot create inside the preceding
taskRun.update() call in the dequeue flow, reducing 2 explicit
BEGIN/COMMIT transactions to 1 per dequeue operation.

This follows the same pattern already used in the completion path
(runAttemptSystem.ts:735) and trigger path (engine/index.ts:674-686).

Side effects (heartbeat enqueue, executionSnapshotCreated event) are
kept outside the transaction and fed the result of the merged write.

Also adds a public enqueueHeartbeatIfNeeded() method to
ExecutionSnapshotSystem for reuse by other flows that will adopt
the same merged pattern.

Refs: TRI-8450
Co-Authored-By: Eric Allam <eallam@icloud.com>

Author: devin-ai-integration[bot]

.server-changes/merge-dequeue-snapshot-into-transaction.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Merge execution snapshot creation into the dequeue taskRun.update transaction, reducing 2 DB commits to 1 per dequeue operation

internal-packages/run-engine/src/engine/systems/dequeueSystem.ts

@@ -435,6 +435,32 @@ export class DequeueSystem {
                   cliVersion: result.worker.cliVersion,
                   maxDurationInSeconds,
                   maxAttempts: maxAttempts ?? undefined,
+                  executionSnapshots: {
+                    create: {
+                      engine: "V2",
+                      executionStatus: "PENDING_EXECUTING",
+                      description: "Run was dequeued for execution",
+                      // Map DEQUEUED -> PENDING for backwards compatibility with older runners
+                      runStatus: "PENDING",
+                      attemptNumber: result.run.attemptNumber ?? undefined,
+                      previousSnapshotId: snapshot.id,
+                      environmentId: snapshot.environmentId,
+                      environmentType: snapshot.environmentType,
+                      projectId: snapshot.projectId,
+                      organizationId: snapshot.organizationId,
+                      checkpointId: snapshot.checkpointId ?? undefined,
+                      batchId: snapshot.batchId ?? undefined,
+                      completedWaitpoints: {
+                        connect: snapshot.completedWaitpoints.map((w) => ({ id: w.id })),
+                      },
+                      completedWaitpointOrder: snapshot.completedWaitpoints
+                        .filter((c) => c.index !== undefined)
+                        .sort((a, b) => a.index! - b.index!)
+                        .map((w) => w.id),
+                      workerId,
+                      runnerId,
+                    },
+                  },
                 },
                 include: {
                   runtimeEnvironment: true,
@@ -516,30 +542,22 @@ export class DequeueSystem {
                 hasPrivateLink = billingResult.val.hasPrivateLink;
               }
 
-              const newSnapshot = await this.executionSnapshotSystem.createExecutionSnapshot(
-                prisma,
-                {
-                  run: {
-                    id: runId,
-                    status: lockedTaskRun.status,
-                    attemptNumber: lockedTaskRun.attemptNumber,
-                  },
-                  snapshot: {
-                    executionStatus: "PENDING_EXECUTING",
-                    description: "Run was dequeued for execution",
-                  },
-                  previousSnapshotId: snapshot.id,
-                  environmentId: snapshot.environmentId,
-                  environmentType: snapshot.environmentType,
-                  projectId: snapshot.projectId,
-                  organizationId: snapshot.organizationId,
-                  checkpointId: snapshot.checkpointId ?? undefined,
-                  batchId: snapshot.batchId ?? undefined,
-                  completedWaitpoints: snapshot.completedWaitpoints,
-                  workerId,
-                  runnerId,
-                }
-              );
+              // Snapshot was created as part of the taskRun.update above (single transaction).
+              // Fetch the enhanced snapshot and handle side effects (heartbeat + event) manually.
+              const newSnapshot = await getLatestExecutionSnapshot(prisma, runId);
+
+              this.$.eventBus.emit("executionSnapshotCreated", {
+                time: newSnapshot.createdAt,
+                run: {
+                  id: newSnapshot.runId,
+                },
+                snapshot: {
+                  ...newSnapshot,
+                  completedWaitpointIds: newSnapshot.completedWaitpoints.map((wp) => wp.id),
+                },
+              });
+
+              await this.executionSnapshotSystem.enqueueHeartbeatIfNeeded(newSnapshot);
 
               return {
                 version: "1" as const,

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

@@ -518,6 +518,27 @@ export class ExecutionSnapshotSystem {
     return executionResultFromSnapshot(latestSnapshot);
   }
 
+  /**
+   * Enqueues a heartbeat job for a snapshot if the execution status requires one.
+   * Use this after nesting a snapshot create inside a taskRun.update() to replicate
+   * the heartbeat side effect that createExecutionSnapshot normally handles.
+   */
+  public async enqueueHeartbeatIfNeeded(snapshot: {
+    id: string;
+    runId: string;
+    executionStatus: TaskRunExecutionStatus;
+  }) {
+    const intervalMs = this.#getHeartbeatIntervalMs(snapshot.executionStatus);
+    if (intervalMs !== null) {
+      await this.$.worker.enqueue({
+        id: `heartbeatSnapshot.${snapshot.runId}`,
+        job: "heartbeatSnapshot",
+        payload: { snapshotId: snapshot.id, runId: snapshot.runId },
+        availableAt: new Date(Date.now() + intervalMs),
+      });
+    }
+  }
+
   #getHeartbeatIntervalMs(status: TaskRunExecutionStatus): number | null {
     switch (status) {
       case "PENDING_EXECUTING": {