@@ -141,14 +141,20 @@ const QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA = createMetricsGaugeComputeLua({
141141} ) ;
142142
143143/** Injected queue-metrics stream emitter; all calls are no-ops when metrics are disabled. */
144- // Insertion points the virtual-time variants add to the ck scripts. The flag-off build of
145- // each script passes none of them, so there is one copy of the shared Lua rather than two
146- // that can drift apart.
147- type CkVtimeParts = Partial < Record < string , string > > ;
144+ // Slots the two builds of each ck script fill differently, so the shared Lua is written once
145+ // rather than copied. Each builder names its own slots: a key that is not one of them is a
146+ // compile error, because a silently-empty slot is the drift this exists to prevent.
147+ //
148+ // Most slots are insertion points the virtual-time build adds to and the flag-off build
149+ // leaves empty. Two builders are not like that. ckEnqueueLua and ckDequeueLua have slots the
150+ // FLAG-OFF build fills too, and in ckDequeueLua some slots open Lua blocks that the shared
151+ // text below them closes, so a slot's block depth is part of its contract and neither build
152+ // renders valid Lua on its own. Check both renders when editing those, never one.
153+ type CkParts < K extends string > = Partial < Record < K , string > > ;
148154
149155// Shared by the ck dead-letter command and its virtual-time variant. Called with no parts it
150156// renders the flag-off script exactly, so that build stays identical to the one in production.
151- const ckDeadLetterLua = ( v : CkVtimeParts ) => `
157+ const ckDeadLetterLua = ( v : CkParts < "keys" | "args" | "drainPark" > ) => `
152158-- Keys:
153159local masterQueueKey = KEYS[1]
154160local messageKey = KEYS[2]
222228` ;
223229
224230// Shared by the ck ack command and its virtual-time variant.
225- const ckAcknowledgeLua = ( v : CkVtimeParts ) => `
231+ const ckAcknowledgeLua = ( v : CkParts < "keys" | "args" | "drainPark" > ) => `
226232-- Keys:
227233local masterQueueKey = KEYS[1]
228234local messageKey = KEYS[2]
303309` ;
304310
305311// Shared by the ck nack command and its virtual-time variant.
306- const ckNackLua = ( v : CkVtimeParts ) => `
312+ const ckNackLua = ( v : CkParts < "keys" | "args" | "register" > ) => `
307313-- Keys:
308314local masterQueueKey = KEYS[1]
309315local messageKey = KEYS[2]
394400` ;
395401
396402// Shared by the ck enqueue command and its virtual-time variant.
397- const ckEnqueueLua = ( v : CkVtimeParts ) => `
403+ const ckEnqueueLua = ( v : CkParts < "keys" | "args" | "fastPathGauge" | "register" > ) => `
398404local masterQueueKey = KEYS[1]
399405local queueKey = KEYS[2]
400406local messageKey = KEYS[3]
@@ -451,10 +457,10 @@ if enableFastPath == '1' then
451457 redis.call('SET', messageKey, messageData)
452458 redis.call('SADD', queueCurrentConcurrencyKey, messageId)
453459 redis.call('SADD', envCurrentConcurrencyKey, messageId)
454- redis.call('RPUSH', workerQueueKey, messageKeyValue)${ v . fastPathGaugeEarly ?? "" }
460+ redis.call('RPUSH', workerQueueKey, messageKeyValue)${ v . fastPathGauge ?? "" }
455461 -- Fast-path skips the CK variant zset entirely; lengthCounter is unchanged.
456462 -- runningCounter is bumped later by dequeueMessageFromKeyTracked when the
457- -- worker pulls the message from the worker queue.${ v . fastPathGaugeLate ?? "" }
463+ -- worker pulls the message from the worker queue.
458464 return __qmret(1)
459465 end
460466 end
@@ -511,15 +517,15 @@ end
511517redis.call('SREM', queueCurrentConcurrencyKey, messageId)
512518redis.call('SREM', envCurrentConcurrencyKey, messageId)
513519redis.call('SREM', queueCurrentDequeuedKey, messageId)
514- redis.call('SREM', envCurrentDequeuedKey, messageId)${ v . expire ?? "" }
520+ redis.call('SREM', envCurrentDequeuedKey, messageId)
515521${ QUEUE_METRICS_CK_ENQUEUE_GAUGE_LUA }
516522
517523return __qmret(0)
518524 ` ;
519525
520526// Shared by the ck enqueue-with-TTL command and its virtual-time variant. The registration
521527// is the same rule as ckEnqueueLua above; it is a slot here rather than a second copy.
522- const ckEnqueueWithTtlLua = ( v : CkVtimeParts ) => `
528+ const ckEnqueueWithTtlLua = ( v : CkParts < "keys" | "args" | "register" > ) => `
523529local masterQueueKey = KEYS[1]
524530local queueKey = KEYS[2]
525531local messageKey = KEYS[3]
@@ -633,14 +639,14 @@ end
633639redis.call('SREM', queueCurrentConcurrencyKey, messageId)
634640redis.call('SREM', envCurrentConcurrencyKey, messageId)
635641redis.call('SREM', queueCurrentDequeuedKey, messageId)
636- redis.call('SREM', envCurrentDequeuedKey, messageId)${ v . expire ?? "" }
642+ redis.call('SREM', envCurrentDequeuedKey, messageId)
637643${ QUEUE_METRICS_CK_ENQUEUE_GAUGE_LUA }
638644
639645return __qmret(0)
640646 ` ;
641647
642648// Shared by the ck TTL sweep and its virtual-time variant.
643- const ckExpireTtlLua = ( v : CkVtimeParts ) => `
649+ const ckExpireTtlLua = ( v : CkParts < "args" | "drain" > ) => `
644650local ttlQueueKey = KEYS[1]
645651local keyPrefix = ARGV[1]
646652local currentTime = tonumber(ARGV[2])
@@ -760,7 +766,24 @@ return results
760766// Shared by the ck dequeue command and its virtual-time variant. This is the pair that
761767// diverges most, so it carries the most slots, but the 120-odd shared lines are shared
762768// rather than duplicated and a rewrite of the command reaches both builds.
763- const ckDequeueLua = ( v : CkVtimeParts ) => `
769+ const ckDequeueLua = (
770+ v : CkParts <
771+ | "keys"
772+ | "args"
773+ | "candidateScan"
774+ | "floorAndEmptyGuard"
775+ | "earlyReturn"
776+ | "minServableTag"
777+ | "attempted"
778+ | "serveLoopHead"
779+ | "servedTag"
780+ | "tagAdvance"
781+ | "idlePark"
782+ | "gcTagRead"
783+ | "readiness"
784+ | "passTwo"
785+ >
786+ ) => `
764787local ckIndexKey = KEYS[1]
765788local queueConcurrencyLimitKey = KEYS[2]
766789local envConcurrencyLimitKey = KEYS[3]
@@ -4807,7 +4830,7 @@ return __qmret(0)
48074830 this . redis . defineCommand ( "enqueueMessageCkTracked" , {
48084831 numberOfKeys : 15 ,
48094832 lua : ckEnqueueLua ( {
4810- fastPathGaugeEarly : `
4833+ fastPathGauge : `
48114834${ QUEUE_METRICS_CK_ENQUEUE_FASTPATH_GAUGE_LUA } `,
48124835 } ) ,
48134836 } ) ;
@@ -4836,7 +4859,7 @@ local stateTtl = ARGV[13]
48364859-- Arrival stacking (only read when this call registers a brand-new variant)
48374860local quantum = tonumber(ARGV[14] or '1')
48384861local arrivalCap = tonumber(ARGV[15] or '4294967296')` ,
4839- fastPathGaugeLate : `
4862+ fastPathGauge : `
48404863${ QUEUE_METRICS_CK_ENQUEUE_FASTPATH_GAUGE_LUA } `,
48414864 register : `
48424865-- Register this variant in the virtual-time index. NX means an already-advanced tag is
@@ -4880,8 +4903,6 @@ if redis.call('ZADD', ckVtimeKey, 'NX', vfloor, queueName) == 1 then
48804903end
48814904redis.call('EXPIRE', ckVtimeKey, stateTtl)
48824905redis.call('EXPIRE', ckVtimeFloorKey, stateTtl)
4883- ` ,
4884- expire : `
48854906` ,
48864907 } ) ,
48874908 } ) ;
@@ -4937,8 +4958,6 @@ if redis.call('ZADD', ckVtimeKey, 'NX', vfloor, queueName) == 1 then
49374958end
49384959redis.call('EXPIRE', ckVtimeKey, stateTtl)
49394960redis.call('EXPIRE', ckVtimeFloorKey, stateTtl)
4940- ` ,
4941- expire : `
49424961` ,
49434962 } ) ,
49444963 } ) ;
@@ -5407,8 +5426,8 @@ for _, ckQueueName in ipairs(ckQueues) do
54075426 // (lowest-tag) order; pass 2 fills the batch + discovers unregistered variants
54085427 // in the existing age order (work conservation, mixed-deploy safety). Only the
54095428 // :ckVtime / :ckVtimeFloor keys hold virtual times; ckIndex and the master
5410- // queue keep their timestamp score domain. The per-candidate serve body is a
5411- // verbatim copy of dequeueMessagesFromCkQueueTracked's, with the marked NEW
5429+ // queue keep their timestamp score domain. The per-candidate serve body has the
5430+ // same shape as dequeueMessagesFromCkQueueTracked's, with the virtual-time
54125431 // lines added (tag advance on serve, ZREM ckVtime on GC, floor advance from pass 1,
54135432 // and the notReady report that lets pass 1 step over a future-headed variant without
54145433 // spending a window slot on it).
0 commit comments