fix: heartbeat race condition (acking) (#2380)

* The logger now supports metadata

* Added metadata to ServiceValidationError in some critical places

* Don't ack the heartbeat if there's a mismatch, it might prevent a brand new one

* Don't ack the heartbeat inside stalled. By returning it will be acked IF the deduplication key matches

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

@@ -59,7 +59,8 @@ export function runStatusFromError(error: TaskRunError): TaskRunStatus {
 export class ServiceValidationError extends Error {
   constructor(
     message: string,
-    public status?: number
+    public status?: number,
+    public metadata?: Record<string, unknown>
   ) {
     super(message);
     this.name = "ServiceValidationError";

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

@@ -1155,7 +1155,6 @@ export class RunEngine {
           }
         );
 
-        await this.worker.ack(`heartbeatSnapshot.${runId}`);
         return;
       }
 

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

@@ -270,11 +270,25 @@ export class CheckpointSystem {
       const snapshot = await getLatestExecutionSnapshot(prisma, runId);
 
       if (snapshot.id !== snapshotId) {
-        throw new ServiceValidationError("Snapshot ID doesn't match the latest snapshot", 400);
+        throw new ServiceValidationError(
+          "Snapshot ID doesn't match the latest snapshot in continueRunExecution",
+          400,
+          {
+            snapshotId,
+            latestSnapshotId: snapshot.id,
+          }
+        );
       }
 
       if (!isPendingExecuting(snapshot.executionStatus)) {
-        throw new ServiceValidationError("Snapshot is not in a valid state to continue", 400);
+        throw new ServiceValidationError(
+          "Snapshot is not in a valid state to continue in continueRunExecution",
+          400,
+          {
+            snapshotId,
+            snapshotStatus: snapshot.executionStatus,
+          }
+        );
       }
 
       // Get the run and update the status

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

@@ -361,7 +361,6 @@ export class ExecutionSnapshotSystem {
         runnerId,
       });
 
-      await this.$.worker.ack(`heartbeatSnapshot.${runId}`);
       return executionResultFromSnapshot(latestSnapshot);
     }
 

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

@@ -318,9 +318,16 @@ export class RunAttemptSystem {
             //if there is a big delay between the snapshot and the attempt, the snapshot might have changed
             //we just want to log because elsewhere it should have been put back into a state where it can be attempted
             this.$.logger.warn(
-              "RunEngine.createRunAttempt(): snapshot has changed since the attempt was created, ignoring."
+              "RunEngine.createRunAttempt(): snapshot has changed since the attempt was created, ignoring.",
+              {
+                snapshotId,
+                latestSnapshotId: latestSnapshot.id,
+              }
             );
-            throw new ServiceValidationError("Snapshot changed", 409);
+            throw new ServiceValidationError("Snapshot changed inside startRunAttempt", 409, {
+              snapshotId,
+              latestSnapshotId: latestSnapshot.id,
+            });
           }
 
           const taskRun = await prisma.taskRun.findFirst({

packages/core/src/logger.ts

@@ -147,6 +147,7 @@ function extractStructuredErrorFromArgs(...args: Array<Record<string, unknown> |
       message: error.message,
       stack: error.stack,
       name: error.name,
+      metadata: "metadata" in error ? error.metadata : undefined,
     };
   }
 
@@ -157,6 +158,7 @@ function extractStructuredErrorFromArgs(...args: Array<Record<string, unknown> |
       message: structuredError.error.message,
       stack: structuredError.error.stack,
       name: structuredError.error.name,
+      metadata: "metadata" in structuredError.error ? structuredError.error.metadata : undefined,
     };
   }