process the queue on nack

Author: ericallam

internal-packages/run-engine/src/engine/tests/heartbeats.test.ts

@@ -350,6 +350,8 @@ describe("RunEngine heartbeats", () => {
         assertNonNullable(executionData2);
         expect(executionData2.snapshot.executionStatus).toBe("QUEUED");
 
+        await setTimeout(1_000);
+
         //have to dequeue again
         const dequeued2 = await engine.dequeueFromWorkerQueue({
           consumerId: "test_12345",

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

@@ -610,11 +610,13 @@ export class RunQueue {
     messageId,
     retryAt,
     incrementAttemptCount = true,
+    skipDequeueProcessing = false,
   }: {
     orgId: string;
     messageId: string;
     retryAt?: number;
     incrementAttemptCount?: boolean;
+    skipDequeueProcessing?: boolean;
   }) {
     return this.#trace(
       "nackMessage",
@@ -648,6 +650,21 @@ export class RunQueue {
           }
         }
 
+        if (!skipDequeueProcessing) {
+          // This will move the message to the worker queue so it can be dequeued
+          await this.worker.enqueueOnce({
+            id: message.queue, // dedupe by environment, queue, and concurrency key
+            job: "processQueueForWorkerQueue",
+            payload: {
+              queueKey: message.queue,
+              environmentId: message.environmentId,
+            },
+            // Add a small delay to dedupe messages so at most one of these will processed,
+            // every 500ms per queue, concurrency key, and environment
+            availableAt: new Date(Date.now() + (this.options.processWorkerQueueDebounceMs ?? 500)), // 500ms from now
+          });
+        }
+
         await this.#callNackMessage({ message, retryAt });
 
         return true;