handle scale down when queue length drops to zero

Author: nicktrn

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

@@ -589,20 +589,6 @@ describe("RunQueueConsumerPool", () => {
   });
 
   describe("Edge cases", () => {
-    it("should handle undefined queue lengths gracefully", async () => {
-      pool = new RunQueueConsumerPool({
-        ...defaultOptions,
-        scaling: { strategy: "smooth" },
-      });
-
-      await pool.start();
-
-      expect(() => pool.updateQueueLength(undefined)).not.toThrow();
-
-      const metrics = pool.getMetrics();
-      expect(metrics.queueLength).toBeUndefined();
-    });
-
     it("should handle empty recent queue lengths", async () => {
       pool = new RunQueueConsumerPool({
         ...defaultOptions,
@@ -688,5 +674,48 @@ describe("RunQueueConsumerPool", () => {
       advanceTimeAndProcessMetrics(1100);
       expect(pool.size).toBeLessThanOrEqual(6);
     });
+
+    it("should scale down when no items are dequeued (zero queue length)", async () => {
+      pool = new RunQueueConsumerPool({
+        ...defaultOptions,
+        scaling: {
+          strategy: "smooth",
+          minConsumerCount: 1,
+          maxConsumerCount: 10,
+          scaleUpCooldownMs: 0,
+          scaleDownCooldownMs: 0,
+          disableJitter: true,
+        },
+      });
+
+      await pool.start();
+      expect(pool.size).toBe(1);
+
+      // Scale up first
+      pool.updateQueueLength(20);
+      advanceTimeAndProcessMetrics(1100);
+      expect(pool.size).toBeGreaterThan(1);
+      const sizeAfterScaleUp = pool.size;
+
+      // Now send multiple zero queue lengths to converge EWMA to 0
+      // The EWMA needs time to converge due to exponential smoothing
+      for (let i = 0; i < 5; i++) {
+        pool.updateQueueLength(0);
+        advanceTimeAndProcessMetrics(1100);
+      }
+
+      // After multiple iterations with zero queue, should scale down but not to minimum yet
+      expect(pool.size).toBeLessThan(sizeAfterScaleUp);
+      expect(pool.size).toBeGreaterThan(1);
+
+      // Continue until we reach minimum
+      for (let i = 0; i < 5; i++) {
+        pool.updateQueueLength(0);
+        advanceTimeAndProcessMetrics(1100);
+      }
+
+      // Should eventually reach minimum
+      expect(pool.size).toBe(1);
+    });
   });
 });

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

@@ -182,11 +182,7 @@ export class RunQueueConsumerPool {
    * Updates the queue length metric and triggers scaling decisions
    * Uses QueueMetricsProcessor for batching and EWMA smoothing
    */
-  updateQueueLength(queueLength: number | undefined) {
-    if (queueLength === undefined) {
-      return;
-    }
-
+  updateQueueLength(queueLength: number) {
     // Track queue length update in metrics
     this.promMetrics?.recordQueueLengthUpdate();
 
@@ -356,10 +352,8 @@ export class RunQueueConsumerPool {
       const consumer = this.consumerFactory({
         ...this.consumerOptions,
         onDequeue: async (messages) => {
-          // Update queue length if provided
-          if (messages.length > 0 && messages[0]?.workerQueueLength !== undefined) {
-            this.updateQueueLength(messages[0]?.workerQueueLength);
-          }
+          // Always update queue length, default to 0 for empty dequeues or missing value
+          this.updateQueueLength(messages[0]?.workerQueueLength ?? 0);
 
           // Forward to the original handler
           await this.consumerOptions.onDequeue(messages);