fix: unflatten crash on large numeric keys and fair-queue age ranking

.changeset/fair-queue-oldest-tenant-priority.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/redis-worker": patch
+---
+
+Fix fair queue tenant selection so that, when a maximum tenant count is set, the tenants that have been waiting the longest are picked first instead of being ranked by their raw timestamp.

.changeset/unflatten-attributes-large-numeric-keys.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Fix a crash when unflattening attributes that hold an object whose keys are all large numbers, such as millisecond timestamps. Those values now come back as an object instead of throwing an "Invalid array length" error.

packages/core/src/v3/utils/flattenAttributes.ts

@@ -5,6 +5,10 @@ export const CIRCULAR_REFERENCE_SENTINEL = "$@circular((";
 
 const DEFAULT_MAX_DEPTH = 128;
 
+// The largest value a JS array length can hold. A numeric key at or above this
+// is not a real array index, so we never try to build an array that big.
+const MAX_ARRAY_INDEX = 2 ** 32 - 1;
+
 // This property name would let a crafted key walk into Object.prototype during
 // reconstruction and pollute the shared process.
 const PROTOTYPE_POLLUTION_KEY = "__proto__";
@@ -342,13 +346,20 @@ export function unflattenAttributes(
   // Convert the result to an array if all top-level keys are numeric indices.
   // Guard against an empty result (e.g. every key was skipped as unsafe), which
   // would otherwise produce Array(-Infinity) and throw.
-  if (Object.keys(result).length > 0 && Object.keys(result).every((k) => /^\d+$/.test(k))) {
-    const maxIndex = Math.max(...Object.keys(result).map((k) => parseInt(k)));
-    const arrayResult = Array(maxIndex + 1);
-    for (const key in result) {
-      arrayResult[parseInt(key)] = result[key];
+  const topLevelKeys = Object.keys(result);
+  if (topLevelKeys.length > 0 && topLevelKeys.every((k) => /^\d+$/.test(k))) {
+    const maxIndex = topLevelKeys.reduce((max, k) => Math.max(max, parseInt(k)), 0);
+    // Only rebuild an array when every key is a real array index (< 2^32 - 1). A
+    // larger numeric key, like a millisecond timestamp used as an object key, is
+    // not an array index, so keep the object form instead of throwing "Invalid
+    // array length" or allocating a huge array.
+    if (maxIndex < MAX_ARRAY_INDEX) {
+      const arrayResult = Array(maxIndex + 1);
+      for (const key of topLevelKeys) {
+        arrayResult[parseInt(key)] = result[key];
+      }
+      return arrayResult as any;
     }
-    return arrayResult as any;
   }
 
   return result;

packages/core/test/flattenAttributes.test.ts

@@ -707,4 +707,20 @@ describe("unflattenAttributes", () => {
       a: { b: ["indexed"] },
     });
   });
+
+  it("keeps large numeric keys as an object instead of throwing Invalid array length", () => {
+    expect(() => unflattenAttributes({ "1699999999999": "value" })).not.toThrow();
+    expect(unflattenAttributes({ "1699999999999": "value" })).toEqual({
+      "1699999999999": "value",
+    });
+  });
+
+  it("keeps a nested object of large numeric keys as an object", () => {
+    expect(() =>
+      unflattenAttributes({ "a.1699999999999": "x", "a.1700000000000": "y" })
+    ).not.toThrow();
+    expect(unflattenAttributes({ "a.1699999999999": "x", "a.1700000000000": "y" })).toEqual({
+      a: { "1699999999999": "x", "1700000000000": "y" },
+    });
+  });
 });

packages/redis-worker/src/fair-queue/schedulers/weighted.ts

@@ -150,7 +150,7 @@ export class WeightedScheduler extends BaseScheduler {
 
     // Apply maximum tenant count if configured
     if (this.maximumTenantCount > 0) {
-      rawQueues = this.#selectTopTenantQueues(rawQueues);
+      rawQueues = this.#selectTopTenantQueues(rawQueues, now);
     }
 
     // Build tenant data
@@ -230,7 +230,7 @@ export class WeightedScheduler extends BaseScheduler {
     return queues;
   }
 
-  #selectTopTenantQueues(queues: QueueWithScore[]): QueueWithScore[] {
+  #selectTopTenantQueues(queues: QueueWithScore[], now: number): QueueWithScore[] {
     // Group by tenant and calculate average age
     const queuesByTenant = new Map<string, QueueWithScore[]>();
     for (const queue of queues) {
@@ -239,9 +239,11 @@ export class WeightedScheduler extends BaseScheduler {
       queuesByTenant.set(queue.tenantId, tenantQueues);
     }
 
-    // Calculate average age per tenant
+    // Calculate average age per tenant. A queue's score is its oldest message
+    // timestamp, so age is now - score. Older queues have a higher age and
+    // should get more weight when we pick the top tenants.
     const tenantAges = Array.from(queuesByTenant.entries()).map(([tenantId, tQueues]) => {
-      const avgAge = tQueues.reduce((sum, q) => sum + q.score, 0) / tQueues.length;
+      const avgAge = tQueues.reduce((sum, q) => sum + (now - q.score), 0) / tQueues.length;
       return { tenantId, avgAge };
     });