Skip to content

Commit 9a541ef

Browse files
committed
fix(run-engine): the repair clear drains the group set for keyless runs
clearMessageFromConcurrencySets now mirrors the group SREM like every other release path, so a stuck keyless run cleared by the repair sweep frees its total-concurrency slot instead of phantom-holding it until a saturation reconcile. The plain gauge test pins the new tail: zeroed CK fields plus the total pair.
1 parent 2130b17 commit 9a541ef

2 files changed

Lines changed: 22 additions & 8 deletions

File tree

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

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3112,8 +3112,9 @@ export class RunQueue {
31123112
const messageKey = this.keys.messageKey(orgId, messageId);
31133113
/**
31143114
* Callers pass the bare TaskRun queue name plus its concurrencyKey; the run's
3115-
* slots live on the ck variant, and the tracked clear additionally mirrors the
3116-
* group set and counters that only keyed queues maintain.
3115+
* slots live on the ck variant. Both variants mirror the per-base-queue group
3116+
* set (keyed and keyless admits populate it); the tracked clear additionally
3117+
* maintains the counters that only keyed queues keep.
31173118
*/
31183119
const fullQueue = concurrencyKey ? this.keys.queueKey(env, queue, concurrencyKey) : queue;
31193120
const queueCurrentConcurrencyKey = this.keys.queueCurrentConcurrencyKey(
@@ -3159,6 +3160,7 @@ export class RunQueue {
31593160
queueCurrentDequeuedKey,
31603161
envCurrentDequeuedKey,
31613162
messageKey,
3163+
this.keys.queueGroupConcurrencyKeyFromQueue(fullQueue),
31623164
messageId,
31633165
this.options.redis.keyPrefix ?? ""
31643166
);
@@ -6238,22 +6240,27 @@ return results
62386240
});
62396241

62406242
this.redis.defineCommand("clearMessageFromConcurrencySets", {
6241-
numberOfKeys: 5,
6243+
numberOfKeys: 6,
62426244
lua: `
62436245
-- Keys:
62446246
local queueCurrentConcurrencyKey = KEYS[1]
62456247
local envCurrentConcurrencyKey = KEYS[2]
62466248
local queueCurrentDequeuedKey = KEYS[3]
62476249
local envCurrentDequeuedKey = KEYS[4]
62486250
local messageKey = KEYS[5]
6251+
local groupConcurrencyKey = KEYS[6]
62496252
62506253
-- Args:
62516254
local messageId = ARGV[1]
62526255
local keyPrefix = ARGV[2]
62536256
${QUEUE_GATES_LUA_HELPERS}
62546257
6255-
-- Update the concurrency keys
6256-
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
6258+
-- Update the concurrency keys. The groupConcurrency SREM mirrors the base SREM
6259+
-- unconditionally (no flag check) so a disabled flag still drains the group set.
6260+
local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId)
6261+
if removedFromCurrentConcurrency == 1 then
6262+
redis.call('SREM', groupConcurrencyKey, messageId)
6263+
end
62576264
redis.call('SREM', envCurrentConcurrencyKey, messageId)
62586265
redis.call('SREM', queueCurrentDequeuedKey, messageId)
62596266
redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -6473,6 +6480,7 @@ declare module "@internal/redis" {
64736480
queueCurrentDequeuedKey: string,
64746481
envCurrentDequeuedKey: string,
64756482
messageKey: string,
6483+
groupConcurrencyKey: string,
64766484
// args
64776485
messageId: string,
64786486
keyPrefix: string,

‎internal-packages/run-engine/src/run-queue/metrics.test.ts‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -156,9 +156,15 @@ describe("RunQueue queue-metrics emission", () => {
156156
for (const f of ["ql", "cc", "lim", "eql", "ec", "elim", "thr"]) {
157157
expect(gauge!.fields[f]).toBeDefined();
158158
}
159-
// Non-CK scripts keep the 7-field gauge (no CK-health tail).
160-
expect(gauge!.fields.ckq).toBeUndefined();
161-
expect(gauge!.fields.ckw).toBeUndefined();
159+
/**
160+
* Non-CK scripts emit the full gauge tail too: zeroed CK-health fields (a base
161+
* queue has no CK variants) followed by the total running/limit pair, so a
162+
* keyless queue with a total limit still charts total concurrency.
163+
*/
164+
expect(gauge!.fields.ckq).toBe("0");
165+
expect(gauge!.fields.ckw).toBe("0");
166+
expect(gauge!.fields.tcc).toBeDefined();
167+
expect(gauge!.fields.tlim).toBeDefined();
162168

163169
// Pins the dequeue script's sample-at-return wrapper: only the dequeue emits the
164170
// post-admission reading (running 1, queued 0); the enqueue gauge sees the inverse.

0 commit comments

Comments
 (0)