supervisor uses new idle interval after empty dequeues and errors

Author: nicktrn

apps/supervisor/src/env.ts

@@ -31,7 +31,8 @@ const Env = z.object({
 
   // Dequeue settings (provider mode)
   TRIGGER_DEQUEUE_ENABLED: BoolEnv.default("true"),
-  TRIGGER_DEQUEUE_INTERVAL_MS: z.coerce.number().int().default(1000),
+  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),
 

apps/supervisor/src/index.ts

@@ -116,6 +116,7 @@ class ManagedSupervisor {
       instanceName: env.TRIGGER_WORKER_INSTANCE_NAME,
       managedWorkerSecret: env.MANAGED_WORKER_SECRET,
       dequeueIntervalMs: env.TRIGGER_DEQUEUE_INTERVAL_MS,
+      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,

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

@@ -4,7 +4,8 @@ import { PreDequeueFn, PreSkipFn } from "./types.js";
 
 type RunQueueConsumerOptions = {
   client: SupervisorHttpClient;
-  intervalMs?: number;
+  intervalMs: number;
+  idleIntervalMs: number;
   preDequeue?: PreDequeueFn;
   preSkip?: PreSkipFn;
   maxRunCount?: number;
@@ -19,11 +20,13 @@ export class RunQueueConsumer {
   private readonly onDequeue: (messages: WorkerApiDequeueResponseBody) => Promise<void>;
 
   private intervalMs: number;
+  private idleIntervalMs: number;
   private isEnabled: boolean;
 
   constructor(opts: RunQueueConsumerOptions) {
     this.isEnabled = false;
-    this.intervalMs = opts.intervalMs ?? 5_000;
+    this.intervalMs = opts.intervalMs;
+    this.idleIntervalMs = opts.idleIntervalMs;
     this.preDequeue = opts.preDequeue;
     this.preSkip = opts.preSkip;
     this.maxRunCount = opts.maxRunCount;
@@ -84,9 +87,11 @@ export class RunQueueConsumer {
         }
       }
 
-      return this.scheduleNextDequeue();
+      return this.scheduleNextDequeue(this.idleIntervalMs);
     }
 
+    let nextIntervalMs = this.idleIntervalMs;
+
     try {
       const response = await this.client.dequeue({
         maxResources: preDequeueResult?.maxResources,
@@ -98,6 +103,10 @@ export class RunQueueConsumer {
       } else {
         try {
           await this.onDequeue(response.data);
+
+          if (response.data.length > 0) {
+            nextIntervalMs = this.intervalMs;
+          }
         } catch (handlerError) {
           console.error("[RunQueueConsumer] onDequeue error", { error: handlerError });
         }
@@ -106,10 +115,10 @@ export class RunQueueConsumer {
       console.error("[RunQueueConsumer] client.dequeue error", { error: clientError });
     }
 
-    this.scheduleNextDequeue();
+    this.scheduleNextDequeue(nextIntervalMs);
   }
 
-  scheduleNextDequeue(delay: number = this.intervalMs) {
-    setTimeout(this.dequeue.bind(this), delay);
+  scheduleNextDequeue(delayMs: number) {
+    setTimeout(this.dequeue.bind(this), delayMs);
   }
 }

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

@@ -14,7 +14,8 @@ type SupervisorSessionOptions = SupervisorClientCommonOptions & {
   queueConsumerEnabled?: boolean;
   runNotificationsEnabled?: boolean;
   heartbeatIntervalSeconds?: number;
-  dequeueIntervalMs?: number;
+  dequeueIntervalMs: number;
+  dequeueIdleIntervalMs: number;
   preDequeue?: PreDequeueFn;
   preSkip?: PreSkipFn;
   maxRunCount?: number;
@@ -47,11 +48,11 @@ export class SupervisorSession extends EventEmitter<WorkerEvents> {
         preSkip: opts.preSkip,
         onDequeue: this.onDequeue.bind(this),
         intervalMs: opts.dequeueIntervalMs,
+        idleIntervalMs: opts.dequeueIdleIntervalMs,
         maxRunCount: opts.maxRunCount,
       });
     });
 
-    // TODO: This should be dynamic and set by (or at least overridden by) the platform
     this.heartbeatIntervalSeconds = opts.heartbeatIntervalSeconds || 30;
     this.heartbeat = new IntervalService({
       onInterval: async () => {