fix(v3): prevent saturated queues from dominating dequeue attempts by adding a cooloff period of successive failed dequeues

Author: ericallam

apps/webapp/app/v3/marqs/index.server.ts

@@ -43,7 +43,7 @@ import {
   MARQS_RETRY_PRIORITY_TIMESTAMP_OFFSET,
   MARQS_SCHEDULED_REQUEUE_AVAILABLE_AT_THRESHOLD_IN_MS,
 } from "./constants.server";
-import { setInterval } from "node:timers/promises";
+import { setInterval as setIntervalAsync } from "node:timers/promises";
 import { tryCatch } from "@trigger.dev/core/utils";
 import { Worker, type WorkerConcurrencyOptions } from "@trigger.dev/redis-worker";
 import z from "zod";
@@ -78,6 +78,8 @@ export type MarQSOptions = {
   subscriber?: MessageQueueSubscriber;
   sharedWorkerQueueConsumerIntervalMs?: number;
   sharedWorkerQueueMaxMessageCount?: number;
+  sharedWorkerQueueCooloffPeriodMs?: number;
+  sharedWorkerQueueCooloffCountThreshold?: number;
   eagerDequeuingEnabled?: boolean;
   workerOptions: {
     pollIntervalMs?: number;
@@ -107,6 +109,9 @@ export class MarQS {
   public keys: MarQSKeyProducer;
   #rebalanceWorkers: Array<AsyncWorker> = [];
   private worker: Worker<typeof workerCatalog>;
+  private queueDequeueCooloffPeriod: Map<string, number> = new Map();
+  private queueDequeueCooloffCounts: Map<string, number> = new Map();
+  private clearCooloffPeriodInterval: NodeJS.Timeout;
 
   constructor(private readonly options: MarQSOptions) {
     this.redis = options.redis;
@@ -116,6 +121,12 @@ export class MarQS {
     this.#startRebalanceWorkers();
     this.#registerCommands();
 
+    // This will prevent these cooloff maps from growing indefinitely
+    this.clearCooloffPeriodInterval = setInterval(() => {
+      this.queueDequeueCooloffCounts.clear();
+      this.queueDequeueCooloffPeriod.clear();
+    }, 60_000 * 10); // 10 minutes
+
     this.worker = new Worker({
       name: "marqs-worker",
       redisOptions: options.workerOptions.redisOptions,
@@ -737,7 +748,7 @@ export class MarQS {
     let processedCount = 0;
 
     try {
-      for await (const _ of setInterval(
+      for await (const _ of setIntervalAsync(
         this.options.sharedWorkerQueueConsumerIntervalMs ?? 500,
         null,
         {
@@ -821,6 +832,7 @@ export class MarQS {
         let attemptedEnvs = 0;
         let attemptedQueues = 0;
         let messageCount = 0;
+        let coolOffPeriodCount = 0;
 
         // Try each queue in order, attempt to dequeue a message from each queue, keep going until we've tried all the queues
         for (const env of envQueues) {
@@ -829,6 +841,20 @@ export class MarQS {
           for (const messageQueue of env.queues) {
             attemptedQueues++;
 
+            const cooloffPeriod = this.queueDequeueCooloffPeriod.get(messageQueue);
+
+            // If the queue is in a cooloff period, skip attempting to dequeue from it
+            if (cooloffPeriod) {
+              // If the cooloff period is still active, skip attempting to dequeue from it
+              if (cooloffPeriod > Date.now()) {
+                coolOffPeriodCount++;
+                continue;
+              } else {
+                // If the cooloff period is over, delete the cooloff period and attempt to dequeue from the queue
+                this.queueDequeueCooloffPeriod.delete(messageQueue);
+              }
+            }
+
             await this.#trace(
               "attemptDequeue",
               async (attemptDequeueSpan) => {
@@ -862,10 +888,32 @@ export class MarQS {
                   );
 
                   if (!messages || messages.length === 0) {
+                    const cooloffCount = this.queueDequeueCooloffCounts.get(messageQueue) ?? 0;
+
+                    const cooloffCountThreshold = Math.max(
+                      10,
+                      this.options.sharedWorkerQueueCooloffCountThreshold ?? 10
+                    ); // minimum of 10
+
+                    if (cooloffCount >= cooloffCountThreshold) {
+                      // If no messages were dequeued, set a cooloff period for the queue
+                      // This is to prevent the queue from being dequeued too frequently
+                      // and to give other queues a chance to dequeue messages more frequently
+                      this.queueDequeueCooloffPeriod.set(
+                        messageQueue,
+                        Date.now() + (this.options.sharedWorkerQueueCooloffPeriodMs ?? 10_000) // defaults to 10 seconds
+                      );
+                      this.queueDequeueCooloffCounts.delete(messageQueue);
+                    } else {
+                      this.queueDequeueCooloffCounts.set(messageQueue, cooloffCount + 1);
+                    }
+
                     attemptDequeueSpan.setAttribute("message_count", 0);
                     return null; // Try next queue if no message was dequeued
                   }
 
+                  this.queueDequeueCooloffCounts.delete(messageQueue);
+
                   messageCount += messages.length;
 
                   attemptDequeueSpan.setAttribute("message_count", messages.length);
@@ -916,6 +964,7 @@ export class MarQS {
         span.setAttribute("attempted_queues", attemptedQueues);
         span.setAttribute("attempted_envs", attemptedEnvs);
         span.setAttribute("message_count", messageCount);
+        span.setAttribute("cooloff_period_count", coolOffPeriodCount);
 
         return;
       },