Additional logging to help track down possible issue with MarQS selection algo

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

@@ -645,17 +645,21 @@ export class MarQS {
     return this.#trace(
       "getRandomQueueFromParentQueue",
       async (span) => {
+        span.setAttribute("consumerId", consumerId);
+
         const { range } = await queuePriorityStrategy.nextCandidateSelection(
           parentQueue,
           consumerId
         );
 
-        const queues = await this.#getChildQueuesWithScores(parentQueue, range);
+        const queues = await this.#getChildQueuesWithScores(parentQueue, range, span);
+        span.setAttribute("queueCount", queues.length);
 
         const queuesWithScores = await this.#calculateQueueScores(queues, calculateCapacities);
+        span.setAttribute("queuesWithScoresCount", queuesWithScores.length);
 
         // We need to priority shuffle here to ensure all workers aren't just working on the highest priority queue
-        const choice = this.queuePriorityStrategy.chooseQueue(
+        const { choice, nextRange } = this.queuePriorityStrategy.chooseQueue(
           queuesWithScores,
           parentQueue,
           consumerId,
@@ -668,17 +672,20 @@ export class MarQS {
         span.setAttributes({
           ...flattenAttributes(queuesWithScores, "marqs.queuesWithScores"),
         });
-        span.setAttribute("nextRange.offset", range.offset);
-        span.setAttribute("nextRange.count", range.count);
-        span.setAttribute("queueCount", queues.length);
+        span.setAttribute("range.offset", range.offset);
+        span.setAttribute("range.count", range.count);
+        span.setAttribute("nextRange.offset", nextRange.offset);
+        span.setAttribute("nextRange.count", nextRange.count);
 
-        if (this.options.verbose) {
+        if (this.options.verbose || nextRange.offset > 0) {
           if (typeof choice === "string") {
             logger.debug(`[${this.name}] getRandomQueueFromParentQueue`, {
               queues,
               queuesWithScores,
-              nextRange: range,
+              range,
+              nextRange,
               queueCount: queues.length,
+              queuesWithScoresCount: queuesWithScores.length,
               queueChoice: choice,
               consumerId,
             });
@@ -686,8 +693,10 @@ export class MarQS {
             logger.debug(`[${this.name}] getRandomQueueFromParentQueue`, {
               queues,
               queuesWithScores,
-              nextRange: range,
+              range,
+              nextRange,
               queueCount: queues.length,
+              queuesWithScoresCount: queuesWithScores.length,
               noQueueChoice: true,
               consumerId,
             });
@@ -752,7 +761,8 @@ export class MarQS {
 
   async #getChildQueuesWithScores(
     key: string,
-    range: QueueRange
+    range: QueueRange,
+    span?: Span
   ): Promise<Array<{ value: string; score: number }>> {
     const valuesWithScores = await this.redis.zrangebyscore(
       key,
@@ -763,6 +773,12 @@ export class MarQS {
       range.offset,
       range.count
     );
+
+    span?.setAttribute("zrangebyscore.valuesWithScores.rawLength", valuesWithScores.length);
+    span?.setAttributes({
+      ...flattenAttributes(valuesWithScores, "zrangebyscore.valuesWithScores.rawValues"),
+    });
+
     const result: Array<{ value: string; score: number }> = [];
 
     for (let i = 0; i < valuesWithScores.length; i += 2) {

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

@@ -1,4 +1,3 @@
-import { RedisOptions } from "ioredis";
 import { nanoid } from "nanoid";
 import {
   MarQSQueuePriorityStrategy,
@@ -32,7 +31,7 @@ export class SimpleWeightedChoiceStrategy implements MarQSQueuePriorityStrategy
     parentQueue: string,
     consumerId: string,
     previousRange: QueueRange
-  ): PriorityStrategyChoice {
+  ): { choice: PriorityStrategyChoice; nextRange: QueueRange } {
     const filteredQueues = filterQueuesAtCapacity(queues);
 
     if (queues.length === this.options.queueSelectionCount) {
@@ -40,6 +39,7 @@ export class SimpleWeightedChoiceStrategy implements MarQSQueuePriorityStrategy
         offset: previousRange.offset + this.options.queueSelectionCount,
         count: this.options.queueSelectionCount,
       };
+
       // If all queues are at capacity, and we were passed the max number of queues, then we will slide the window "to the right"
       this._nextRangesByParentQueue.set(`${consumerId}:${parentQueue}`, nextRange);
     } else {
@@ -47,12 +47,20 @@ export class SimpleWeightedChoiceStrategy implements MarQSQueuePriorityStrategy
     }
 
     if (filteredQueues.length === 0) {
-      return { abort: true };
+      return {
+        choice: { abort: true },
+        nextRange: this.nextRangeForParentQueue(parentQueue, consumerId),
+      };
     }
 
     const queueWeights = this.#calculateQueueWeights(filteredQueues);
 
-    return weightedRandomChoice(queueWeights);
+    const choice = weightedRandomChoice(queueWeights);
+
+    return {
+      choice,
+      nextRange: this.nextRangeForParentQueue(parentQueue, consumerId),
+    };
   }
 
   async nextCandidateSelection(
@@ -117,8 +125,8 @@ export class NoopWeightedChoiceStrategy implements MarQSQueuePriorityStrategy {
     queues: QueueWithScores[],
     parentQueue: string,
     selectionId: string
-  ): PriorityStrategyChoice {
-    return { abort: true };
+  ): { choice: PriorityStrategyChoice; nextRange: QueueRange } {
+    return { choice: { abort: true }, nextRange: { offset: 0, count: 0 } };
   }
 
   nextCandidateSelection(parentQueue: string): Promise<{ range: QueueRange; selectionId: string }> {

apps/webapp/app/v3/marqs/types.ts

@@ -65,7 +65,7 @@ export interface MarQSQueuePriorityStrategy {
     parentQueue: string,
     consumerId: string,
     previousRange: QueueRange
-  ): PriorityStrategyChoice;
+  ): { choice: PriorityStrategyChoice; nextRange: QueueRange };
 
   /**
    * This function is called to get the next candidate selection for the queue