ensure no executions or child processes are being reused

Author: nicktrn

packages/cli-v3/src/entryPoints/managed/controller.ts

@@ -178,7 +178,17 @@ export class ManagedRunController {
     }
 
     const execution = async () => {
-      if (!this.currentExecution || !this.currentExecution.isPreparedForNextRun) {
+      // If we have an existing execution that isn't prepared for the next run, kill it
+      if (this.currentExecution && !this.currentExecution.canExecute) {
+        this.sendDebugLog({
+          runId: runFriendlyId,
+          message: "killing existing execution before starting new run",
+        });
+        await this.currentExecution.kill().catch(() => {});
+        this.currentExecution = null;
+      }
+
+      if (!this.currentExecution || !this.currentExecution.canExecute) {
         this.currentExecution = new RunExecution({
           workerManifest: this.workerManifest,
           env: this.env,
@@ -267,11 +277,12 @@ export class ManagedRunController {
       if (this.currentExecution?.taskRunEnv) {
         this.sendDebugLog({
           runId: this.runFriendlyId,
-          message: "waitForNextRun: eagerly recreating task run process",
+          message: "waitForNextRun: eagerly creating fresh execution for next run",
         });
 
         const previousTaskRunEnv = this.currentExecution.taskRunEnv;
 
+        // Create a fresh execution for the next run
         this.currentExecution = new RunExecution({
           workerManifest: this.workerManifest,
           env: this.env,

packages/cli-v3/src/entryPoints/managed/execution.ts

@@ -87,10 +87,6 @@ export class RunExecution {
       throw new Error("prepareForExecution called after process was already created");
     }
 
-    if (this.isPreparedForNextRun) {
-      throw new Error("prepareForExecution called after execution was already prepared");
-    }
-
     this.taskRunProcess = this.createTaskRunProcess({
       envVars: opts.taskRunEnv,
       isWarmStart: true,
@@ -151,9 +147,14 @@ export class RunExecution {
   }
 
   /**
-   * Returns true if the execution has been prepared with task run env.
+   * Returns true if no run has been started yet and the process is prepared for the next run.
    */
-  get isPreparedForNextRun(): boolean {
+  get canExecute(): boolean {
+    // If we've ever had a run ID, this execution can't be reused
+    if (this._runFriendlyId) {
+      return false;
+    }
+
     return !!this.taskRunProcess?.isPreparedForNextRun;
   }
 
@@ -609,8 +610,18 @@ export class RunExecution {
     metrics: TaskRunExecutionMetrics;
     isWarmStart?: boolean;
   }) {
+    // For immediate retries, we need to ensure the task run process is prepared for the next attempt
+    if (
+      this.runFriendlyId &&
+      this.taskRunProcess &&
+      !this.taskRunProcess.isPreparedForNextAttempt
+    ) {
+      this.sendDebugLog("killing existing task run process before executing next attempt");
+      await this.kill().catch(() => {});
+    }
+
     // To skip this step and eagerly create the task run process, run prepareForExecution first
-    if (!this.taskRunProcess || !this.isPreparedForNextRun) {
+    if (!this.taskRunProcess || !this.taskRunProcess.isPreparedForNextRun) {
       this.taskRunProcess = this.createTaskRunProcess({ envVars, isWarmStart });
     }
 
@@ -667,11 +678,15 @@ export class RunExecution {
   }
 
   public exit() {
-    if (this.isPreparedForNextRun) {
+    if (this.taskRunProcess?.isPreparedForNextRun) {
       this.taskRunProcess?.forceExit();
     }
   }
 
+  public async kill() {
+    await this.taskRunProcess?.kill("SIGKILL");
+  }
+
   private async complete({ completion }: { completion: TaskRunExecutionResult }): Promise<void> {
     if (!this.runFriendlyId || !this.currentSnapshotId) {
       throw new Error("Cannot complete run: missing run or snapshot ID");

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

@@ -89,15 +89,21 @@ export class TaskRunProcess {
   public onWait: Evt<OnWaitMessage> = new Evt();
 
   private _isPreparedForNextRun: boolean = false;
+  private _isPreparedForNextAttempt: boolean = false;
 
   constructor(public readonly options: TaskRunProcessOptions) {
     this._isPreparedForNextRun = true;
+    this._isPreparedForNextAttempt = true;
   }
 
   get isPreparedForNextRun() {
     return this._isPreparedForNextRun;
   }
 
+  get isPreparedForNextAttempt() {
+    return this._isPreparedForNextAttempt;
+  }
+
   async cancel() {
     this._isPreparedForNextRun = false;
     this._isBeingCancelled = true;
@@ -223,6 +229,7 @@ export class TaskRunProcess {
     isWarmStart?: boolean
   ): Promise<TaskRunExecutionResult> {
     this._isPreparedForNextRun = false;
+    this._isPreparedForNextAttempt = false;
 
     let resolver: (value: TaskRunExecutionResult) => void;
     let rejecter: (err?: any) => void;
@@ -266,6 +273,7 @@ export class TaskRunProcess {
     const result = await promise;
 
     this._currentExecution = undefined;
+    this._isPreparedForNextAttempt = true;
 
     return result;
   }