feat(supervisor): dynamic queue consumer pool (#2461)

* feat(supervisor): dynamic queue consumer pool

* add changeset

* fix: correctly handle zero median and even samples

* feat(supervisor): consumer pool metrics

* fix tests

* more tests and fixes

* decrease default scaling cooldowns

* don't treat initial pool size as scale up

* handle scale down when queue length drops to zero

* remove changeset, supervisor changes only

* add damping factor env var

Author: nicktrn

apps/supervisor/src/env.ts

@@ -35,8 +35,16 @@ const Env = z.object({
   TRIGGER_DEQUEUE_ENABLED: BoolEnv.default(true),
   TRIGGER_DEQUEUE_INTERVAL_MS: z.coerce.number().int().default(250),
   TRIGGER_DEQUEUE_IDLE_INTERVAL_MS: z.coerce.number().int().default(1000),
-  TRIGGER_DEQUEUE_MAX_RUN_COUNT: z.coerce.number().int().default(10),
-  TRIGGER_DEQUEUE_MAX_CONSUMER_COUNT: z.coerce.number().int().default(1),
+  TRIGGER_DEQUEUE_MAX_RUN_COUNT: z.coerce.number().int().default(1),
+  TRIGGER_DEQUEUE_MIN_CONSUMER_COUNT: z.coerce.number().int().default(1),
+  TRIGGER_DEQUEUE_MAX_CONSUMER_COUNT: z.coerce.number().int().default(10),
+  TRIGGER_DEQUEUE_SCALING_STRATEGY: z.enum(["none", "smooth", "aggressive"]).default("none"),
+  TRIGGER_DEQUEUE_SCALING_UP_COOLDOWN_MS: z.coerce.number().int().default(5000), // 5 seconds
+  TRIGGER_DEQUEUE_SCALING_DOWN_COOLDOWN_MS: z.coerce.number().int().default(30000), // 30 seconds
+  TRIGGER_DEQUEUE_SCALING_TARGET_RATIO: z.coerce.number().default(1.0), // Target ratio of queue items to consumers (1.0 = 1 item per consumer)
+  TRIGGER_DEQUEUE_SCALING_EWMA_ALPHA: z.coerce.number().min(0).max(1).default(0.3), // Smooths queue length measurements (0=historical, 1=current)
+  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)
 
   // Optional services
   TRIGGER_WARM_START_URL: z.string().optional(),

apps/supervisor/src/index.ts

@@ -128,7 +128,18 @@ class ManagedSupervisor {
       dequeueIdleIntervalMs: env.TRIGGER_DEQUEUE_IDLE_INTERVAL_MS,
       queueConsumerEnabled: env.TRIGGER_DEQUEUE_ENABLED,
       maxRunCount: env.TRIGGER_DEQUEUE_MAX_RUN_COUNT,
-      maxConsumerCount: env.TRIGGER_DEQUEUE_MAX_CONSUMER_COUNT,
+      metricsRegistry: register,
+      scaling: {
+        strategy: env.TRIGGER_DEQUEUE_SCALING_STRATEGY,
+        minConsumerCount: env.TRIGGER_DEQUEUE_MIN_CONSUMER_COUNT,
+        maxConsumerCount: env.TRIGGER_DEQUEUE_MAX_CONSUMER_COUNT,
+        scaleUpCooldownMs: env.TRIGGER_DEQUEUE_SCALING_UP_COOLDOWN_MS,
+        scaleDownCooldownMs: env.TRIGGER_DEQUEUE_SCALING_DOWN_COOLDOWN_MS,
+        targetRatio: env.TRIGGER_DEQUEUE_SCALING_TARGET_RATIO,
+        ewmaAlpha: env.TRIGGER_DEQUEUE_SCALING_EWMA_ALPHA,
+        batchWindowMs: env.TRIGGER_DEQUEUE_SCALING_BATCH_WINDOW_MS,
+        dampingFactor: env.TRIGGER_DEQUEUE_SCALING_DAMPING_FACTOR,
+      },
       runNotificationsEnabled: env.TRIGGER_WORKLOAD_API_ENABLED,
       heartbeatIntervalSeconds: env.TRIGGER_WORKER_HEARTBEAT_INTERVAL_SECONDS,
       sendRunDebugLogs: env.SEND_RUN_DEBUG_LOGS,

packages/core/src/v3/runEngineWorker/index.ts

@@ -2,6 +2,7 @@ export * from "./consts.js";
 export * from "./supervisor/http.js";
 export * from "./supervisor/schemas.js";
 export * from "./supervisor/session.js";
+export * from "./supervisor/consumerPool.js";
 export * from "./workload/http.js";
 export * from "./workload/schemas.js";
 export * from "./types.js";