Add some randomization to the list of queues within each evn

Author: ericallam

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

@@ -20,6 +20,14 @@ export type FairDequeuingStrategyBiases = {
    * 0 = no bias, 1 = full bias based on available capacity
    */
   availableCapacityBias: number;
+
+  /**
+   * Controls randomization of queue ordering within environments
+   * 0 = strict age-based ordering (oldest first)
+   * 1 = completely random ordering
+   * Values between 0-1 blend between age-based and random ordering
+   */
+  queueAgeRandomization: number;
 };
 
 export type FairDequeuingStrategyOptions = {
@@ -57,6 +65,11 @@ type WeightedEnv = {
   weight: number;
 };
 
+type WeightedQueue = {
+  queue: FairQueue;
+  weight: number;
+};
+
 const emptyFairQueueSnapshot: FairQueueSnapshot = {
   id: "empty",
   orgs: {},
@@ -67,6 +80,7 @@ const emptyFairQueueSnapshot: FairQueueSnapshot = {
 const defaultBiases: FairDequeuingStrategyBiases = {
   concurrencyLimitBias: 0,
   availableCapacityBias: 0,
+  queueAgeRandomization: 0, // Default to completely age-based ordering
 };
 
 export class FairDequeuingStrategy implements MarQSFairDequeueStrategy {
@@ -213,14 +227,65 @@ export class FairDequeuingStrategy implements MarQSFairDequeueStrategy {
 
     const queues = envs.reduce((acc, envId) => {
       if (queuesByEnv[envId]) {
-        acc.push(...queuesByEnv[envId].sort((a, b) => b.age - a.age));
+        // Instead of sorting by age, use weighted random selection
+        acc.push(...this.#weightedRandomQueueOrder(queuesByEnv[envId]));
       }
       return acc;
     }, [] as Array<FairQueue>);
 
     return queues.map((queue) => queue.id);
   }
 
+  #weightedRandomQueueOrder(queues: FairQueue[]): FairQueue[] {
+    if (queues.length <= 1) return queues;
+
+    const biases = this.options.biases ?? defaultBiases;
+
+    // When queueAgeRandomization is 0, use strict age-based ordering
+    if (biases.queueAgeRandomization === 0) {
+      return [...queues].sort((a, b) => b.age - a.age);
+    }
+
+    // Find the maximum age for normalization
+    const maxAge = Math.max(...queues.map((q) => q.age));
+
+    // Calculate weights for each queue
+    const weightedQueues: WeightedQueue[] = queues.map((queue) => {
+      // Normalize age to be between 0 and 1
+      const normalizedAge = queue.age / maxAge;
+
+      // Calculate weight: combine base weight with configurable age influence
+      const baseWeight = 1;
+      const weight = baseWeight + normalizedAge * biases.queueAgeRandomization;
+
+      return { queue, weight };
+    });
+
+    // Perform weighted random selection for ordering
+    const result: FairQueue[] = [];
+    let remainingQueues = [...weightedQueues];
+    let totalWeight = remainingQueues.reduce((sum, wq) => sum + wq.weight, 0);
+
+    while (remainingQueues.length > 0) {
+      let random = this._rng() * totalWeight;
+      let index = 0;
+
+      // Find queue based on weighted random selection
+      while (random > 0 && index < remainingQueues.length) {
+        random -= remainingQueues[index].weight;
+        index++;
+      }
+      index = Math.max(0, index - 1);
+
+      // Add selected queue to result and remove from remaining
+      result.push(remainingQueues[index].queue);
+      totalWeight -= remainingQueues[index].weight;
+      remainingQueues.splice(index, 1);
+    }
+
+    return result;
+  }
+
   #shuffle<T>(array: Array<T>): Array<T> {
     let currentIndex = array.length;
     let temporaryValue;

apps/webapp/test/fairDequeuingStrategy.test.ts

@@ -535,6 +535,7 @@ describe("FairDequeuingStrategy", () => {
             biases: {
               concurrencyLimitBias: 0.8,
               availableCapacityBias: 0.5,
+              queueAgeRandomization: 0.0,
             },
           })
       );
@@ -593,4 +594,99 @@ describe("FairDequeuingStrategy", () => {
       expect(highLimitPercentage).toBeGreaterThan(lowLimitPercentage);
     }
   );
+
+  redisTest("should respect ageInfluence parameter for queue ordering", async ({ redis }) => {
+    const keyProducer = createKeyProducer("test");
+    const now = Date.now();
+
+    // Setup queues with different ages in the same environment
+    const queueAges = [
+      { id: "queue-1", age: 5000 }, // oldest
+      { id: "queue-2", age: 3000 },
+      { id: "queue-3", age: 1000 }, // newest
+    ];
+
+    // Helper function to run iterations with a specific age influence
+    async function runWithQueueAgeRandomization(queueAgeRandomization: number) {
+      const strategy = new FairDequeuingStrategy({
+        tracer,
+        redis,
+        keys: keyProducer,
+        defaultOrgConcurrency: 10,
+        defaultEnvConcurrency: 5,
+        parentQueueLimit: 100,
+        checkForDisabledOrgs: true,
+        seed: "fixed-seed",
+        biases: {
+          concurrencyLimitBias: 0,
+          availableCapacityBias: 0,
+          queueAgeRandomization,
+        },
+      });
+
+      const positionCounts: Record<string, number[]> = {
+        "queue-1": [0, 0, 0],
+        "queue-2": [0, 0, 0],
+        "queue-3": [0, 0, 0],
+      };
+
+      const iterations = 1000;
+      for (let i = 0; i < iterations; i++) {
+        const result = await strategy.distributeFairQueuesFromParentQueue(
+          "parent-queue",
+          "consumer-1"
+        );
+
+        result.forEach((queueId, position) => {
+          const baseQueueId = queueId.split(":").pop()!;
+          positionCounts[baseQueueId][position]++;
+        });
+      }
+
+      return positionCounts;
+    }
+
+    // Setup test data
+    for (const { id, age } of queueAges) {
+      await setupQueue({
+        redis,
+        keyProducer,
+        parentQueue: "parent-queue",
+        score: now - age,
+        queueId: id,
+        orgId: "org-1",
+        envId: "env-1",
+      });
+    }
+
+    await setupConcurrency({
+      redis,
+      keyProducer,
+      org: { id: "org-1", currentConcurrency: 0, limit: 10 },
+      env: { id: "env-1", currentConcurrency: 0, limit: 5 },
+    });
+
+    // Test with different age influence values
+    const strictAge = await runWithQueueAgeRandomization(0); // Strict age-based ordering
+    const mixed = await runWithQueueAgeRandomization(0.5); // Mix of age and random
+    const fullyRandom = await runWithQueueAgeRandomization(1); // Completely random
+
+    console.log("Distribution with strict age ordering (0.0):", strictAge);
+    console.log("Distribution with mixed ordering (0.5):", mixed);
+    console.log("Distribution with random ordering (1.0):", fullyRandom);
+
+    // With strict age ordering (0.0), oldest should always be first
+    expect(strictAge["queue-1"][0]).toBe(1000); // Always in first position
+    expect(strictAge["queue-3"][0]).toBe(0); // Never in first position
+
+    // With fully random (1.0), positions should still allow for some age bias
+    const randomFirstPositionSpread = Math.abs(
+      fullyRandom["queue-1"][0] - fullyRandom["queue-3"][0]
+    );
+    expect(randomFirstPositionSpread).toBeLessThan(200); // Allow for larger spread in distribution
+
+    // With mixed (0.5), should show preference for age but not absolute
+    expect(mixed["queue-1"][0]).toBeGreaterThan(mixed["queue-3"][0]); // Older preferred
+    expect(mixed["queue-3"][0]).toBeGreaterThan(0); // But newer still gets chances
+  });
 });