From b53fbdbbd2548ecd89c91be8d3833229eb6787ee Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sun, 5 Jul 2026 20:57:22 +0100 Subject: [PATCH] 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. --- .../run-engine/src/engine/systems/enqueueSystem.ts | 13 +++++++++---- .../run-engine/src/engine/tests/ttl.test.ts | 9 +++++++++ .../run-engine/src/run-queue/index.ts | 9 +++++++-- .../run-engine/src/run-queue/keyProducer.ts | 14 ++++++++------ 4 files changed, 33 insertions(+), 12 deletions(-) diff --git a/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts b/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts index 7ae551d7c..4b236aefc 100644 --- a/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts +++ b/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); diff --git a/internal-packages/run-engine/src/engine/tests/ttl.test.ts b/internal-packages/run-engine/src/engine/tests/ttl.test.ts index 949e47f85..e33b361ab 100644 --- a/internal-packages/run-engine/src/engine/tests/ttl.test.ts +++ b/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(); } diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 6f9211f42..4e6ca89d8 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/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): 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); diff --git a/internal-packages/run-engine/src/run-queue/keyProducer.ts b/internal-packages/run-engine/src/run-queue/keyProducer.ts index b185435f6..18a2727b7 100644 --- a/internal-packages/run-engine/src/run-queue/keyProducer.ts +++ b/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 {