feat(run-engine): count PENDING_EXECUTING stalls by outcome and plan type

Adds a `run_engine.pending_executing.stalls` counter to the run engine,
recorded when a run's PENDING_EXECUTING heartbeat deadline passes before
it starts, with `outcome` (`requeued` or `dropped`), `environment_type`
and `plan_type` attributes. The stalled-snapshot log line now includes
the organization, project and environment ids, and a redundant second
log line on the same path has been removed.

Mono-RevId: 88dc6e40f2f0946042f4a69d92d7c546fa0595b1

Author: nicktrn

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

@@ -125,6 +125,7 @@ export class RunEngine {
   private tracer: Tracer;
   private meter: Meter;
   private snapshotsSinceReplicaMissCounter: Counter;
+  private pendingExecutingStallsCounter: Counter;
   private snapshotsSinceReplicaRetryDelay: { minMs: number; maxMs: number };
   private heartbeatTimeouts: HeartbeatTimeouts;
   private repairSnapshotTimeoutMs: number;
@@ -369,6 +370,15 @@ export class RunEngine {
       }
     );
 
+    this.pendingExecutingStallsCounter = this.meter.createCounter(
+      "run_engine.pending_executing.stalls",
+      {
+        description:
+          "PENDING_EXECUTING snapshots whose heartbeat deadline passed before the run started, by whether the run was requeued or dropped",
+        unit: "runs",
+      }
+    );
+
     // Normalize the bounds, but keep maxMs <= 0 meaning "skip the replica retry".
     const retryDelay = options.readReplicaSnapshotsSinceRetryDelay ?? { minMs: 50, maxMs: 200 };
     const retryMinMs = Math.max(0, retryDelay.minMs);
@@ -2811,6 +2821,9 @@ export class RunEngine {
         snapshotId: latestSnapshot.id,
         executionStatus: latestSnapshot.executionStatus,
         environmentType: latestSnapshot.environmentType,
+        organizationId: latestSnapshot.organizationId,
+        projectId: latestSnapshot.projectId,
+        environmentId: latestSnapshot.environmentId,
       });
 
       switch (latestSnapshot.executionStatus) {
@@ -2824,11 +2837,6 @@ export class RunEngine {
           throw new NotImplementedError("There shouldn't be a heartbeat for QUEUED_EXECUTING");
         }
         case "PENDING_EXECUTING": {
-          this.logger.log("RunEngine stalled snapshot PENDING_EXECUTING", {
-            runId,
-            snapshotId: latestSnapshot.id,
-          });
-
           //the run didn't start executing, we need to requeue it
           const run = await this.runStore.findRun({ id: runId }, prisma);
 
@@ -2846,7 +2854,7 @@ export class RunEngine {
           }
 
           //it will automatically be requeued X times depending on the queue retry settings
-          await this.runAttemptSystem.tryNackAndRequeue({
+          const { wasRequeued } = await this.runAttemptSystem.tryNackAndRequeue({
             run,
             environment: {
               id: latestSnapshot.environmentId,
@@ -2864,6 +2872,13 @@ export class RunEngine {
             },
             tx: prisma,
           });
+
+          // run.planType rather than billingCache: the cache reports paid when billing is unreachable
+          this.pendingExecutingStallsCounter.add(1, {
+            outcome: wasRequeued ? "requeued" : "dropped",
+            environment_type: latestSnapshot.environmentType,
+            plan_type: run.planType ?? "unknown",
+          });
           break;
         }
         case "EXECUTING":