Add ability to enable v2 marqs requeuing (and remove deprecated marqs visibility requeuing)

ericallam · ericallam · commit 393af1b7c554 · 2024-06-06T17:14:15.000+01:00
diff --git a/apps/webapp/app/v3/marqs/index.server.ts b/apps/webapp/app/v3/marqs/index.server.ts
@@ -78,8 +78,6 @@ export class MarQS {
     this.keys = options.keysProducer;
     this.queuePriorityStrategy = options.queuePriorityStrategy;
 
-    // Spawn options.workers workers to requeue visible messages
-    this.#startRequeuingWorkers();
     this.#startRebalanceWorkers();
     this.#registerCommands();
   }
@@ -790,66 +788,6 @@ export class MarQS {
     }
   }
 
-  #startRequeuingWorkers() {
-    // Start a new worker to requeue visible messages
-    for (let i = 0; i < this.options.workers; i++) {
-      const worker = new AsyncWorker(this.#requeueVisibleMessages.bind(this), 1000);
-
-      this.#requeueingWorkers.push(worker);
-
-      worker.start();
-    }
-  }
-
-  async #requeueVisibleMessages() {
-    // Remove any of the messages from the timeoutQueue that have expired
-    const messages = await this.redis.zrangebyscore(
-      constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
-      0,
-      Date.now(),
-      "LIMIT",
-      0,
-      10
-    );
-
-    if (messages.length === 0) {
-      return;
-    }
-
-    for (let i = 0; i < messages.length; i++) {
-      const message = messages[i];
-
-      const messageData = await this.redis.get(this.keys.messageKey(message));
-
-      if (!messageData) {
-        // The message has been removed for some reason (TTL, etc.), so we should remove it from the timeout queue
-        await this.redis.zrem(constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE, message);
-
-        continue;
-      }
-
-      const parsedMessage = MessagePayload.safeParse(JSON.parse(messageData));
-
-      if (!parsedMessage.success) {
-        await this.redis.zrem(constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE, message);
-
-        continue;
-      }
-
-      await this.#callNackMessage({
-        messageKey: this.keys.messageKey(message),
-        messageQueue: parsedMessage.data.queue,
-        parentQueue: parsedMessage.data.parentQueue,
-        concurrencyKey: this.keys.currentConcurrencyKeyFromQueue(parsedMessage.data.queue),
-        envConcurrencyKey: this.keys.envCurrentConcurrencyKeyFromQueue(parsedMessage.data.queue),
-        orgConcurrencyKey: this.keys.orgCurrentConcurrencyKeyFromQueue(parsedMessage.data.queue),
-        visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
-        messageId: parsedMessage.data.messageId,
-        messageScore: parsedMessage.data.timestamp,
-      });
-    }
-  }
-
   async #rebalanceParentQueues() {
     return await new Promise<void>((resolve, reject) => {
       // Scan for sorted sets with the parent queue pattern
diff --git a/apps/webapp/app/v3/marqs/v2.server.ts b/apps/webapp/app/v3/marqs/v2.server.ts
@@ -82,7 +82,7 @@ function getMarQSClient() {
     defaultEnvConcurrency: env.V2_MARQS_DEFAULT_ENV_CONCURRENCY, // this is so we aren't limited by the environment concurrency
     defaultOrgConcurrency: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT,
     visibilityTimeoutInMs: env.V2_MARQS_VISIBILITY_TIMEOUT_MS, // 15 minutes
-    enableRebalancing: !env.MARQS_DISABLE_REBALANCING,
+    enableRebalancing: env.V2_MARQS_CONSUMER_POOL_ENABLED === "1",
   });
 }
 

Original file line number	Diff line number	Diff line change
`@@ -82,7 +82,7 @@ function getMarQSClient() {`
`82`	`82`	`defaultEnvConcurrency: env.V2_MARQS_DEFAULT_ENV_CONCURRENCY, // this is so we aren't limited by the environment concurrency`
`83`	`83`	`defaultOrgConcurrency: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT,`
`84`	`84`	`visibilityTimeoutInMs: env.V2_MARQS_VISIBILITY_TIMEOUT_MS, // 15 minutes`
`85`		`- enableRebalancing: !env.MARQS_DISABLE_REBALANCING,`
	`85`	`+ enableRebalancing: env.V2_MARQS_CONSUMER_POOL_ENABLED === "1",`
`86`	`86`	`});`
`87`	`87`	`}`
`88`	`88`