bring back internal duration timers

Author: nicktrn

.changeset/hungry-sloths-promise.md

@@ -8,6 +8,5 @@ Fix issues that could result in unreezable state run crashes. Details:
 - Never checkpoint between attempts
 - Some messages and socket data now include attempt numbers
 - Remove attempt completion replays
-- Require external resume message after duration wait
 - Additional prod entry point logging
 - Fail runs that receive deprecated (pre-lazy attempt) execute messages

packages/cli-v3/src/workers/prod/entry-point.ts

@@ -27,6 +27,7 @@ import { randomUUID } from "node:crypto";
 import { readFile } from "node:fs/promises";
 import { createServer } from "node:http";
 import { setTimeout as timeout } from "node:timers/promises";
+import { Evt } from "evt";
 
 declare const __PROJECT_CONFIG__: Config;
 
@@ -64,6 +65,8 @@ class ProdWorker {
   private waitForPostStart = false;
   private connectionCount = 0;
 
+  private restoreNotification = Evt.create();
+
   private waitForTaskReplay:
     | {
         idempotencyKey: string;
@@ -500,6 +503,17 @@ class ProdWorker {
         // checkpointSafeInternalTimeout is accurate even after non-simulated restores
         await Promise.race([internalTimeout, checkpointSafeInternalTimeout]);
 
+        const idempotencyKey = randomUUID();
+        this.durationResumeFallback = { idempotencyKey };
+
+        try {
+          await this.restoreNotification.waitFor(5_000);
+        } catch (error) {
+          logger.error("Did not receive restore notification in time", {
+            error,
+          });
+        }
+
         try {
           // The coordinator should cancel any in-progress checkpoints so we don't end up with race conditions
           const { checkpointCanceled } = await this.#coordinatorSocket.socket
@@ -518,9 +532,6 @@ class ProdWorker {
 
           logger.log("Waiting for external duration resume as we may have been restored");
 
-          const idempotencyKey = randomUUID();
-          this.durationResumeFallback = { idempotencyKey };
-
           setTimeout(() => {
             if (!this.durationResumeFallback) {
               logger.error("Already resumed after duration, skipping fallback");
@@ -633,6 +644,8 @@ class ProdWorker {
     this.nextResumeAfter = undefined;
     this.waitForPostStart = false;
 
+    this.durationResumeFallback = undefined;
+
     this.#backgroundWorker.waitCompletedNotification();
   }
 
@@ -875,8 +888,6 @@ class ProdWorker {
             return;
           }
 
-          this.durationResumeFallback = undefined;
-
           this.#resumeAfterDuration();
         },
         EXECUTE_TASK_RUN: async () => {
@@ -1316,6 +1327,7 @@ class ProdWorker {
               }
               case "restore": {
                 await this.#reconnectAfterPostStart();
+                this.restoreNotification.post();
                 break;
               }
               default: {