fix(supervisor): replace a requeued cold start's Runner instead of failing every redelivery on it

Supervisor: when a run is requeued after its Runner never started, replace the stale Runner instead of retrying into a name conflict until it expires.
Mono-RevId: e402081626e486d4ac4b25b173e65fe2d477802f

Author: 1stvamp

apps/supervisor/src/workloadManager/runCrd.test.ts

@@ -516,6 +516,126 @@ describe("run-crd carries every shared create-option or excludes it on purpose",
   });
 });
 
+describe("RunCrdWorkloadManager.create", () => {
+  const manager = () =>
+    new RunCrdWorkloadManager({
+      workloadApiProtocol: "http",
+      workloadApiPort: 8020,
+      namespace: "v4-runs",
+      runtime: "microvm",
+    });
+
+  beforeEach(() => {
+    createRunner.mockReset();
+    createRunner.mockResolvedValue({ metadata: { uid: "uid-created" } });
+    getRunner.mockReset();
+    deleteRunner.mockReset();
+    deleteRunner.mockResolvedValue({});
+  });
+
+  // createOptions() is dequeued at 03:00:00, so the default is an earlier delivery.
+  const inTheWay = (snapshotFriendlyID: string, dequeuedAt?: string) => ({
+    metadata: { name: "runner-abc123", uid: "uid-existing" },
+    spec: {
+      bootstrap: {
+        runFriendlyID: "run_abc123",
+        snapshotFriendlyID,
+        dequeuedAt: dequeuedAt ?? "2026-08-27T02:59:00.000Z",
+      },
+    },
+  });
+
+  // The platform requeued an earlier delivery that never started the attempt.
+  it("replaces a Runner an earlier delivery left, guarded by its uid", async () => {
+    createRunner
+      .mockRejectedValueOnce({ code: 409 })
+      .mockResolvedValueOnce({ metadata: { uid: "uid-recreated" } });
+    getRunner.mockResolvedValue(inTheWay("snapshot_earlier"));
+
+    await manager().create(createOptions());
+
+    expect(deleteRunner).toHaveBeenCalledWith(
+      expect.objectContaining({
+        name: "runner-abc123",
+        body: { preconditions: { uid: "uid-existing" } },
+      })
+    );
+    expect(createRunner).toHaveBeenCalledTimes(2);
+  });
+
+  // The first create found the old Runner's Secret, which the collector takes with it.
+  it("gives the replacement a token Secret of its own", async () => {
+    createRunner
+      .mockRejectedValueOnce({ code: 409 })
+      .mockResolvedValueOnce({ metadata: { uid: "uid-recreated" } });
+    getRunner.mockResolvedValue(inTheWay("snapshot_earlier"));
+    createSecret.mockReset();
+    createSecret.mockResolvedValue({ metadata: { uid: "uid-secret" } });
+    patchObject.mockResolvedValue({});
+
+    await manager().create(createOptions({ deploymentToken: "tok" }));
+
+    const [first, second] = createRunner.mock.calls.map(
+      ([{ body }]) => body.spec.deployment.token.name
+    );
+    expect(second).not.toBe(first);
+  });
+
+  it("leaves a Runner this delivery already made", async () => {
+    createRunner.mockRejectedValueOnce({ code: 409 });
+    getRunner.mockResolvedValue(inTheWay("snapshot_abc"));
+
+    await manager().create(createOptions());
+
+    expect(deleteRunner).not.toHaveBeenCalled();
+    expect(createRunner).toHaveBeenCalledTimes(1);
+  });
+
+  // A delivery held up past the requeue must not remove the one that followed it.
+  it.each([
+    ["a later delivery", "2026-08-27T03:01:00.000Z"],
+    ["a delivery dequeued at the same time", "2026-08-27T03:00:00.000Z"],
+    ["a delivery of unknown order", ""],
+  ])("leaves a Runner from %s and fails as stale", async (_, dequeuedAt) => {
+    createRunner.mockRejectedValueOnce({ code: 409 });
+    getRunner.mockResolvedValue(inTheWay("snapshot_later", dequeuedAt));
+
+    await expect(manager().create(createOptions())).rejects.toThrow("stale");
+    expect(deleteRunner).not.toHaveBeenCalled();
+    expect(createRunner).toHaveBeenCalledTimes(1);
+  });
+
+  // Another delivery of this snapshot made the replacement in the gap.
+  it("accepts a replacement for this snapshot that another delivery made first", async () => {
+    createRunner.mockRejectedValue({ code: 409 });
+    getRunner
+      .mockResolvedValueOnce(inTheWay("snapshot_earlier"))
+      .mockResolvedValueOnce(inTheWay("snapshot_abc", "2026-08-27T03:00:00.000Z"));
+
+    await manager().create(createOptions());
+
+    expect(createRunner).toHaveBeenCalledTimes(2);
+  });
+
+  it("replaces it only once per delivery", async () => {
+    createRunner.mockRejectedValue({ code: 409 });
+    getRunner.mockResolvedValue(inTheWay("snapshot_earlier"));
+
+    await expect(manager().create(createOptions())).rejects.toThrow("still in the way");
+    expect(createRunner).toHaveBeenCalledTimes(2);
+  });
+
+  it("creates again when the Runner went between the create and the read", async () => {
+    createRunner.mockRejectedValueOnce({ code: 409 });
+    getRunner.mockRejectedValue({ code: 404 });
+
+    await manager().create(createOptions());
+
+    expect(deleteRunner).not.toHaveBeenCalled();
+    expect(createRunner).toHaveBeenCalledTimes(2);
+  });
+});
+
 describe("RunCrdWorkloadManager.restore", () => {
   const checkpoint = { id: "checkpoint_abc", location: "node-a/6f1c2a9e-snap" };
 
@@ -709,12 +829,6 @@ describe("RunCrdWorkloadManager.restore", () => {
     expect(createRunner).toHaveBeenCalledTimes(2);
   });
 
-  it("still fails a cold start that finds a Runner in the way", async () => {
-    createRunner.mockRejectedValue({ code: 409 });
-
-    await expect(manager().create(createOptions())).rejects.toEqual({ code: 409 });
-  });
-
   it("deletes a failed resume only while it is the Runner the watch saw", async () => {
     deleteRunner.mockReset();
     deleteRunner.mockResolvedValue({});

apps/supervisor/src/workloadManager/runCrd.ts

@@ -339,11 +339,16 @@ export class RunCrdWorkloadManager implements WorkloadManager, RunnerSnapshotter
     return outcome ? { ...outcome, createdAt: createdAtOf(runner), uid: uidOf(runner) } : timedOut;
   }
 
+  /** Deletes a failed resume so its redelivery can create it again. */
+  async deleteRestoreRunner(runnerId: string, uid: string): Promise<void> {
+    await this.deleteRunner(runnerId, uid);
+  }
+
   /**
-   * Deletes a failed resume so its redelivery can create it again. Preconditioned
-   * on uid, so a Runner that already replaced it is left alone; one already gone is fine.
+   * Preconditioned on uid, so a Runner that already replaced this one is left
+   * alone; one already gone is fine.
    */
-  async deleteRestoreRunner(runnerId: string, uid: string): Promise<void> {
+  private async deleteRunner(runnerId: string, uid: string): Promise<void> {
     try {
       await this.k8s.custom.deleteNamespacedCustomObject({
         group: GROUP,
@@ -378,8 +383,54 @@ export class RunCrdWorkloadManager implements WorkloadManager, RunnerSnapshotter
     return this.runtime === "microvm" && checkpoint.type === "COMPUTE";
   }
 
+  /**
+   * A cold start's Runner is named for its attempt, so one already in the way is
+   * another delivery of this attempt. One for this snapshot is this delivery's
+   * own create, made again. One dequeued before this delivery is one the
+   * platform has since requeued (it never started the attempt), which can't
+   * start it now, so it is replaced, once and guarded by its uid. Left in place,
+   * it failed every redelivery's create until the operator's terminal TTL took
+   * it. One dequeued after this delivery is newer, and this delivery is the
+   * stale one.
+   */
   async create(opts: WorkloadManagerCreateOptions) {
-    await this.createRunner(opts, getRunnerId(opts.runFriendlyId, opts.nextAttemptNumber));
+    const runnerId = getRunnerId(opts.runFriendlyId, opts.nextAttemptNumber);
+    if (await this.createRunner(opts, runnerId)) {
+      return;
+    }
+    try {
+      const existing = (await this.getRunner(runnerId)) as RunnerBootstrapSpec | null;
+      const bootstrap = existing?.spec?.bootstrap;
+      if (bootstrap?.snapshotFriendlyID === opts.snapshotFriendlyId) {
+        return;
+      }
+      // Unknown order counts as newer: only a Runner shown to be older is deleted.
+      if (!(Date.parse(bootstrap?.dequeuedAt ?? "") < opts.dequeuedAt.getTime())) {
+        throw new Error(
+          `Runner ${runnerId} belongs to a delivery dequeued after this one, which is stale`
+        );
+      }
+      const uid = uidOf(existing);
+      if (!uid) {
+        throw new Error(`Runner ${runnerId} is in the way, with no uid to guard its delete`);
+      }
+      await this.deleteRunner(runnerId, uid);
+    } catch (err: unknown) {
+      // Deleted between the create and the read.
+      if (statusCodeOf(err) !== 404) {
+        throw err;
+      }
+    }
+    // A Secret of its own: the one this create found is the deleted Runner's, and
+    // the collector takes it.
+    if (await this.createRunner(opts, runnerId, undefined, true)) {
+      return;
+    }
+    // Another delivery of this snapshot may have made it in the gap.
+    const raced = (await this.getRunner(runnerId)) as RunnerBootstrapSpec | null;
+    if (raced?.spec?.bootstrap?.snapshotFriendlyID !== opts.snapshotFriendlyId) {
+      throw new Error(`Runner ${runnerId} is still in the way after replacing it`);
+    }
   }
 
   /**
@@ -462,13 +513,14 @@ export class RunCrdWorkloadManager implements WorkloadManager, RunnerSnapshotter
     }
   }
 
-  /** Returns false when a resume's Runner was already there. */
+  /** Returns false when the Runner was already there. */
   private async createRunner(
     opts: WorkloadManagerCreateOptions,
     runnerId: string,
-    restore?: RunnerRestore
+    restore?: RunnerRestore,
+    replacing = false
   ): Promise<{ uid?: string } | false> {
-    const token = await this.ensureRunnerToken(opts, runnerId, !!restore);
+    const token = await this.ensureRunnerToken(opts, runnerId, !!restore || replacing);
 
     const body = runnerBodyFor(opts, {
       name: runnerId,
@@ -495,8 +547,8 @@ export class RunCrdWorkloadManager implements WorkloadManager, RunnerSnapshotter
       await this.releaseRunnerToken(token, err);
       // A resume's name carries the checkpoint, so the Runner in the way is this
       // resume, still restoring or held terminal for the operator's TTL. A cold
-      // start's carries the attempt, so its 409 is an earlier terminal Runner.
-      if (restore && statusCodeOf(err) === 409) {
+      // start's carries the attempt: create sorts out which delivery made it.
+      if (statusCodeOf(err) === 409) {
         return false;
       }
       this.logger.error("[RunCrdWorkloadManager] Create failed", { runnerId, rawError: err });
@@ -971,7 +1023,9 @@ export type UnwatchedRestoreFailure = {
 };
 
 type RunnerBootstrapSpec = {
-  spec?: { bootstrap?: { runFriendlyID?: string; snapshotFriendlyID?: string } };
+  spec?: {
+    bootstrap?: { runFriendlyID?: string; snapshotFriendlyID?: string; dequeuedAt?: string };
+  };
 };
 
 /**