fix: refactor, cleanu, improve concurrency

Author: lindesvard

example/index.ts

@@ -20,49 +20,30 @@ const run = async () => {
     jobShouldTakeThisLongMs: number;
     groupId: string;
   }>({
+    logger: true,
     redis: redis,
     namespace: 'example',
-    keepCompleted: 100,
+    keepCompleted: 100000,
     keepFailed: 100,
   });
 
-  const cronQueue = new Queue<{
-    id: string;
-    jobShouldTakeThisLongMs: number;
-    groupId: string;
-  }>({
-    redis: redis,
-    namespace: 'cron',
-    keepCompleted: 100,
-    keepFailed: 100,
-  });
+  const CONCURRENCY = 4;
+  const WORKERS = 4;
 
-  const worker = new Worker({
-    concurrency: 2,
-    queue: queue,
-    async handler(job) {
-      await sleep(job.data.jobShouldTakeThisLongMs);
-      if (Math.random() > 0.9) {
-        throw new Error('test error');
-      }
-      return `basic worker completed job ${job.id}`;
-    },
-  });
-
-  const cronWorker = new Worker({
-    queue: cronQueue,
-    // Run cleanup/scheduler frequently so repeating jobs trigger on time
-    cleanupIntervalMs: 1000,
-    async handler(job) {
-      await sleep(job.data.jobShouldTakeThisLongMs);
-      if (Math.random() > 0.9) {
-        throw new Error('test error');
-      }
-      return `cron worker completed job ${job.id}`;
-    },
-  });
+  const createWorker = (index: number, concurrency: number) => {
+    return new Worker({
+      concurrency: concurrency,
+      queue: queue,
+      async handler(job) {
+        console.log(`processing job worker ${index}`, job.id);
+        await sleep(job.data.jobShouldTakeThisLongMs);
+      },
+    });
+  };
 
