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.
This commit is contained in:
Eric Allam
2026-07-05 20:57:22 +01:00
parent c17b1cd4b6
commit b53fbdbbd2
4 changed files with 33 additions and 12 deletions
@@ -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);
@@ -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();
}
@@ -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);
@@ -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 {