improve prod worker logging

Author: nicktrn

packages/cli-v3/src/workers/prod/backgroundWorker.ts

@@ -90,6 +90,8 @@ export class ProdBackgroundWorker {
   ) {}
 
   async close(gracefulExitTimeoutElapsed = false) {
+    console.log("Closing worker", { gracefulExitTimeoutElapsed, closed: this._closed });
+
     if (this._closed) {
       return;
     }
@@ -104,6 +106,8 @@ export class ProdBackgroundWorker {
   }
 
   async #killTaskRunProcess(flush = true, initialSignal: number | NodeJS.Signals = "SIGTERM") {
+    console.log("Killing task run process", { flush, initialSignal, closed: this._closed });
+
     if (this._closed || !this._taskRunProcess) {
       return;
     }
@@ -119,6 +123,10 @@ export class ProdBackgroundWorker {
       console.error("Error while trying graceful exit", error);
     });
 
+    console.log("Killed task run process, setting closed to true", {
+      closed: this._closed,
+      pid: currentTaskRunProcess.pid,
+    });
     this._closed = true;
   }
 
@@ -232,6 +240,9 @@ export class ProdBackgroundWorker {
       payload.execution.worker.version
     );
 
+    console.log("Getting fresh task run process, setting closed to false", {
+      closed: this._closed,
+    });
     this._closed = false;
 
     await this.#killCurrentTaskRunProcessBeforeAttempt();
@@ -250,6 +261,8 @@ export class ProdBackgroundWorker {
     );
 
     taskRunProcess.onExit.attach(({ pid }) => {
+      console.log("Task run process exited", { pid });
+
       // Only delete the task run process if the pid matches
       if (this._taskRunProcess?.pid === pid) {
         this._taskRunProcess = undefined;
@@ -320,12 +333,21 @@ export class ProdBackgroundWorker {
   }
 
   async #killCurrentTaskRunProcessBeforeAttempt() {
+    console.log("killCurrentTaskRunProcessBeforeAttempt()", {
+      hasTaskRunProcess: !!this._taskRunProcess,
+    });
+
     if (!this._taskRunProcess) {
       return;
     }
 
     const currentTaskRunProcess = this._taskRunProcess;
 
+    console.log("Killing current task run process", {
+      isBeingKilled: currentTaskRunProcess?.isBeingKilled,
+      totalBeingKilled: this._taskRunProcessesBeingKilled.size,
+    });
+
     if (currentTaskRunProcess.isBeingKilled) {
       if (this._taskRunProcessesBeingKilled.size > 1) {
         await this.#tryGracefulExit(currentTaskRunProcess);
@@ -382,6 +404,11 @@ export class ProdBackgroundWorker {
     try {
       const taskRunProcess = await this.#getFreshTaskRunProcess(payload, messageId);
 
+      console.log("executing task run", {
+        attempt: payload.execution.attempt.id,
+        taskRunPid: taskRunProcess.pid,
+      });
+
       const result = await taskRunProcess.executeTaskRun(payload);
 
       if (result.ok) {
@@ -477,7 +504,12 @@ export class ProdBackgroundWorker {
   }
 
   async cancelAttempt(attemptId: string) {
-    await this._taskRunProcess?.cancel();
+    if (!this._taskRunProcess) {
+      console.error("No task run process to cancel attempt", { attemptId });
+      return;
+    }
+
+    await this._taskRunProcess.cancel();
   }
 
   async executeTaskRunLazyAttempt(payload: TaskRunExecutionLazyAttemptPayload) {
@@ -701,6 +733,8 @@ class TaskRunProcess {
   }
 
   async cleanup(kill = false, gracefulExitTimeoutElapsed = false) {
+    console.log("cleanup()", { kill, gracefulExitTimeoutElapsed });
+
     if (kill && this._isBeingKilled) {
       return;
     }
@@ -715,6 +749,11 @@ class TaskRunProcess {
     // Kill parent unless graceful exit timeout has elapsed and we're in the middle of an execution
     const killParentProcess = kill && !killChildProcess;
 
+    console.log("Cleaning up task run process", {
+      killChildProcess,
+      killParentProcess,
+    });
+
     await this._ipc?.sendWithAck("CLEANUP", {
       flush: true,
       kill: killParentProcess,
@@ -780,9 +819,13 @@ class TaskRunProcess {
   }
 
   async #handleExit(code: number | null, signal: NodeJS.Signals | null) {
+    console.log("handling child exit", { code, signal });
+
     // Go through all the attempts currently pending and reject them
     for (const [id, status] of this._attemptStatuses.entries()) {
       if (status === "PENDING") {
+        console.log("found pending attempt", { id });
+
         this._attemptStatuses.set(id, "REJECTED");
 
         const attemptPromise = this._attemptPromises.get(id);

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

@@ -134,7 +134,9 @@ class ProdWorker {
         });
       }
     } catch (error) {
-      logger.error("taskinfo read error during reconnect", { error });
+      logger.error("taskinfo read error during reconnect", {
+        error: error instanceof Error ? error.message : error,
+      });
     } finally {
       this.#coordinatorSocket = this.#createCoordinatorSocket(coordinatorHost);
     }
@@ -181,6 +183,8 @@ class ProdWorker {
         }
       );
 
+      logger.log("onCancelCheckpoint coordinator response", { checkpointCanceled });
+
       if (checkpointCanceled) {
         if (message.reason === "WAIT_FOR_DURATION") {
           // Worker will resume immediately
@@ -385,11 +389,8 @@ class ProdWorker {
       extraHeaders["x-trigger-attempt-friendly-id"] = this.attemptFriendlyId;
     }
 
-    logger.log("connecting to coordinator", {
-      host,
-      port: COORDINATOR_PORT,
-      extraHeaders,
-    });
+    logger.log(`connecting to coordinator: ${host}:${COORDINATOR_PORT}`);
+    logger.debug(`connecting with extra headers`, { extraHeaders });
 
     const coordinatorConnection = new ZodSocketConnection({
       namespace: "prod-worker",
@@ -574,14 +575,16 @@ class ProdWorker {
         },
       },
       onConnection: async (socket, handler, sender, logger) => {
+        logger.log("connected to coordinator", { status: this.#status });
+
         if (this.waitForPostStart) {
           logger.log("skip connection handler, waiting for post start hook");
           return;
         }
 
         if (this.paused) {
           if (!this.nextResumeAfter) {
-            logger.error("Missing next resume reason");
+            logger.error("Missing next resume reason", { status: this.#status });
 
             this.#emitUnrecoverableError(
               "NoNextResume",
@@ -592,7 +595,7 @@ class ProdWorker {
           }
 
           if (!this.attemptFriendlyId) {
-            logger.error("Missing friendly ID");
+            logger.error("Missing friendly ID", { status: this.#status });
 
             this.#emitUnrecoverableError(
               "NoAttemptId",

packages/core/src/v3/schemas/messages.ts

@@ -856,6 +856,7 @@ export const ProdWorkerToCoordinatorMessages = {
   },
 };
 
+// TODO: The coordinator can only safely use v1 worker messages, higher versions will need a new flag, e.g. SUPPORTS_VERSIONED_MESSAGES
 export const CoordinatorToProdWorkerMessages = {
   RESUME_AFTER_DEPENDENCY: {
     message: z.object({