feat(core): freeze consumer-pool scale-up under backpressure

Add optional shouldPauseScaling to ScalingOptions; when it returns true the
pool stops scaling up (scale-down still allowed), so it won't add consumers to
drain a queue backpressure is deliberately holding.

Author: nicktrn

.changeset/backpressure-scale-up-freeze.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Add optional `shouldPauseScaling` to the supervisor consumer pool scaling options to freeze scale-up while it returns true (scale-down stays allowed).

packages/core/src/v3/runEngineWorker/supervisor/consumerPool.test.ts

@@ -718,4 +718,65 @@ describe("RunQueueConsumerPool", () => {
       expect(pool.size).toBe(1);
     });
   });
+
+  describe("Backpressure scale-up freeze", () => {
+    it("freezes scale-up while shouldPauseScaling returns true, then resumes", async () => {
+      let paused = true;
+      pool = new RunQueueConsumerPool({
+        ...defaultOptions,
+        scaling: {
+          strategy: "smooth",
+          minConsumerCount: 1,
+          maxConsumerCount: 10,
+          scaleUpCooldownMs: 0,
+          disableJitter: true,
+          shouldPauseScaling: () => paused,
+        },
+      });
+      await pool.start();
+      expect(pool.size).toBe(1);
+
+      // A high queue would normally scale up, but backpressure freezes it.
+      pool.updateQueueLength(10);
+      advanceTimeAndProcessMetrics(1100);
+      expect(pool.size).toBe(1);
+
+      // Once backpressure releases, scaling resumes.
+      paused = false;
+      pool.updateQueueLength(10);
+      advanceTimeAndProcessMetrics(1100);
+      expect(pool.size).toBeGreaterThan(1);
+    });
+
+    it("still allows scale-down while paused", async () => {
+      let paused = false;
+      pool = new RunQueueConsumerPool({
+        ...defaultOptions,
+        scaling: {
+          strategy: "smooth",
+          minConsumerCount: 1,
+          maxConsumerCount: 10,
+          scaleUpCooldownMs: 0,
+          scaleDownCooldownMs: 0,
+          disableJitter: true,
+          shouldPauseScaling: () => paused,
+        },
+      });
+      await pool.start();
+
+      pool.updateQueueLength(10);
+      advanceTimeAndProcessMetrics(1100);
+      const scaledUp = pool.size;
+      expect(scaledUp).toBeGreaterThan(1);
+
+      // Pausing must not block shrinking - we want to drain down, just not grow.
+      // Loop to let the EWMA-smoothed queue length fall (one batch isn't enough).
+      paused = true;
+      for (let i = 0; i < 5; i++) {
+        pool.updateQueueLength(0);
+        advanceTimeAndProcessMetrics(1100);
+      }
+      expect(pool.size).toBeLessThan(scaledUp);
+    });
+  });
 });

packages/core/src/v3/runEngineWorker/supervisor/consumerPool.ts

@@ -22,6 +22,12 @@ export type ScalingOptions = {
   batchWindowMs?: number;
   disableJitter?: boolean;
   dampingFactor?: number;
+  /**
+   * When this returns true, scale-up is frozen (scale-down still allowed). Used to
+   * stop the pool from adding consumers to drain a queue that backpressure is
+   * deliberately holding. Synchronous and hot-path-safe.
+   */
+  shouldPauseScaling?: () => boolean;
 };
 
 export type ConsumerPoolOptions = {
@@ -49,6 +55,7 @@ export class RunQueueConsumerPool {
   private readonly maxConsumerCount: number;
   private readonly scalingStrategy: ScalingStrategy;
   private readonly disableJitter: boolean;
+  private readonly shouldPauseScaling?: () => boolean;
 
   private consumers: Map<string, QueueConsumer> = new Map();
   private readonly consumerFactory: QueueConsumerFactory;
@@ -79,6 +86,7 @@ export class RunQueueConsumerPool {
     this.scaleUpCooldownMs = opts.scaling.scaleUpCooldownMs ?? 10000; // 10 seconds default
     this.scaleDownCooldownMs = opts.scaling.scaleDownCooldownMs ?? 60000; // 60 seconds default
     this.disableJitter = opts.scaling.disableJitter ?? false;
+    this.shouldPauseScaling = opts.scaling.shouldPauseScaling;
 
     // Configure EWMA parameters from options
     this.ewmaAlpha = opts.scaling.ewmaAlpha ?? 0.3;
@@ -259,6 +267,16 @@ export class RunQueueConsumerPool {
 
     // Check cooldown periods with jitter
     if (targetCount > this.consumers.size) {
+      // Freeze scale-up while backpressure is engaged - don't add consumers to
+      // drain a queue we're deliberately holding. Scale-down stays allowed.
+      if (this.shouldPauseScaling?.()) {
+        this.logger.debug("Scale up frozen by backpressure", {
+          currentCount: this.consumers.size,
+          targetCount,
+        });
+        return;
+      }
+
       // Scale up
       const effectiveCooldown = this.scaleUpCooldownMs + jitterMs;
       if (timeSinceLastScale < effectiveCooldown) {