diff --git a/apps/webapp/app/v3/marqs/index.server.ts b/apps/webapp/app/v3/marqs/index.server.ts index 3d84fefb9..159ccd3a2 100644 --- a/apps/webapp/app/v3/marqs/index.server.ts +++ b/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> { 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) { diff --git a/apps/webapp/app/v3/marqs/simpleWeightedPriorityStrategy.server.ts b/apps/webapp/app/v3/marqs/simpleWeightedPriorityStrategy.server.ts index c2e42a479..b53e984e9 100644 --- a/apps/webapp/app/v3/marqs/simpleWeightedPriorityStrategy.server.ts +++ b/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 }> { diff --git a/apps/webapp/app/v3/marqs/types.ts b/apps/webapp/app/v3/marqs/types.ts index 305d6a14e..fe5331b35 100644 --- a/apps/webapp/app/v3/marqs/types.ts +++ b/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