Acking should also cause the master queue to be processed

ericallam · ericallam · commit d1a2723dfc63 · 2025-05-29T14:33:09.000+01:00
diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts
@@ -112,6 +112,13 @@ export class RunEngine {
       logger: new Logger("RunQueue", "debug"),
       redis: { ...options.queue.redis, keyPrefix: `${options.queue.redis.keyPrefix}runqueue:` },
       retryOptions: options.queue?.retryOptions,
+      workerOptions: {
+        disabled: options.worker.disabled,
+        concurrency: options.worker,
+        pollIntervalMs: options.worker.pollIntervalMs,
+        immediatePollIntervalMs: options.worker.immediatePollIntervalMs,
+        shutdownTimeoutMs: options.worker.shutdownTimeoutMs,
+      },
     });
 
     this.worker = new Worker({
diff --git a/internal-packages/run-engine/src/run-queue/index.test.ts b/internal-packages/run-engine/src/run-queue/index.test.ts
@@ -375,6 +375,113 @@ describe("RunQueue", () => {
     }
   );
 
+  redisTest(
+    "Dequeue a message when another message on the same queue is acked",
+    async ({ redisContainer }) => {
+      const queue = new RunQueue({
+        ...testOptions,
+        masterQueueConsumersDisabled: true,
+        processWorkerQueueDebounceMs: 50,
+        queueSelectionStrategy: new FairQueueSelectionStrategy({
+          redis: {
+            keyPrefix: "runqueue:test:",
+            host: redisContainer.getHost(),
+            port: redisContainer.getPort(),
+          },
+          keys: testOptions.keys,
+        }),
+        redis: {
+          keyPrefix: "runqueue:test:",
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+        },
+      });
+
+      try {
+        // Set queue concurrency limit to 1
+        await queue.updateQueueConcurrencyLimits(authenticatedEnvProd, messageProd.queue, 1);
+
+        //initial queue length
+        const result = await queue.lengthOfQueue(authenticatedEnvProd, messageProd.queue);
+        expect(result).toBe(0);
+        const envQueueLength = await queue.lengthOfEnvQueue(authenticatedEnvProd);
+        expect(envQueueLength).toBe(0);
+
+        //initial oldest message
+        const oldestScore = await queue.oldestMessageInQueue(
+          authenticatedEnvProd,
+          messageProd.queue
+        );
+        expect(oldestScore).toBe(undefined);
+
+        //enqueue message
+        await queue.enqueueMessage({
+          env: authenticatedEnvProd,
+          message: messageProd,
+          workerQueue: "main",
+          skipDequeueProcessing: true,
+        });
+
+        // Enqueue another message
+        await queue.enqueueMessage({
+          env: authenticatedEnvProd,
+          message: { ...messageProd, runId: "r4322" },
+          workerQueue: "main",
+          skipDequeueProcessing: true,
+        });
+
+        //queue length
+        const queueLength = await queue.lengthOfQueue(authenticatedEnvProd, messageProd.queue);
+        expect(queueLength).toBe(2);
+        const envLength = await queue.lengthOfEnvQueue(authenticatedEnvProd);
+        expect(envLength).toBe(2);
+
+        //oldest message
+        const oldestScore2 = await queue.oldestMessageInQueue(
+          authenticatedEnvProd,
+          messageProd.queue
+        );
+        expect(oldestScore2).toBe(messageProd.timestamp);
+
+        //concurrencies
+        const queueConcurrency = await queue.currentConcurrencyOfQueue(
+          authenticatedEnvProd,
+          messageProd.queue
+        );
+        expect(queueConcurrency).toBe(0);
+        const envConcurrency = await queue.currentConcurrencyOfEnvironment(authenticatedEnvProd);
+        expect(envConcurrency).toBe(0);
+
+        // Process the message so it can be dequeued
+        await queue.processMasterQueueForEnvironment(authenticatedEnvProd.id, 1);
+
+        //dequeue
+        const dequeued = await queue.dequeueMessageFromWorkerQueue("test_12345", "main");
+
+        assertNonNullable(dequeued);
+        expect(dequeued).toBeDefined();
+        expect(dequeued!.messageId).toEqual(messageProd.runId);
+        expect(dequeued!.message.orgId).toEqual(messageProd.orgId);
+        expect(dequeued!.message.version).toEqual("2");
+
+        // Now lets ack the message
+        await queue.acknowledgeMessage(messageProd.orgId, messageProd.runId);
+
+        await setTimeout(1000);
+
+        // Now we can dequeue the other message
+        const dequeued2 = await queue.dequeueMessageFromWorkerQueue("test_12345", "main");
+        assertNonNullable(dequeued2);
+        expect(dequeued2).toBeDefined();
+        expect(dequeued2!.messageId).toEqual("r4322");
+        expect(dequeued2!.message.orgId).toEqual(messageProd.orgId);
+        expect(dequeued2!.message.version).toEqual("2");
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
+
   redisTest("Enqueue/Dequeue a 8 shards", async ({ redisContainer }) => {
     const queue = new RunQueue({
       ...testOptions,
diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts
@@ -88,11 +88,8 @@ const defaultRetrySettings = {
 const workerCatalog = {
   processQueueForWorkerQueue: {
     schema: z.object({
-      queue: z.string(),
-      concurrencyKey: z.string().optional(),
+      queueKey: z.string(),
       environmentId: z.string(),
-      projectId: z.string(),
-      orgId: z.string(),
     }),
     visibilityTimeoutMs: 30_000,
   },
@@ -148,13 +145,7 @@ export class RunQueue {
       logger: new Logger("RunQueueWorker", options.logLevel ?? "log"),
       jobs: {
         processQueueForWorkerQueue: async (job) => {
-          await this.#processQueueForWorkerQueue(
-            job.payload.queue,
-            job.payload.environmentId,
-            job.payload.projectId,
-            job.payload.orgId,
-            job.payload.concurrencyKey
-          );
+          await this.#processQueueForWorkerQueue(job.payload.queueKey, job.payload.environmentId);
         },
       },
     });
@@ -426,11 +417,8 @@ export class RunQueue {
             id: queueKey, // dedupe by environment, queue, and concurrency key
             job: "processQueueForWorkerQueue",
             payload: {
-              queue: message.queue,
-              concurrencyKey,
+              queueKey,
               environmentId: env.id,
-              orgId: env.organization.id,
-              projectId: env.project.id,
             },
             // Add a small delay to dedupe messages so at most one of these will processed,
             // every 500ms per queue, concurrency key, and environment
@@ -501,7 +489,11 @@ export class RunQueue {
    * This is done when the run is in a final state.
    * @param messageId
    */
-  public async acknowledgeMessage(orgId: string, messageId: string) {
+  public async acknowledgeMessage(
+    orgId: string,
+    messageId: string,
+    options?: { skipDequeueProcessing?: boolean }
+  ) {
     return this.#trace(
       "acknowledgeMessage",
       async (span) => {
@@ -519,6 +511,21 @@ export class RunQueue {
           [SemanticAttributes.CONCURRENCY_KEY]: message.concurrencyKey,
         });
 
+        if (!options?.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.#callAcknowledgeMessage({
           message,
         });
@@ -1035,15 +1042,7 @@ export class RunQueue {
     );
   }
 
-  async #processQueueForWorkerQueue(
-    queue: string,
-    environmentId: string,
-    projectId: string,
-    orgId: string,
-    concurrencyKey?: string
-  ) {
-    const queueKey = this.keys.queueKey(orgId, projectId, environmentId, queue, concurrencyKey);
-
+  async #processQueueForWorkerQueue(queueKey: string, environmentId: string) {
     const shard = this.keys.masterQueueShardForEnvironment(environmentId, this.shardCount);
 
     this.logger.debug("processQueueForWorkerQueue", {
