Fixed the heartbeating

packages/cli-v3/src/dev/backgroundWorker.ts

@@ -143,7 +143,7 @@ export class BackgroundWorkerCoordinator {
     this._backgroundWorkers.clear();
   }
 
-  async executeTaskRun(id: string, payload: TaskRunExecutionPayload, messageId?: string) {
+  async executeTaskRun(id: string, payload: TaskRunExecutionPayload, messageId: string) {
     const worker = this._backgroundWorkers.get(id);
 
     if (!worker) {
@@ -152,7 +152,7 @@ export class BackgroundWorkerCoordinator {
     }
 
     try {
-      const completion = await worker.executeTaskRun(payload);
+      const completion = await worker.executeTaskRun(payload, messageId);
 
       this.onTaskCompleted.post({
         completion,
@@ -351,7 +351,7 @@ export class BackgroundWorker {
 
   async #getFreshTaskRunProcess(
     payload: TaskRunExecutionPayload,
-    messageId?: string
+    messageId: string
   ): Promise<TaskRunProcess> {
     logger.debug(this.#prefixedMessage(payload, "getFreshTaskRunProcess()"));
 
@@ -494,7 +494,7 @@ export class BackgroundWorker {
   // We need to fork the process before we can execute any tasks
   async executeTaskRun(
     payload: TaskRunExecutionPayload,
-    messageId?: string
+    messageId: string
   ): Promise<TaskRunExecutionResult> {
     if (this._closed) {
       throw new Error("Worker is closed");
@@ -523,7 +523,7 @@ export class BackgroundWorker {
 
   async #doExecuteTaskRun(
     payload: TaskRunExecutionPayload,
-    messageId?: string
+    messageId: string
   ): Promise<TaskRunExecutionResult> {
     try {
       const taskRunProcess = await this.#getFreshTaskRunProcess(payload, messageId);

packages/cli-v3/src/dev/workerRuntime.ts

@@ -213,46 +213,6 @@ class DevWorkerRuntime implements WorkerRuntime {
     eventBus.emit("backgroundWorkerInitialized", backgroundWorker);
   }
 
-  async #fetchTaskFiles(
-    sources: Record<string, { contents: string; contentHash: string }>,
-    tasks: TaskManifest[]
-  ) {
-    const tasksGroupedByFile: Record<string, TaskManifest[]> = {};
-
-    for (const task of tasks) {
-      if (!tasksGroupedByFile[task.filePath]) {
-        tasksGroupedByFile[task.filePath] = [];
-      }
-
-      tasksGroupedByFile[task.filePath]!.push(task);
-    }
-
-    const taskFiles: Array<{
-      taskIds: string[];
-      contents: string;
-      contentHash: string;
-      filePath: string;
-    }> = [];
-
-    for (const [filePath, tasks] of Object.entries(tasksGroupedByFile)) {
-      const source = sources[filePath];
-
-      if (!source) {
-        continue;
-      }
-
-      const taskIds = tasks.map((task) => task.id);
-
-      taskFiles.push({
-        ...source,
-        taskIds,
-        filePath,
-      });
-    }
-
-    return taskFiles;
-  }
-
   async #getEnvVars(): Promise<Record<string, string>> {
     const environmentVariablesResponse = await this.options.client.getEnvironmentVariables(
       this.options.config.project

packages/cli-v3/src/entryPoints/deploy-executor.ts

@@ -360,7 +360,7 @@ runtime.setGlobalRuntimeManager(prodRuntimeManager);
 
 process.title = "trigger-dev-worker";
 
-for await (const _ of setInterval(15)) {
+for await (const _ of setInterval(15_000)) {
   if (_isRunning && _execution) {
     try {
       await zodIpc.send("TASK_HEARTBEAT", { id: _execution.attempt.id });

packages/cli-v3/src/entryPoints/deploy.ts

@@ -734,6 +734,7 @@ class ProdWorker {
             env,
             serverWorker: execution.worker,
             payload: createAttempt.result.executionPayload,
+            messageId: message.lazyPayload.messageId,
           });
 
           this._taskRunProcess.onTaskRunHeartbeat.attach((heartbeatId) => {

packages/cli-v3/src/executions/taskRunProcess.ts

@@ -44,9 +44,9 @@ export type TaskRunProcessOptions = {
   serverWorker: ServerBackgroundWorker;
   env: Record<string, string>;
   payload: TaskRunExecutionPayload;
+  messageId: string;
 
   cwd?: string;
-  messageId?: string;
 };
 
 export class TaskRunProcess {
@@ -64,10 +64,7 @@ export class TaskRunProcess {
   private _isBeingCancelled: boolean = false;
   private _stderr: Array<string> = [];
   private _flushingProcess?: FlushingProcess;
-  /**
-   * @deprecated use onTaskRunHeartbeat instead
-   */
-  public onTaskHeartbeat: Evt<string> = new Evt();
+
   public onTaskRunHeartbeat: Evt<string> = new Evt();
   public onExit: Evt<{ code: number | null; signal: NodeJS.Signals | null; pid?: number }> =
     new Evt();
@@ -171,15 +168,7 @@ export class TaskRunProcess {
           this.onReadyToDispose.post(this);
         },
         TASK_HEARTBEAT: async (message) => {
-          if (messageId) {
-            this.onTaskRunHeartbeat.post(messageId);
-          } else {
-            logger.debug(
-              "No message id for task heartbeat, falling back to (deprecated) attempt heartbeat",
-              { id: message.id }
-            );
-            this.onTaskHeartbeat.post(message.id);
-          }
+          this.onTaskRunHeartbeat.post(messageId);
         },
         WAIT_FOR_TASK: async (message) => {
           this.onWaitForTask.post(message);

references/v3-catalog/src/trigger/longRunning.ts

@@ -5,11 +5,11 @@ export const longRunning = task({
   run: async (payload: { message: string }, { ctx }) => {
     logger.info("Long running", { payload });
 
-    await new Promise((resolve) => setTimeout(resolve, 20000));
+    await new Promise((resolve) => setTimeout(resolve, 200000)); // 200 seconds
 
     await wait.for({ seconds: 10 });
 
-    await new Promise((resolve) => setTimeout(resolve, 20000));
+    await new Promise((resolve) => setTimeout(resolve, 200000)); // 200 seconds
   },
 });