feat(supervisor): wire dequeue backpressure into preDequeue

Gate dequeues on the backpressure verdict via the existing preDequeue hook,
on all paths including k8s (where the resource monitor is a no-op). Construct
the Redis-backed monitor only when TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED is set;
require a Redis host when enabled. Off by default - no Redis client, no effect.

Author: nicktrn

.server-changes/supervisor-dequeue-backpressure.md

@@ -0,0 +1,6 @@
+---
+area: supervisor
+type: feature
+---
+
+Add opt-in dequeue backpressure to the supervisor. When enabled, the supervisor reads a verdict from Redis and pauses dequeuing while the worker cluster is saturated, then resumes once capacity is available. Disabled by default - no behavior change for existing deployments.

apps/supervisor/src/env.ts

@@ -47,6 +47,24 @@ const Env = z
     TRIGGER_DEQUEUE_SCALING_BATCH_WINDOW_MS: z.coerce.number().int().positive().default(1000), // Batch window for metrics processing (ms)
     TRIGGER_DEQUEUE_SCALING_DAMPING_FACTOR: z.coerce.number().min(0).max(1).default(0.7), // Smooths consumer count changes after EWMA (0=no scaling, 1=immediate)
 
+    // Dequeue backpressure - off by default. When enabled, the supervisor reads a
+    // verdict from Redis (written by the cluster-side aggregator) and pauses dequeues
+    // while the worker cluster can't schedule pods. Disabled = total no-op: no Redis
+    // client is created, no reads happen, and the dequeue loop is unaffected.
+    TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED: BoolEnv.default(false),
+    TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_KEY: z.string().default("engine:dequeue:backpressure"),
+    TRIGGER_DEQUEUE_BACKPRESSURE_REFRESH_MS: z.coerce.number().int().positive().default(1000),
+    TRIGGER_DEQUEUE_BACKPRESSURE_MAX_VERDICT_AGE_MS: z.coerce
+      .number()
+      .int()
+      .positive()
+      .default(15_000), // Stale verdict → fail-open (treat as not engaged)
+    TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST: z.string().optional(),
+    TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_PORT: z.coerce.number().int().optional(),
+    TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_USERNAME: z.string().optional(),
+    TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_PASSWORD: z.string().optional(),
+    TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_TLS_DISABLED: BoolEnv.default(false),
+
     // Optional services
     TRIGGER_WARM_START_URL: z.string().optional(),
     TRIGGER_CHECKPOINT_URL: z.string().optional(),
@@ -281,6 +299,14 @@ const Env = z
         path: ["TRIGGER_WORKLOAD_API_DOMAIN"],
       });
     }
+    if (data.TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED && !data.TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST) {
+      ctx.addIssue({
+        code: z.ZodIssueCode.custom,
+        message:
+          "TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST is required when TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED is true",
+        path: ["TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST"],
+      });
+    }
   })
   .transform((data) => ({
     ...data,

apps/supervisor/src/index.ts

@@ -28,6 +28,9 @@ import { FailedPodHandler } from "./services/failedPodHandler.js";
 import { getWorkerToken } from "./workerToken.js";
 import { OtlpTraceService } from "./services/otlpTraceService.js";
 import { extractTraceparent, getRestoreRunnerId } from "./util.js";
+import { createRedisClient } from "@internal/redis";
+import { BackpressureMonitor } from "./backpressure/backpressureMonitor.js";
+import { RedisBackpressureSignalSource } from "./backpressure/redisBackpressureSignalSource.js";
 import {
   fromContext,
   recordPhaseSince,
@@ -54,6 +57,7 @@ class ManagedSupervisor {
   private readonly podCleaner?: PodCleaner;
   private readonly failedPodHandler?: FailedPodHandler;
   private readonly tracing?: OtlpTraceService;
+  private readonly backpressureMonitor?: BackpressureMonitor;
 
   private readonly isKubernetes = isKubernetesEnvironment(env.KUBERNETES_FORCE_ENABLED);
   private readonly warmStartUrl = env.TRIGGER_WARM_START_URL;
@@ -181,6 +185,38 @@ class ManagedSupervisor {
       );
     }
 
+    if (env.TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED) {
+      const backpressureRedis = createRedisClient(
+        {
+          host: env.TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST,
+          port: env.TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_PORT,
+          username: env.TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_USERNAME,
+          password: env.TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_PASSWORD,
+          ...(env.TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_TLS_DISABLED ? {} : { tls: {} }),
+        },
+        {
+          onError: (error) =>
+            this.logger.error("Backpressure redis error", { error: error.message }),
+        }
+      );
+
+      this.backpressureMonitor = new BackpressureMonitor({
+        enabled: true,
+        source: new RedisBackpressureSignalSource(
+          backpressureRedis,
+          env.TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_KEY
+        ),
+        refreshIntervalMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_REFRESH_MS,
+        maxVerdictAgeMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_MAX_VERDICT_AGE_MS,
+      });
+
+      this.logger.log("🛑 Dequeue backpressure enabled", {
+        key: env.TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_KEY,
+        refreshIntervalMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_REFRESH_MS,
+        maxVerdictAgeMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_MAX_VERDICT_AGE_MS,
+      });
+    }
+
     this.workerSession = new SupervisorSession({
       workerToken: getWorkerToken(),
       apiUrl: env.TRIGGER_API_URL,
@@ -206,13 +242,12 @@ class ManagedSupervisor {
       heartbeatIntervalSeconds: env.TRIGGER_WORKER_HEARTBEAT_INTERVAL_SECONDS,
       sendRunDebugLogs: env.SEND_RUN_DEBUG_LOGS,
       preDequeue: async () => {
-        if (!env.RESOURCE_MONITOR_ENABLED) {
-          return {};
-        }
+        // Synchronous, hot-path-safe cached read; undefined when backpressure is disabled.
+        const skipForBackpressure = this.backpressureMonitor?.shouldSkipDequeue() ?? false;
 
-        if (this.isKubernetes) {
-          // Not used in k8s for now
-          return {};
+        if (!env.RESOURCE_MONITOR_ENABLED || this.isKubernetes) {
+          // Resource monitor is not used in k8s; backpressure is the only gate there.
+          return { skipDequeue: skipForBackpressure };
         }
 
         const resources = await this.resourceMonitor.getNodeResources();
@@ -222,7 +257,10 @@ class ManagedSupervisor {
             cpu: resources.cpuAvailable,
             memory: resources.memoryAvailable,
           },
-          skipDequeue: resources.cpuAvailable < 0.25 || resources.memoryAvailable < 0.25,
+          skipDequeue:
+            skipForBackpressure ||
+            resources.cpuAvailable < 0.25 ||
+            resources.memoryAvailable < 0.25,
         };
       },
       preSkip: async () => {
@@ -552,6 +590,7 @@ class ManagedSupervisor {
     this.logger.log("Starting up");
 
     // Optional services
+    this.backpressureMonitor?.start();
     await this.podCleaner?.start();
     await this.failedPodHandler?.start();
     await this.metricsServer?.start();
@@ -576,6 +615,7 @@ class ManagedSupervisor {
     await this.workerSession.stop();
 
     // Optional services
+    this.backpressureMonitor?.stop();
     await this.podCleaner?.stop();
     await this.failedPodHandler?.stop();
     await this.metricsServer?.stop();