feat(supervisor): release a run whose Runner create the API refused

With `RELEASE_UNSTARTED_RUNS_ENABLED=true`, the supervisor now also gives back a run whose Runner the cluster refused to create, when the run-crd workload manager is on, so the run is requeued at once instead of waiting out the pending execution timeout. Off by default, nothing changes.

Mono-RevId: ecf6db0d6700593ba3d2b644f74ae8c19c15c37d

Author: myftija

apps/supervisor/src/env.ts

@@ -363,8 +363,8 @@ export const Env = z
     // Report a run whose runner the runtime killed, so it fails as out of memory only
     // when the runtime says so. Without it the run waits out the heartbeat timeout.
     RUNNER_KILL_REPORTS_ENABLED: BoolEnv.default(true),
-    // Release a run whose container or pod could not be created, so it is requeued at once instead
-    // of waiting out the pending execution timeout.
+    // Release a run whose container, pod or Runner could not be created, so it is requeued at once
+    // instead of waiting out the pending execution timeout.
     RELEASE_UNSTARTED_RUNS_ENABLED: BoolEnv.default(false),
     // Release a run whose pod failed before the run started, likewise. Independent of the flag above.
     RELEASE_FAILED_POD_RUNS_ENABLED: BoolEnv.default(false),

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

@@ -28,11 +28,11 @@ function socketError(code: string, message: string) {
 }
 
 describe("k8sCreateRefused", () => {
-  it.each([401, 403, 422])("is true for a %i", (code) => {
+  it.each([400, 401, 403, 404, 422])("is true for a %i", (code) => {
     expect(k8sCreateRefused(apiError(code, "refused"))).toBe(true);
   });
 
-  it.each([408, 409, 500, 504])("is false for a %i, after which a pod may exist", (code) => {
+  it.each([408, 409, 500, 504])("is false for a %i, after which the object may exist", (code) => {
     expect(k8sCreateRefused(apiError(code, "maybe created"))).toBe(false);
   });
 

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

@@ -30,6 +30,7 @@ import {
   type Watch,
 } from "@kubernetes/client-node";
 import type { K8sApi } from "../clients/kubernetes.js";
+import { WorkloadNotCreatedError } from "./createErrors.js";
 import { getRestoreRunnerId, getRunnerId } from "../util.js";
 import type { WorkloadManagerCreateOptions } from "./types.js";
 
@@ -574,12 +575,13 @@ describe("run-crd carries every shared create-option or excludes it on purpose",
 });
 
 describe("RunCrdWorkloadManager.create", () => {
-  const manager = () =>
+  const manager = (releaseUnstartedRuns?: boolean) =>
     new RunCrdWorkloadManager({
       workloadApiProtocol: "http",
       workloadApiPort: 8020,
       namespace: "v4-runs",
       runtime: "microvm",
+      releaseUnstartedRuns,
     });
 
   beforeEach(() => {
@@ -588,6 +590,12 @@ describe("RunCrdWorkloadManager.create", () => {
     getRunner.mockReset();
     deleteRunner.mockReset();
     deleteRunner.mockResolvedValue({});
+    createSecret.mockReset();
+    createSecret.mockResolvedValue({ metadata: { uid: "uid-secret" } });
+    deleteSecret.mockReset();
+    deleteSecret.mockResolvedValue({});
+    patchObject.mockReset();
+    patchObject.mockResolvedValue({});
   });
 
   // keyedOptions() is dequeued at 03:00:00, so the default is an earlier delivery.
@@ -641,9 +649,6 @@ describe("RunCrdWorkloadManager.create", () => {
       .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(keyedOptions({ deploymentToken: "tok" }));
 
@@ -706,17 +711,130 @@ describe("RunCrdWorkloadManager.create", () => {
     expect(deleteRunner).not.toHaveBeenCalled();
     expect(createRunner).toHaveBeenCalledTimes(2);
   });
+
+  const apiError = (code: number, message = "refused") => ({
+    code,
+    body: JSON.stringify({ kind: "Status", code, message }),
+  });
+  const withToken = keyedOptions({ deploymentToken: "tok" });
+
+  it.each([
+    ["off", false],
+    ["unset", undefined],
+  ])("rethrows a refused Runner create as it is with the option %s", async (_, flag) => {
+    const failure = apiError(422);
+    createRunner.mockRejectedValueOnce(failure);
+
+    await expect(manager(flag).create(keyedOptions())).rejects.toBe(failure);
+    expect(createRunner).toHaveBeenCalledTimes(1);
+  });
+
+  it("rethrows a refused token Secret create as it is with the option off", async () => {
+    const failure = apiError(403);
+    createSecret.mockRejectedValueOnce(failure);
+
+    await expect(manager(false).create(withToken)).rejects.toBe(failure);
+    expect(createRunner).not.toHaveBeenCalled();
+  });
+
+  describe("with releases on", () => {
+    it("reports a Runner create the API refused as not created", async () => {
+      createRunner.mockRejectedValueOnce(apiError(422, "spec.machine: Unsupported value"));
+
+      const failed = manager(true).create(keyedOptions());
+
+      await expect(failed).rejects.toBeInstanceOf(WorkloadNotCreatedError);
+      await expect(failed).rejects.toMatchObject({ message: "spec.machine: Unsupported value" });
+      expect(createRunner).toHaveBeenCalledTimes(1);
+      expect(getRunner).not.toHaveBeenCalled();
+    });
+
+    it("releases the token Secret of a refused create", async () => {
+      createRunner.mockRejectedValueOnce(apiError(422));
+
+      await expect(manager(true).create(withToken)).rejects.toThrow(WorkloadNotCreatedError);
+
+      expect(createSecret).toHaveBeenCalledTimes(1);
+      expect(deleteSecret).toHaveBeenCalledTimes(1);
+    });
+
+    it("reports a refused token Secret create as not created, with nothing to delete", async () => {
+      createSecret.mockRejectedValueOnce(apiError(403, "secrets is forbidden"));
+
+      const failed = manager(true).create(withToken);
+
+      await expect(failed).rejects.toBeInstanceOf(WorkloadNotCreatedError);
+      await expect(failed).rejects.toMatchObject({ message: "secrets is forbidden" });
+      expect(createRunner).not.toHaveBeenCalled();
+      expect(deleteSecret).not.toHaveBeenCalled();
+    });
+
+    it("keeps the token Secret a redrive found, which it did not create", async () => {
+      createSecret.mockRejectedValueOnce({ code: 409 });
+      createRunner.mockRejectedValueOnce(apiError(422));
+
+      await expect(manager(true).create(withToken)).rejects.toThrow(WorkloadNotCreatedError);
+
+      expect(deleteSecret).not.toHaveBeenCalled();
+    });
+
+    it("rethrows a Runner create that failed with 500 as it is", async () => {
+      const failure = apiError(500);
+      createRunner.mockRejectedValueOnce(failure);
+
+      await expect(manager(true).create(keyedOptions())).rejects.toBe(failure);
+      expect(createRunner).toHaveBeenCalledTimes(1);
+    });
+
+    it("rethrows a token Secret create that failed with 500 as it is", async () => {
+      const failure = apiError(500);
+      createSecret.mockRejectedValueOnce(failure);
+
+      await expect(manager(true).create(withToken)).rejects.toBe(failure);
+      expect(createRunner).not.toHaveBeenCalled();
+    });
+
+    it("reports a refused replacement as not created, with both attempts' token Secrets deleted", async () => {
+      createRunner.mockRejectedValueOnce({ code: 409 }).mockRejectedValueOnce(apiError(403));
+      createSecret
+        .mockResolvedValueOnce({ metadata: { uid: "uid-secret-1" } })
+        .mockResolvedValueOnce({ metadata: { uid: "uid-secret-2" } });
+      getRunner.mockResolvedValue(inTheWay("snapshot_earlier"));
+
+      await expect(manager(true).create(withToken)).rejects.toThrow(WorkloadNotCreatedError);
+
+      expect(deleteRunner).toHaveBeenCalledTimes(1);
+      expect(createRunner).toHaveBeenCalledTimes(2);
+      const [first, second] = createSecret.mock.calls.map(([{ body }]) => body.metadata.name);
+      const deleted = deleteSecret.mock.calls.map(([{ name, body }]) => [
+        name,
+        body.preconditions.uid,
+      ]);
+      expect(deleted).toEqual([
+        [first, "uid-secret-1"],
+        [second, "uid-secret-2"],
+      ]);
+    });
+
+    it("does not wrap a template key refusal", async () => {
+      const failed = manager(true).create(createOptions({ templateKey: undefined }));
+
+      await expect(failed).rejects.toThrow("TemplateKeyMissing");
+      await expect(failed).rejects.not.toBeInstanceOf(WorkloadNotCreatedError);
+    });
+  });
 });
 
 describe("RunCrdWorkloadManager.restore", () => {
   const checkpoint = { id: "checkpoint_abc", location: "node-a/6f1c2a9e-snap" };
 
-  function manager(runtime: "container" | "microvm" = "microvm") {
+  function manager(runtime: "container" | "microvm" = "microvm", releaseUnstartedRuns?: boolean) {
     return new RunCrdWorkloadManager({
       workloadApiProtocol: "http",
       workloadApiPort: 8020,
       namespace: "v4-runs",
       runtime,
+      releaseUnstartedRuns,
     });
   }
 
@@ -753,6 +871,14 @@ describe("RunCrdWorkloadManager.restore", () => {
     expect(body.spec.restore).toEqual({ snapshotID: "6f1c2a9e-snap", node: "node-a" });
   });
 
+  it("reports a refused resume create as not created, with releases on", async () => {
+    createRunner.mockRejectedValueOnce({ code: 422 });
+
+    await expect(
+      manager("microvm", true).restore(keyedOptions(), checkpoint)
+    ).rejects.toBeInstanceOf(WorkloadNotCreatedError);
+  });
+
   it("creates no resume without a template key", async () => {
     await expect(manager().restore(createOptions(), checkpoint)).rejects.toThrow(
       "TemplateKeyMissing"

apps/supervisor/src/workloadManager/runCrd.ts

@@ -25,6 +25,7 @@ import type {
   WorkloadManagerOptions,
 } from "./types.js";
 import { killCode, type RunnerKill } from "../services/runnerKills.js";
+import { k8sCreateRefused, k8sErrorMessage, WorkloadNotCreatedError } from "./createErrors.js";
 
 const GROUP = "compute.trigger.dev";
 const VERSION = "v1alpha1";
@@ -115,11 +116,13 @@ export class RunCrdWorkloadManager implements WorkloadManager, RunnerSnapshotter
   private readonly suspendTimeoutMs: number;
   private readonly restoreTimeoutMs: number;
   private readonly restoreInformer?: RunnerRestoreInformer;
+  private readonly releaseUnstartedRuns: boolean;
 
   constructor(opts: RunCrdWorkloadManagerOptions) {
     this.k8s = createK8sApi();
     this.namespace = opts.namespace;
     this.runtime = opts.runtime;
+    this.releaseUnstartedRuns = opts.releaseUnstartedRuns ?? false;
     this.snapshots = opts.snapshots;
     this.suspendPollMs = opts.suspendPollMs ?? 5_000;
     this.suspendTimeoutMs = opts.suspendTimeoutMs ?? 6 * 60_000;
@@ -557,6 +560,10 @@ export class RunCrdWorkloadManager implements WorkloadManager, RunnerSnapshotter
       if (statusCodeOf(err) === 409) {
         return false;
       }
+      if (this.releaseUnstartedRuns && k8sCreateRefused(err)) {
+        this.logger.error("[RunCrdWorkloadManager] Create refused", { runnerId, rawError: err });
+        throw new WorkloadNotCreatedError(k8sErrorMessage(err), { cause: err });
+      }
       this.logger.error("[RunCrdWorkloadManager] Create failed", { runnerId, rawError: err });
       throw err;
     }
@@ -698,6 +705,9 @@ export class RunCrdWorkloadManager implements WorkloadManager, RunnerSnapshotter
         name,
         rawError: err,
       });
+      if (this.releaseUnstartedRuns && k8sCreateRefused(err)) {
+        throw new WorkloadNotCreatedError(k8sErrorMessage(err), { cause: err });
+      }
       throw err;
     }
   }