fix(supervisor): source-aware backpressure config, 5s scrape interval, parser hardening

Author: nicktrn

apps/supervisor/src/backpressure/k8sPodCountSignalSource.test.ts

@@ -24,6 +24,16 @@ describe("parsePodCount", () => {
 
   it("throws when the pods metric is absent", () => {
     const text = 'apiserver_storage_objects{resource="configmaps"} 17';
+    expect(() => parsePodCount(text)).toThrow(/not found/);
+  });
+
+  it("throws on a non-finite value (e.g. 1e999)", () => {
+    const text = 'apiserver_storage_objects{resource="pods"} 1e999';
+    expect(() => parsePodCount(text)).toThrow();
+  });
+
+  it("throws on a negative value", () => {
+    const text = 'apiserver_storage_objects{resource="pods"} -5';
     expect(() => parsePodCount(text)).toThrow();
   });
 });

apps/supervisor/src/backpressure/k8sPodCountSignalSource.ts

@@ -1,7 +1,7 @@
 import type { BackpressureSignalSource, BackpressureVerdict } from "./backpressureMonitor.js";
 
 // Reads the apiserver's stored-pod-object count from a Prometheus /metrics scrape.
-const POD_COUNT_RE = /^apiserver_storage_objects\{[^}]*resource="pods"[^}]*\}\s+([0-9.eE+-]+)/m;
+const POD_COUNT_RE = /^apiserver_storage_objects\{[^}]*resource="pods"[^}]*\}\s+([0-9.eE+]+)/m;
 
 export function parsePodCount(metricsText: string): number {
   const match = metricsText.match(POD_COUNT_RE);

apps/supervisor/src/clients/kubernetes.ts

@@ -68,6 +68,7 @@ export function createApiserverMetricsFetcher(): () => Promise<string> {
       throw new Error("no current cluster in kubeconfig");
     }
     const requestInit = await kubeConfig.applyToFetchOptions({ method: "GET" });
+    // node-fetch vs DOM RequestInit: structurally compatible, declaration-only mismatch
     const response = await fetch(`${cluster.server}/metrics`, requestInit as unknown as RequestInit);
     if (!response.ok) {
       throw new Error(`apiserver /metrics scrape failed: ${response.status}`);

apps/supervisor/src/env.test.ts

@@ -0,0 +1,44 @@
+import { describe, it, expect, vi } from "vitest";
+
+// Mock std-env before importing env.ts so the module-level `Env.parse(stdEnv)`
+// doesn't fail in a test environment that lacks required vars.
+vi.mock("std-env", () => ({
+  env: {
+    TRIGGER_API_URL: "http://localhost:3030",
+    TRIGGER_WORKER_TOKEN: "test-token",
+    MANAGED_WORKER_SECRET: "test-secret",
+    OTEL_EXPORTER_OTLP_ENDPOINT: "http://localhost:4318",
+  },
+}));
+
+const { Env } = await import("./env.js");
+
+// Minimal env that satisfies all required fields; everything else has defaults.
+const base = {
+  TRIGGER_API_URL: "http://localhost:3030",
+  TRIGGER_WORKER_TOKEN: "test-token",
+  MANAGED_WORKER_SECRET: "test-secret",
+  OTEL_EXPORTER_OTLP_ENDPOINT: "http://localhost:4318",
+};
+
+describe("Env superRefine - backpressure source awareness", () => {
+  it("accepts k8s-pod-count source without a Redis host", () => {
+    expect(() =>
+      Env.parse({
+        ...base,
+        TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED: "true",
+        TRIGGER_DEQUEUE_BACKPRESSURE_SOURCE: "k8s-pod-count",
+      })
+    ).not.toThrow();
+  });
+
+  it("rejects redis source when Redis host is absent", () => {
+    expect(() =>
+      Env.parse({
+        ...base,
+        TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED: "true",
+        TRIGGER_DEQUEUE_BACKPRESSURE_SOURCE: "redis",
+      })
+    ).toThrow();
+  });
+});

apps/supervisor/src/env.ts

@@ -3,7 +3,7 @@ import { env as stdEnv } from "std-env";
 import { z } from "zod";
 import { AdditionalEnvVars, BoolEnv } from "./envUtil.js";
 
-const Env = z
+export const Env = z
   .object({
     // This will come from `spec.nodeName` in k8s
     TRIGGER_WORKER_INSTANCE_NAME: z.string().default(randomUUID()),
@@ -79,6 +79,7 @@ const Env = z
     TRIGGER_DEQUEUE_BACKPRESSURE_SOURCE: z.enum(["redis", "k8s-pod-count"]).default("redis"),
     TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENGAGE: z.coerce.number().int().positive().default(10_000),
     TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_RELEASE: z.coerce.number().int().positive().default(5_000),
+    TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_REFRESH_MS: z.coerce.number().int().positive().default(5_000),
 
     // Optional services
     TRIGGER_WARM_START_URL: z.string().optional(),
@@ -332,7 +333,11 @@ const Env = z
         path: ["TRIGGER_WORKLOAD_API_DOMAIN"],
       });
     }
-    if (data.TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED && !data.TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST) {
+    if (
+      data.TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED &&
+      data.TRIGGER_DEQUEUE_BACKPRESSURE_SOURCE === "redis" &&
+      !data.TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST
+    ) {
       ctx.addIssue({
         code: z.ZodIssueCode.custom,
         message:

apps/supervisor/src/index.ts

@@ -219,8 +219,10 @@ class ManagedSupervisor {
 
     if (env.TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED) {
       let source: BackpressureSignalSource;
+      let refreshIntervalMs = env.TRIGGER_DEQUEUE_BACKPRESSURE_REFRESH_MS;
 
       if (env.TRIGGER_DEQUEUE_BACKPRESSURE_SOURCE === "k8s-pod-count") {
+        refreshIntervalMs = env.TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_REFRESH_MS;
         if (
           env.TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_RELEASE >=
           env.TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENGAGE
@@ -243,7 +245,7 @@ class ManagedSupervisor {
         this.logger.log("🛑 Dequeue backpressure enabled (pod-count source)", {
           engage: env.TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENGAGE,
           release: env.TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_RELEASE,
-          refreshIntervalMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_REFRESH_MS,
+          refreshIntervalMs,
           dryRun: env.TRIGGER_DEQUEUE_BACKPRESSURE_DRY_RUN,
         });
       } else {
@@ -274,7 +276,7 @@ class ManagedSupervisor {
       this.backpressureMonitor = new BackpressureMonitor({
         enabled: true,
         source,
-        refreshIntervalMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_REFRESH_MS,
+        refreshIntervalMs,
         maxVerdictAgeMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_MAX_VERDICT_AGE_MS,
         rampMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_RAMP_MS,
         dryRun: env.TRIGGER_DEQUEUE_BACKPRESSURE_DRY_RUN,