feat(redis-worker,run-engine): report the age of the oldest waiting batch item

Adds a metric for how long the oldest batch item has been waiting to be turned into a run, and makes the batch queue wait histograms cover waits longer than 10 seconds.

FairQueue now emits `<name>.oldest_message_age`, a per-shard gauge in ms of the oldest message that is due but not yet dispatched (0 when nothing is due). It reads the head of each dispatch shard, and of the legacy master queue shard while it drains, so collection costs two `ZRANGEBYSCORE ... LIMIT 0 1` calls per shard. For the batch queue this is `batch-queue.oldest_message_age`. Unlike throughput or queue length, it shows a backlog the same way at any traffic level, so it is the signal to alert on.

`batch_queue.item_queue_time` and `batch_queue.batch_processing_duration` used the default histogram buckets, which end at 10s, so every longer wait landed in the overflow bucket. They now have buckets up to 30 minutes.

Mono-RevId: 31e0228bfd25ce7fb3bc26ded7c0f95e3159f554

Author: 0ski

internal-packages/run-engine/src/batch-queue/index.ts

@@ -54,6 +54,12 @@ export type { BatchQueueOptions, CompleteBatchResult, InitializeBatchOptions } f
 // Redis key for environment concurrency limits
 const ENV_CONCURRENCY_KEY_PREFIX = "batch:env_concurrency";
 
+// The default histogram buckets stop at 10s, which hides multi-minute backlogs
+const QUEUE_WAIT_BUCKETS_MS = [
+  10, 50, 100, 250, 500, 1_000, 2_500, 5_000, 10_000, 30_000, 60_000, 120_000, 300_000, 600_000,
+  1_800_000,
+];
+
 // Single worker queue ID for all batch items
 // BatchQueue uses a single shared worker queue - FairQueue handles fair scheduling,
 // then all messages are routed to this queue for BatchQueue's own consumer loop.
@@ -601,6 +607,7 @@ export class BatchQueue {
       {
         description: "Duration from batch creation to completion",
         unit: "ms",
+        advice: { explicitBucketBoundaries: QUEUE_WAIT_BUCKETS_MS },
       }
     );
 
