more eager dequeuing, queue cooloff periods, return workerQueueLength when dequeueing

Author: ericallam

internal-packages/run-engine/src/engine/index.ts

@@ -566,6 +566,8 @@ export class RunEngine {
     runnerId,
     tx,
     skipObserving,
+    blockingPop,
+    blockingPopTimeoutSeconds,
   }: {
     consumerId: string;
     workerQueue: string;
@@ -574,6 +576,8 @@ export class RunEngine {
     runnerId?: string;
     tx?: PrismaClientOrTransaction;
     skipObserving?: boolean;
+    blockingPop?: boolean;
+    blockingPopTimeoutSeconds?: number;
   }): Promise<DequeuedMessage[]> {
     if (!skipObserving) {
       // We only do this with "prod" worker queues because we don't want to observe dev (e.g. environment) worker queues
@@ -587,6 +591,8 @@ export class RunEngine {
       workerId,
       runnerId,
       tx,
+      blockingPop,
+      blockingPopTimeoutSeconds,
     });
 
     if (!dequeuedMessage) {
@@ -619,6 +625,8 @@ export class RunEngine {
       runnerId,
       tx,
       skipObserving: true,
+      blockingPop: true,
+      blockingPopTimeoutSeconds: 10,
     });
   }
 

internal-packages/run-engine/src/engine/systems/dequeueSystem.ts

@@ -46,13 +46,17 @@ export class DequeueSystem {
     workerId,
     runnerId,
     tx,
+    blockingPop,
+    blockingPopTimeoutSeconds,
   }: {
     consumerId: string;
     workerQueue: string;
     backgroundWorkerId?: string;
     workerId?: string;
     runnerId?: string;
     tx?: PrismaClientOrTransaction;
+    blockingPop?: boolean;
+    blockingPopTimeoutSeconds?: number;
   }): Promise<DequeuedMessage | undefined> {
     const prisma = tx ?? this.$.prisma;
 
@@ -63,7 +67,11 @@ export class DequeueSystem {
         //gets multiple runs from the queue
         const message = await this.$.runQueue.dequeueMessageFromWorkerQueue(
           consumerId,
-          workerQueue
+          workerQueue,
+          {
+            blockingPop,
+            blockingPopTimeoutSeconds,
+          }
         );
         if (!message) {
           return;
@@ -452,6 +460,7 @@ export class DequeueSystem {
               return {
                 version: "1" as const,
                 dequeuedAt: new Date(),
+                workerQueueLength: message.workerQueueLength,
                 snapshot: {
                   id: newSnapshot.id,
                   friendlyId: newSnapshot.friendlyId,