v3: prevent envs with excess queues from degrading dequeue performance

apps/webapp/app/env.server.ts

@@ -384,6 +384,7 @@ const EnvironmentSchema = z.object({
     .int()
     .default(60 * 1000 * 15),
   MARQS_SHARED_QUEUE_LIMIT: z.coerce.number().int().default(1000),
+  MARQS_MAXIMUM_QUEUE_PER_ENV_COUNT: z.coerce.number().int().default(50),
   MARQS_DEV_QUEUE_LIMIT: z.coerce.number().int().default(1000),
   MARQS_MAXIMUM_NACK_COUNT: z.coerce.number().int().default(64),
   MARQS_CONCURRENCY_LIMIT_BIAS: z.coerce.number().default(0.75),

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

@@ -43,6 +43,11 @@ export type FairDequeuingStrategyOptions = {
   biases?: FairDequeuingStrategyBiases;
   reuseSnapshotCount?: number;
   maximumEnvCount?: number;
+  /**
+   * Maximum number of queues to process per environment
+   * If not provided, all queues in an environment will be processed
+   */
+  maximumQueuePerEnvCount?: number;
 };
 
 type FairQueueConcurrency = {
@@ -216,8 +221,6 @@ export class FairDequeuingStrategy implements MarQSFairDequeueStrategy {
     return result;
   }
 
-  // Helper method to maintain DRY principle
-  // Update return type
   #orderQueuesByEnvs(envs: string[], snapshot: FairQueueSnapshot): Array<EnvQueues> {
     const queuesByEnv = snapshot.queues.reduce((acc, queue) => {
       if (!acc[queue.env]) {
@@ -231,11 +234,17 @@ export class FairDequeuingStrategy implements MarQSFairDequeueStrategy {
       if (queuesByEnv[envId]) {
         // Get ordered queues for this env
         const orderedQueues = this.#weightedRandomQueueOrder(queuesByEnv[envId]);
+
+        // Apply queue limit if maximumQueuePerEnvCount is set
+        const limitedQueues = this.options.maximumQueuePerEnvCount
+          ? orderedQueues.slice(0, this.options.maximumQueuePerEnvCount)
+          : orderedQueues;
+
         // Only add the env if it has queues
-        if (orderedQueues.length > 0) {
+        if (limitedQueues.length > 0) {
           acc.push({
             envId,
-            queues: orderedQueues.map((queue) => queue.id),
+            queues: limitedQueues.map((queue) => queue.id),
           });
         }
       }
@@ -512,6 +521,10 @@ export class FairDequeuingStrategy implements MarQSFairDequeueStrategy {
 
       span.setAttribute("queue_count", result.length);
 
+      if (result.length === this.options.parentQueueLimit) {
+        span.setAttribute("parent_queue_limit_reached", true);
+      }
+
       return result;
     });
   }

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

@@ -2189,6 +2189,7 @@ function getMarQSClient() {
       },
       reuseSnapshotCount: env.MARQS_REUSE_SNAPSHOT_COUNT,
       maximumEnvCount: env.MARQS_MAXIMUM_ENV_COUNT,
+      maximumQueuePerEnvCount: env.MARQS_MAXIMUM_QUEUE_PER_ENV_COUNT,
     }),
     envQueuePriorityStrategy: new FairDequeuingStrategy({
       tracer: tracer,