fix: blocked groups (try 2)

Author: lindesvard

src/lua/complete-and-reserve-next-with-metadata.lua

@@ -174,4 +174,4 @@ if nextHead and #nextHead >= 2 then
   redis.call("ZADD", readyKey, nextScore, groupId)
 end
 
-return id .. "|||" .. groupId .. "|||" .. payload .. "|||" .. attempts .. "|||" .. maxAttempts .. "|||" .. seq .. "|||" .. enq .. "|||" .. orderMs .. "|||" .. score .. "|||" .. deadline
+return id .. "||GROUPMQ||" .. groupId .. "||GROUPMQ||" .. payload .. "||GROUPMQ||" .. attempts .. "||GROUPMQ||" .. maxAttempts .. "||GROUPMQ||" .. seq .. "||GROUPMQ||" .. enq .. "||GROUPMQ||" .. orderMs .. "||GROUPMQ||" .. score .. "||GROUPMQ||" .. deadline

src/lua/reserve-atomic.lua

@@ -111,4 +111,4 @@ if nextHead and #nextHead >= 2 then
   redis.call("ZADD", readyKey, nextScore, groupId)
 end
 
-return id .. "|||" .. groupId .. "|||" .. payload .. "|||" .. attempts .. "|||" .. maxAttempts .. "|||" .. seq .. "|||" .. enq .. "|||" .. orderMs .. "|||" .. score .. "|||" .. deadline
+return id .. "||GROUPMQ||" .. groupId .. "||GROUPMQ||" .. payload .. "||GROUPMQ||" .. attempts .. "||GROUPMQ||" .. maxAttempts .. "||GROUPMQ||" .. seq .. "||GROUPMQ||" .. enq .. "||GROUPMQ||" .. orderMs .. "||GROUPMQ||" .. score .. "||GROUPMQ||" .. deadline

src/lua/reserve-batch.lua

@@ -116,7 +116,7 @@ for i = 1, #groups, 2 do
               redis.call("ZADD", readyKey, nextScore, gid)
             end
 
-            table.insert(out, id .. "|||" .. groupId .. "|||" .. payload .. "|||" .. attempts .. "|||" .. maxAttempts .. "|||" .. seq .. "|||" .. enq .. "|||" .. orderMs .. "|||" .. score .. "|||" .. deadline)
+            table.insert(out, id .. "||GROUPMQ||" .. groupId .. "||GROUPMQ||" .. payload .. "||GROUPMQ||" .. attempts .. "||GROUPMQ||" .. maxAttempts .. "||GROUPMQ||" .. seq .. "||GROUPMQ||" .. enq .. "||GROUPMQ||" .. orderMs .. "||GROUPMQ||" .. score .. "||GROUPMQ||" .. deadline)
             table.insert(processedGroups, gid)
           end
         end

src/lua/reserve.lua

@@ -56,6 +56,9 @@ if (not groups or #groups == 0) or shouldCheckStalled then
           local headScore = tonumber(head[2])
           redis.call("ZADD", readyKey, headScore, gid)
         end
+        -- Remove from group active list (BullMQ-style) - CRITICAL to unblock the group
+        local groupActiveKey = ns .. ":g:" .. gid .. ":active"
+        redis.call("LREM", groupActiveKey, 1, jobId)
         redis.call("DEL", ns .. ":lock:" .. gid)
         redis.call("DEL", procKey)
         redis.call("ZREM", processingKey, jobId)
@@ -165,6 +168,6 @@ if nextHead and #nextHead >= 2 then
   redis.call("ZADD", readyKey, nextScore, chosenGid)
 end
 
-return id .. "|||" .. groupId .. "|||" .. payload .. "|||" .. attempts .. "|||" .. maxAttempts .. "|||" .. seq .. "|||" .. enq .. "|||" .. orderMs .. "|||" .. score .. "|||" .. deadline
+return id .. "||GROUPMQ||" .. groupId .. "||GROUPMQ||" .. payload .. "||GROUPMQ||" .. attempts .. "||GROUPMQ||" .. maxAttempts .. "||GROUPMQ||" .. seq .. "||GROUPMQ||" .. enq .. "||GROUPMQ||" .. orderMs .. "||GROUPMQ||" .. score .. "||GROUPMQ||" .. deadline
 
 

src/queue.ts

@@ -717,7 +717,7 @@ export class Queue<T = any> {
 
     if (!raw) return null;
 
-    const parts = raw.split('|||');
+    const parts = raw.split('||GROUPMQ||');
     if (parts.length !== 10) return null;
 
     let data: T;
@@ -857,7 +857,7 @@ export class Queue<T = any> {
       }
 
       // Parse the result (same format as reserve methods)
-      const parts = result.split('|||');
+      const parts = result.split('||GROUPMQ||');
       if (parts.length !== 10) {
         this.logger.error(
           'Queue completeAndReserveNextWithMetadata: unexpected result format:',
@@ -1503,7 +1503,7 @@ export class Queue<T = any> {
     if (!result) return null;
 
     // Parse the delimited string response (same format as regular reserve)
-    const parts = result.split('|||');
+    const parts = result.split('||GROUPMQ||');
     if (parts.length < 10) return null;
 
     const [
@@ -1549,7 +1549,7 @@ export class Queue<T = any> {
     const out: Array<ReservedJob<T>> = [];
     for (const r of results || []) {
       if (!r) continue;
-      const parts = r.split('|||');
+      const parts = r.split('||GROUPMQ||');
       if (parts.length !== 10) continue;
       out.push({
         id: parts[0],