@@ -612,6 +619,7 @@ export class BatchQueue {
     this.itemQueueTimeHistogram = meter.createHistogram("batch_queue.item_queue_time", {
       description: "Time from item enqueue to processing start",
       unit: "ms",
+      advice: { explicitBucketBoundaries: QUEUE_WAIT_BUCKETS_MS },
     });
 
     this.rateLimitDeniedCounter = meter.createCounter("batch_queue.rate_limit_denied", {

packages/redis-worker/src/fair-queue/index.ts

@@ -263,6 +263,9 @@ export class FairQueue<TPayloadSchema extends AnyZodSchema = z.ZodUnknown> {
       getDLQLength: async (tenantId: string) => {
         return await this.getDeadLetterQueueLength(tenantId);
       },
+      getOldestMessageAge: async (shardId: number) => {
+        return await this.getOldestMessageAge(shardId);
+      },
       shardCount: this.shardCount,
       observedTenants: options?.observedTenants,
     });
@@ -752,6 +755,24 @@ export class FairQueue<TPayloadSchema extends AnyZodSchema = z.ZodUnknown> {
     return dispatchCount + legacyCount;
   }
 
+  /**
+   * Get the age in ms of the oldest due message not yet completed in a shard, or 0 if none.
+   * Claims don't refresh the index scores, so claimed messages count until their queue next completes or empties.
+   */
+  async getOldestMessageAge(shardId: number): Promise<number> {
+    const now = Date.now();
+    const [[oldestTenant], [oldestLegacyQueue]] = await Promise.all([
+      this.tenantDispatch.getTenantsFromShard(shardId, 1, now),
+      this.masterQueue.getQueuesFromShard(shardId, 1, now),
+    ]);
+    const oldestScore = Math.min(
+      oldestTenant?.score ?? Infinity,
+      oldestLegacyQueue?.score ?? Infinity
+    );
+    if (!Number.isFinite(oldestScore)) return 0;
+    return Math.max(0, now - oldestScore);
+  }
+
   /**
    * Get total in-flight message count.
    */

packages/redis-worker/src/fair-queue/telemetry.ts

@@ -59,6 +59,7 @@ export interface FairQueueMetrics {
   dispatchLength: ObservableGauge;
   inflightCount: ObservableGauge;
   dlqLength: ObservableGauge;
+  oldestMessageAge: ObservableGauge;
 }
 
 /**
@@ -254,6 +255,7 @@ export class FairQueueTelemetry {
     getDispatchLength?: (shardId: number) => Promise<number>;
     getInflightCount?: (shardId: number) => Promise<number>;
     getDLQLength?: (tenantId: string) => Promise<number>;
+    getOldestMessageAge?: (shardId: number) => Promise<number>;
     shardCount?: number;
     observedQueues?: string[];
     observedTenants?: string[];
@@ -320,6 +322,20 @@ export class FairQueueTelemetry {
       });
     }
 
+    if (callbacks.getOldestMessageAge && callbacks.shardCount) {
+      const getOldestMessageAge = callbacks.getOldestMessageAge;
+      const shardCount = callbacks.shardCount;
+
+      this.metrics.oldestMessageAge.addCallback(async (observableResult) => {
+        for (let shardId = 0; shardId < shardCount; shardId++) {
+          const age = await getOldestMessageAge(shardId);
+          observableResult.observe(age, {
+            [FairQueueAttributes.SHARD_ID]: shardId.toString(),
+          });
+        }
+      });
+    }
+
     // DLQ length gauge
     if (callbacks.getDLQLength && callbacks.observedTenants) {
       const getDLQLength = callbacks.getDLQLength;
@@ -446,6 +462,11 @@ export class FairQueueTelemetry {
         description: "Number of messages in dead letter queue",
         unit: "messages",
       }),
+      oldestMessageAge: this.meter.createObservableGauge(`${this.name}.oldest_message_age`, {
+        description:
+          "Age of the oldest due message not yet completed in a shard, including claimed messages until their queue next completes or empties",
+        unit: "ms",
+      }),
     };
   }
 }

packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts

@@ -1113,6 +1113,62 @@ describe("FairQueue", () => {
 
       await queue.close();
     });
+
+    redisTest(
+      "should report the age of the oldest due message",
+      { timeout: 10000 },
+      async ({ redisOptions }) => {
+        keys = new DefaultFairQueueKeyProducer({ prefix: "test" });
+
+        const scheduler = new DRRScheduler({
+          redis: redisOptions,
+          keys,
+          quantum: 10,
+          maxDeficit: 100,
+        });
+
+        const queue = new FairQueue({
+          redis: redisOptions,
+          keys,
+          scheduler,
+          shardCount: 1,
+          startConsumers: false,
+          workerQueue: {
+            resolveWorkerQueue: () => TEST_WORKER_QUEUE_ID,
+          },
+        });
+
+        expect(await queue.getOldestMessageAge(0)).toBe(0);
+
+        await queue.enqueue({
+          queueId: "tenant:t1:queue:q1",
+          tenantId: "t1",
+          payload: { value: "scheduled" },
+          timestamp: Date.now() + 60_000,
+        });
+
+        expect(await queue.getOldestMessageAge(0)).toBe(0);
+
+        await queue.enqueue({
+          queueId: "tenant:t2:queue:q1",
+          tenantId: "t2",
+          payload: { value: "recent" },
+          timestamp: Date.now() - 5_000,
+        });
+        await queue.enqueue({
+          queueId: "tenant:t3:queue:q1",
+          tenantId: "t3",
+          payload: { value: "old" },
+          timestamp: Date.now() - 120_000,
+        });
+
+        const age = await queue.getOldestMessageAge(0);
+        expect(age).toBeGreaterThanOrEqual(120_000);
+        expect(age).toBeLessThan(125_000);
+
+        await queue.close();
+      }
+    );
   });
 
   describe("two-stage processing with concurrency limits", () => {