fix(worker): process job when closing and it was moved to active

src/classes/worker.ts

@@ -189,6 +189,7 @@ export class Worker<
 
   private stalledCheckStopper?: () => void;
   private waiting: Promise<number> | null = null;
+  private waitingRun: Promise<void> = Promise.resolve();
   private _repeat: Repeat; // To be deprecated in v6 in favor of Job Scheduler
 
   protected _jobScheduler: JobScheduler;
@@ -431,6 +432,12 @@ export class Worker<
   }
 
   async run() {
+    this.waitingRun = this.waitRun();
+
+    await this.waitingRun;
+  }
+
+  async waitRun() {
     if (!this.processFn) {
       throw new Error('No process function is defined.');
     }
@@ -535,8 +542,8 @@ export class Worker<
         }
       }
 
+      await asyncFifoQueue.waitAll();
       this.running = false;
-      return await asyncFifoQueue.waitAll();
     } catch (error) {
       this.running = false;
       throw error;
@@ -819,7 +826,7 @@ will never work with more accuracy than 1ms. */
     fetchNextCallback = () => true,
     jobsInProgress: Set<{ job: Job; ts: number }>,
   ): Promise<void | Job<DataType, ResultType, NameType>> {
-    if (!job || this.closing || this.paused) {
+    if (!job) {
       return;
     }
 
@@ -1164,7 +1171,7 @@ will never work with more accuracy than 1ms. */
     }
 
     if (this.asyncFifoQueue) {
-      await this.asyncFifoQueue.waitAll();
+      await this.waitingRun;
     }
 
     reconnect && (await this.blockingConnection.reconnect());

src/commands/addDelayedJob-6.lua

@@ -88,7 +88,7 @@ else
     end
 end
 
-local deduplicationJobId = deduplicateJob(args[1], opts['de'],
+local deduplicationJobId = deduplicateJob(opts['de'],
   jobId, deduplicationKey, eventsKey, maxEvents)
 if deduplicationJobId then
   return deduplicationJobId

src/commands/addParentJob-4.lua

@@ -81,7 +81,7 @@ else
     end
 end
 
-local deduplicationJobId = deduplicateJob(args[1], opts['de'],
+local deduplicationJobId = deduplicateJob(opts['de'],
   jobId, deduplicationKey, eventsKey, maxEvents)
 if deduplicationJobId then
   return deduplicationJobId

src/commands/addPrioritizedJob-8.lua

@@ -90,7 +90,7 @@ else
     end
 end
 
-local deduplicationJobId = deduplicateJob(args[1], opts['de'],
+local deduplicationJobId = deduplicateJob(opts['de'],
   jobId, deduplicationKey, eventsKey, maxEvents)
 if deduplicationJobId then
   return deduplicationJobId

src/commands/addStandardJob-8.lua

@@ -94,7 +94,7 @@ else
     end
 end
 
-local deduplicationJobId = deduplicateJob(args[1], opts['de'],
+local deduplicationJobId = deduplicateJob(opts['de'],
   jobId, deduplicationKey, eventsKey, maxEvents)
 if deduplicationJobId then
   return deduplicationJobId

src/commands/includes/deduplicateJob.lua

@@ -1,24 +1,24 @@
 --[[
-  Function to debounce a job.
+  Function to deduplicate a job.
 ]]
 
-local function deduplicateJob(prefixKey, deduplicationOpts, jobId, deduplicationKey, eventsKey, maxEvents)
+local function deduplicateJob(deduplicationOpts, jobId, deduplicationKey, eventsKey, maxEvents)
   local deduplicationId = deduplicationOpts and deduplicationOpts['id']
   if deduplicationId then
     local ttl = deduplicationOpts['ttl']
     local deduplicationKeyExists
     if ttl then
-      deduplicationKeyExists = not rcall('SET', deduplicationKey, jobId, 'PX', ttl, 'NX')
+      deduplicationKeyExists = rcall('SET', deduplicationKey, jobId, 'PX', ttl, 'NX')
     else
-      deduplicationKeyExists = not rcall('SET', deduplicationKey, jobId, 'NX')
+      deduplicationKeyExists = rcall('SET', deduplicationKey, jobId, 'NX')
     end
-    if deduplicationKeyExists then
-      local currentDebounceJobId = rcall('GET', deduplicationKey)
+    if deduplicationKeyExists == false  then
+      local currentDeduplicatedJobId = rcall('GET', deduplicationKey)
       rcall("XADD", eventsKey, "MAXLEN", "~", maxEvents, "*", "event",
-        "debounced", "jobId", currentDebounceJobId, "debounceId", deduplicationId)
+        "debounced", "jobId", currentDeduplicatedJobId, "debounceId", deduplicationId)
       rcall("XADD", eventsKey, "MAXLEN", "~", maxEvents, "*", "event",
-        "deduplicated", "jobId", currentDebounceJobId, "deduplicationId", deduplicationId)
-      return currentDebounceJobId
+        "deduplicated", "jobId", currentDeduplicatedJobId, "deduplicationId", deduplicationId)
+      return currentDeduplicatedJobId
     end
   end
 end

tests/test_events.ts

@@ -587,9 +587,15 @@ describe('events', function () {
           );
 
           let debouncedCounter = 0;
-          queueEvents.on('debounced', ({ jobId }) => {
-            debouncedCounter++;
+          const debounced = new Promise<void>(resolve => {
+            queueEvents.on('debounced', () => {
+              debouncedCounter++;
+              if (debouncedCounter == 2) {
+                resolve();
+              }
+            });
           });
+
           await job.remove();
 
           await queue.add(testName, { foo: 'bar' }, { debounce: { id: 'a1' } });
@@ -602,6 +608,11 @@ describe('events', function () {
             { debounce: { id: 'a1' } },
           );
           await secondJob.remove();
+          await debounced;
+
+          const getDeboundedJobId = await queue.getDebounceJobId('a1');
+
+          expect(getDeboundedJobId).to.be.null;
 
           expect(debouncedCounter).to.be.equal(2);
         });
@@ -822,6 +833,9 @@ describe('events', function () {
           await secondJob.remove();
           await deduplication;
 
+          const getDeduplicationJobId = await queue.getDeduplicationJobId('a1');
+
+          expect(getDeduplicationJobId).to.be.null;
           expect(deduplicatedCounter).to.be.equal(2);
         });
       });