-  const workers = [worker, cronWorker];
+  const workers = Array.from({ length: WORKERS }, (_, i) =>
+    createWorker(i, CONCURRENCY),
+  );
 
   workers.forEach((worker) => {
     worker.on('completed', (job) => {
@@ -106,58 +87,43 @@ const run = async () => {
   const serverAdapter = new HonoAdapter(serveStatic);
 
   createBullBoard({
-    queues: [
-      new BullBoardGroupMQAdapter(queue),
-      new BullBoardGroupMQAdapter(cronQueue),
-    ],
+    queues: [new BullBoardGroupMQAdapter(queue)],
     serverAdapter,
   });
 
   const basePath = '/ui';
   serverAdapter.setBasePath(basePath);
   app.route(basePath, serverAdapter.registerPlugin());
 
-  // Add a cron job that runs every 5 seconds
-  await cronQueue.removeRepeatingJob('groupId', { every: 5000 });
-  await cronQueue.add({
-    data: {
-      id: Math.random().toString(36).substring(2, 15),
-      jobShouldTakeThisLongMs: Math.random() * 1000,
-      groupId: 'groupId',
-    },
-    groupId: 'groupId',
-    repeat: { every: 5000 }, // every 5 seconds
-  });
-
   // Add basic job every 2.5 seconds
   const groups = [
     ...new Array(25).fill(null).map((_, i) => `groupId_${i + 1}`),
   ];
-  setInterval(async () => {
-    const groupId = groups[Math.floor(Math.random() * groups.length)]!;
-    await queue.add({
-      data: {
-        id: Math.random().toString(36).substring(2, 15),
-        jobShouldTakeThisLongMs: Math.random() * 1000,
-        groupId,
-      },
-      groupId,
-    });
-  }, 1_000);
+  // setInterval(async () => {
+  //   const groupId = groups[Math.floor(Math.random() * groups.length)]!;
+  //   await queue.add({
+  //     data: {
+  //       id: Math.random().toString(36).substring(2, 15),
+  //       jobShouldTakeThisLongMs: Math.random() * 1000,
+  //       groupId,
+  //     },
+  //     groupId,
+  //   });
+  // }, 1_000);
 
   app.get('/add', async (c) => {
     const groups = [
-      ...new Array(25).fill(null).map((_, i) => `groupId_${i + 1}`),
+      ...new Array(1000).fill(null).map((_, i) => `groupId_${i + 1}`),
     ];
     const events = [
-      ...new Array(50).fill(null).map((_, i) => `event_${i + 1}`),
+      ...new Array(200).fill(null).map((_, i) => `event_${i + 1}`),
     ];
     for (const event of events) {
       const groupId = groups[Math.floor(Math.random() * groups.length)]!;
       await queue.add({
         data: {
           id: event,
-          jobShouldTakeThisLongMs: Math.random() * 500 + 500,
+          jobShouldTakeThisLongMs: Math.random() * 40 + 20,
           groupId,
         },
         groupId,
@@ -167,6 +133,20 @@ const run = async () => {
     return c.json({ message: `${events.length} jobs added` });
   });
 
+  app.get('/add-single', async (c) => {
+    const groupId = `groupId_${Math.floor(Math.random() * 100) + 1}`;
+    const event = `event_${Math.floor(Math.random() * 200) + 1}`;
+    await queue.add({
+      data: {
+        id: event,
+        jobShouldTakeThisLongMs: Math.random() * 40 + 20,
+        groupId,
+      },
+      groupId,
+    });
+    return c.json({ message: 'job added' });
+  });
+
   showRoutes(app);
 
   serve({ fetch: app.fetch, port: 3000 }, ({ address, port }) => {

src/lua/ack.lua

@@ -14,12 +14,18 @@ local circuitBreakerKey = ns .. ":stalled:circuit"
 local lastCheck = redis.call("GET", circuitBreakerKey)
 if lastCheck then
   local lastCheckTime = tonumber(lastCheck)
-  if lastCheckTime and (now - lastCheckTime) < 1000 then
-    -- Circuit breaker: only check stalled jobs once per second
+  -- More aggressive circuit breaker for high concurrency: 2 seconds instead of 1
+  local circuitBreakerInterval = 2000
+  if lastCheckTime and (now - lastCheckTime) < circuitBreakerInterval then
+    -- Circuit breaker: only check stalled jobs once per 2 seconds
     return {}
   end
 end
-redis.call("SET", circuitBreakerKey, now, "PX", 2000)
+redis.call("SET", circuitBreakerKey, now, "PX", 3000)
+
+local processingKey = ns .. ':processing'
+local groupsKey = ns .. ':groups'
+local stalledKey = ns .. ':stalled'
 
 -- BullMQ-inspired: Two-phase stalled detection for better accuracy
 -- Phase 1: Get potentially stalled jobs (jobs past their deadline)
@@ -28,10 +34,6 @@ if not potentiallyStalled or #potentiallyStalled == 0 then
   return {}
 end
 
-local processingKey = ns .. ':processing'
-local groupsKey = ns .. ':groups'
-local stalledKey = ns .. ':stalled'
-
 local results = {}
 
 -- Get all jobs in processing state
@@ -57,9 +59,7 @@ for _, jobId in ipairs(processingJobs) do
       -- Remove from processing
       redis.call('ZREM', processingKey, jobId)
 
-      -- Decrement active counter (job is being failed)
-      local activeCountKey = ns .. ':count:active'
-      redis.call('DECR', activeCountKey)
+      -- No counter operations - use ZCARD for counts
 
       -- Remove from group if it's there
       local groupKey = ns .. ':g:' .. groupId
@@ -109,11 +109,7 @@ for _, jobId in ipairs(processingJobs) do
         -- Job is confirmed to still be in processing, safe to recover
         redis.call('ZREM', processingKey, jobId)
 
-        -- Update counters: active -> waiting
-        local activeCountKey = ns .. ':count:active'
-        local waitingCountKey = ns .. ':count:waiting'
-        redis.call('DECR', activeCountKey)
-        redis.call('INCR', waitingCountKey)
+        -- No counter operations - use ZCARD for counts
 
         -- Release group lock if this job holds it
         local lockKey = ns .. ':lock:' .. groupId

src/lua/check-stalled.lua

@@ -32,11 +32,7 @@ for _, jobId in ipairs(expiredJobs) do
         redis.call("DEL", procKey)
         redis.call("ZREM", processingKey, jobId)
 
-        -- Update counters: active -> waiting (job recovered from stalled state)
-        local activeCountKey = ns .. ":count:active"
-        local waitingCountKey = ns .. ":count:waiting"
-        redis.call("DECR", activeCountKey)
-        redis.call("INCR", waitingCountKey)
+        -- No counter operations - use ZCARD for counts
 
         cleaned = cleaned + 1
       end

src/lua/cleanup.lua

@@ -20,9 +20,7 @@ local vt = tonumber(ARGV[14])
 redis.call("DEL", ns .. ":processing:" .. completedJobId)
 redis.call("ZREM", ns .. ":processing", completedJobId)
 
--- Decrement active counter
-local activeCountKey = ns .. ":count:active"
-redis.call("DECR", activeCountKey)
+-- No counter operations - use ZCARD for counts
 
 -- Part 2: Record job metadata (completed or failed)
 local jobKey = ns .. ":job:" .. completedJobId
@@ -121,8 +119,7 @@ redis.call("HSET", procKey, "groupId", groupId, "deadlineAt", tostring(deadline)
 local processingKey = ns .. ":processing"
 redis.call("ZADD", processingKey, deadline, id)
 
--- Increment active counter for new job (completed job was already decremented above)
-redis.call("INCR", activeCountKey)
+-- No counter operations - use ZCARD for counts
 
 local nextHead = redis.call("ZRANGE", gZ, 0, 0, "WITHSCORES")
 if nextHead and #nextHead >= 2 then

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

@@ -18,9 +18,7 @@ local maxAttempts = ARGV[12]
 redis.call("DEL", ns .. ":processing:" .. jobId)
 redis.call("ZREM", ns .. ":processing", jobId)
 
--- Decrement active counter
-local activeCountKey = ns .. ":count:active"
-redis.call("DECR", activeCountKey)
+-- No counter operations - use ZCARD for counts
 
 local lockKey = ns .. ":lock:" .. gid
 local val = redis.call("GET", lockKey)

src/lua/complete-with-metadata.lua

@@ -12,9 +12,7 @@ redis.call("ZREM", gZ, jobId)
 redis.call("DEL", ns .. ":processing:" .. jobId)
 redis.call("ZREM", ns .. ":processing", jobId)
 
--- Decrement active counter (job is being dead-lettered, not re-queued)
-local activeCountKey = ns .. ":count:active"
-redis.call("DECR", activeCountKey)
+-- No counter operations - use ZCARD for counts
 
 -- Remove idempotence mapping to allow reuse
 redis.call("DEL", ns .. ":unique:" .. jobId)