fix(run-engine): anchor scheduling delay to the current queue stint on re-enqueue

The wait metric measured dequeue time minus the original trigger time
even when a run re-entered the queue after a waitpoint, checkpoint, or
pending version, so the whole wait or checkpoint duration showed up as
scheduling delay. Re-enqueues now anchor to the re-enqueue time while
first enqueues keep the trigger or delay anchor; queue ordering is
unchanged, so re-enqueued runs keep their original position. Nacked
runs never left the queue stint and keep the original anchor.

Also replaces the :ck: suffix regexes on user-controlled queue names
with indexOf slicing (identical semantics) to remove a polynomial
regex flagged by code scanning.

Author: ericallam

internal-packages/run-engine/src/engine/systems/enqueueSystem.ts

@@ -98,11 +98,16 @@ export class EnqueueSystem {
       // Force development runs to use the environment id as the worker queue.
       const workerQueue = env.type === "DEVELOPMENT" ? env.id : run.workerQueue;
 
-      const eligibleAtMs = (run.queueTimestamp ?? run.createdAt).getTime();
-      const timestamp = eligibleAtMs - run.priorityMs;
+      // Ordering keeps the run's original position; the scheduling-delay anchor is the
+      // trigger/delay time only on first enqueue (includeTtl). Re-enqueues anchor to now,
+      // else the wait metric absorbs the whole waitpoint/checkpoint duration.
+      const queuePositionMs = (run.queueTimestamp ?? run.createdAt).getTime();
+      const timestamp = queuePositionMs - run.priorityMs;
+      const eligibleAtMs = includeTtl ? queuePositionMs : Date.now();
 
-      // Include TTL only when explicitly requested (first enqueue from trigger).
-      // Re-enqueues (waitpoint, checkpoint, delayed, pending version) must not add TTL.
+      // Include TTL only when explicitly requested (first enqueue from trigger or the
+      // delayed-run system). Re-enqueues (waitpoint, checkpoint, pending version) must
+      // not add TTL.
       let ttlExpiresAt: number | undefined;
       if (includeTtl && run.ttl) {
         const expireAt = parseNaturalLanguageDuration(run.ttl);

internal-packages/run-engine/src/engine/tests/ttl.test.ts

@@ -293,7 +293,12 @@ describe("RunEngine ttl", () => {
         );
         assertNonNullable(messageAfterTrigger);
         expect(messageAfterTrigger.ttlExpiresAt).toBeDefined();
+        // First enqueue anchors the scheduling-delay clock at the trigger time.
+        expect(messageAfterTrigger.eligibleAtMs).toBe(
+          (run.queueTimestamp ?? run.createdAt).getTime()
+        );
 
+        const beforeReenqueue = Date.now();
         await engine.enqueueSystem.enqueueRun({
           run,
           env: authenticatedEnvironment,
@@ -308,6 +313,10 @@ describe("RunEngine ttl", () => {
         );
         assertNonNullable(messageAfterReenqueue);
         expect(messageAfterReenqueue.ttlExpiresAt).toBeUndefined();
+        // Re-enqueues anchor to now so the wait metric measures only this queue stint,
+        // while the ordering timestamp keeps the run's original position.
+        expect(messageAfterReenqueue.eligibleAtMs).toBeGreaterThanOrEqual(beforeReenqueue);
+        expect(messageAfterReenqueue.timestamp).toBe(messageAfterTrigger.timestamp);
       } finally {
         await engine.quit();
       }

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

@@ -603,7 +603,7 @@ export class RunQueue {
       const queuedResult = stats?.[i * 2];
       const runningResult = stats?.[i * 2 + 1];
       return {
-        concurrencyKey: member.match(/:ck:(.+)$/)?.[1] ?? "",
+        concurrencyKey: this.#concurrencyKeyFromQueue(member) ?? "",
         queued: queuedResult && !queuedResult[0] ? ((queuedResult[1] as number) ?? 0) : 0,
         running: runningResult && !runningResult[0] ? ((runningResult[1] as number) ?? 0) : 0,
         oldestEnqueuedAt: score,
@@ -2033,6 +2033,11 @@ export class RunQueue {
     this.options.queueMetrics?.emitGauge(queue, fields);
   }
 
+  #concurrencyKeyFromQueue(queue: string): string | undefined {
+    const idx = queue.indexOf(":ck:");
+    return idx === -1 || idx + 4 >= queue.length ? undefined : queue.slice(idx + 4);
+  }
+
   #emitQueueMetric(shardKey: string, fields: Record<string, string | number>): void {
     // Counters roll up per BASE queue: normalize the CK-qualified queue to its base so all
     // concurrency keys share one monotonic odometer (and one shard/order key), matching the
@@ -2042,7 +2047,7 @@ export class RunQueue {
     let baseFields = fields;
     if (typeof fields.q === "string") {
       baseFields = { ...fields, q: this.keys.baseQueueKeyFromQueue(fields.q) };
-      const ck = fields.q.match(/:ck:(.+)$/)?.[1];
+      const ck = this.#concurrencyKeyFromQueue(fields.q);
       if (ck && ck !== "*") baseFields.ck = ck;
     }
     this.options.queueMetrics?.emit(baseQueue, baseFields);

internal-packages/run-engine/src/run-queue/keyProducer.ts

@@ -141,8 +141,7 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer {
   }
 
   queueConcurrencyLimitKeyFromQueue(queue: string) {
-    const concurrencyQueueName = queue.replace(/:ck:.+$/, "");
-    return `${concurrencyQueueName}:${constants.CONCURRENCY_LIMIT_PART}`;
+    return `${this.baseQueueKeyFromQueue(queue)}:${constants.CONCURRENCY_LIMIT_PART}`;
   }
 
   queueCurrentConcurrencyKeyFromQueue(queue: string) {
@@ -313,12 +312,14 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer {
   }
 
   ckIndexKeyFromQueue(queue: string): string {
-    const baseQueue = queue.replace(/:ck:.+$/, "");
-    return `${baseQueue}:${constants.CK_INDEX_PART}`;
+    return `${this.baseQueueKeyFromQueue(queue)}:${constants.CK_INDEX_PART}`;
   }
 
+  // indexOf instead of /:ck:.+$/ (queue names are user-controlled; polynomial regex).
+  // Only strips when at least one character follows ":ck:", matching the old semantics.
   baseQueueKeyFromQueue(queue: string): string {
-    return queue.replace(/:ck:.+$/, "");
+    const idx = queue.indexOf(":ck:");
+    return idx === -1 || idx + 4 >= queue.length ? queue : queue.slice(0, idx);
   }
 
   queueLengthCounterKey(env: RunQueueKeyProducerEnvironment, queue: string): string {
@@ -342,7 +343,8 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer {
   }
 
   toCkWildcard(queue: string): string {
-    return queue.replace(/:ck:.+$/, ":ck:*");
+    const base = this.baseQueueKeyFromQueue(queue);
+    return base === queue ? queue : `${base}:ck:*`;
   }
 
   descriptorFromQueue(queue: string): QueueDescriptor {