optionally skip dequeue

Author: nicktrn

packages/worker/src/supervisor/queueConsumer.ts

@@ -1,21 +1,17 @@
-import { MachineResources } from "@trigger.dev/core/v3";
 import { SupervisorHttpClient } from "./http.js";
 import { WorkerApiDequeueResponseBody } from "./schemas.js";
+import { PreDequeueFn } from "./types.js";
 
 type RunQueueConsumerOptions = {
   client: SupervisorHttpClient;
   intervalMs?: number;
-  preDequeue?: () => Promise<{
-    maxResources?: MachineResources;
-  }>;
+  preDequeue?: PreDequeueFn;
   onDequeue: (messages: WorkerApiDequeueResponseBody) => Promise<void>;
 };
 
 export class RunQueueConsumer {
   private readonly client: SupervisorHttpClient;
-  private readonly preDequeue?: () => Promise<{
-    maxResources?: MachineResources;
-  }>;
+  private readonly preDequeue?: PreDequeueFn;
   private readonly onDequeue: (messages: WorkerApiDequeueResponseBody) => Promise<void>;
 
   private intervalMs: number;
@@ -54,18 +50,27 @@ export class RunQueueConsumer {
       return;
     }
 
-    let maxResources: MachineResources | undefined;
+    let preDequeueResult: Awaited<ReturnType<PreDequeueFn>> | undefined;
     if (this.preDequeue) {
       try {
-        const preDequeueResult = await this.preDequeue();
-        maxResources = preDequeueResult.maxResources;
+        preDequeueResult = await this.preDequeue();
       } catch (preDequeueError) {
         console.error("[RunQueueConsumer] preDequeue error", { error: preDequeueError });
       }
     }
 
+    if (
+      preDequeueResult?.skipDequeue ||
+      preDequeueResult?.maxResources?.cpu === 0 ||
+      preDequeueResult?.maxResources?.memory === 0
+    ) {
+      return this.scheduleNextDequeue();
+    }
+
     try {
-      const response = await this.client.dequeue({ maxResources });
+      const response = await this.client.dequeue({
+        maxResources: preDequeueResult?.maxResources,
+      });
 
       if (!response.success) {
         console.error("[RunQueueConsumer] Failed to dequeue", { error: response.error });

packages/worker/src/supervisor/session.ts

@@ -1,6 +1,6 @@
 import { HeartbeatService, MachineResources } from "@trigger.dev/core/v3";
 import { SupervisorHttpClient } from "./http.js";
-import { SupervisorClientCommonOptions } from "./types.js";
+import { PreDequeueFn, SupervisorClientCommonOptions } from "./types.js";
 import { WorkerApiDequeueResponseBody, WorkerApiHeartbeatRequestBody } from "./schemas.js";
 import { RunQueueConsumer } from "./queueConsumer.js";
 import { WorkerEvents } from "./events.js";
@@ -13,9 +13,7 @@ import { getDefaultWorkerHeaders } from "./util.js";
 type SupervisorSessionOptions = SupervisorClientCommonOptions & {
   heartbeatIntervalSeconds?: number;
   dequeueIntervalMs?: number;
-  preDequeue?: () => Promise<{
-    maxResources?: MachineResources;
-  }>;
+  preDequeue?: PreDequeueFn;
 };
 
 export class SupervisorSession extends EventEmitter<WorkerEvents> {

packages/worker/src/supervisor/types.ts

@@ -1,7 +1,14 @@
+import type { MachineResources } from "@trigger.dev/core/v3";
+
 export type SupervisorClientCommonOptions = {
   apiUrl: string;
   workerToken: string;
   instanceName: string;
   deploymentId?: string;
   managedWorkerSecret?: string;
 };
+
+export type PreDequeueFn = () => Promise<{
+  maxResources?: MachineResources;
+  skipDequeue?: boolean;
+}>;