fix(obliterate): safer implementation

Author: manast

src/classes/queue.ts

@@ -217,17 +217,18 @@ export class Queue<
    * Note: This operation requires to iterate on all the jobs stored in the queue
    * and can be slow for very large queues.
    *
-   * @param { { force: boolean }} opts. Use force = true to force obliteration even
-   * with active jobs in the queue.
+   * @param { { force: boolean, count: number }} opts. Use force = true to force obliteration even
+   * with active jobs in the queue. Use count with the maximun number of deleted keys per iteration,
+   * 1000 is the default.
    */
-  async obliterate(opts?: { force: boolean }) {
+  async obliterate(opts?: { force?: boolean; count?: number }) {
     await this.pause();
 
     let cursor = 0;
     do {
-      cursor = await Scripts.obliterate(this, cursor, {
+      cursor = await Scripts.obliterate(this, {
         force: false,
-        count: 5000,
+        count: 1000,
         ...opts,
       });
     } while (cursor);

src/classes/scripts.ts

@@ -474,15 +474,12 @@ export class Scripts {
 
   static async obliterate(
     queue: Queue,
-    cursor: number,
     opts: { force: boolean; count: number },
   ) {
     const client = await queue.client;
 
-    const pattern = `${queue.opts.prefix}:${queue.name}:*`;
-
-    const keys: (string | number)[] = [queue.keys.meta, queue.keys.active];
-    const args = [cursor, pattern, opts.count, opts.force ? 'force' : null];
+    const keys: (string | number)[] = [queue.keys.meta, queue.toKey('')];
+    const args = [opts.count, opts.force ? 'force' : null];
 
     const result = await (<any>client).obliterate(keys.concat(args));
     if (result < 0) {

src/commands/obliterate-2.lua

@@ -1,14 +1,12 @@
 --[[
-    Completely obliterates a queue and all of its content
+    Completely obliterates a queue and all of its contents
      Input:
 
         KEYS[1] meta
-        KEYS[2] active
+        KEYS[2] base
         
-        ARGV[1]  cursor
-        ARGV[2]  pattern
-        ARGV[3]  count
-        ARGV[4]  force
+        ARGV[1]  count
+        ARGV[2]  force
 ]]
 
 -- This command completely destroys a queue including all of its jobs, current or past 
@@ -18,26 +16,104 @@
 -- The queue needs to be "paused" or it will return an error
 -- If the queue has currently active jobs then the script by default will return error,
 -- however this behaviour can be overrided using the `force` option.
+local maxCount = tonumber(ARGV[1])
+local count = 0
+local baseKey = KEYS[2]
 
 local rcall = redis.call
+local function getListItems(keyName)
+    return rcall('LRANGE', keyName, 0, -1)
+end
+
+local function getZSetItems(keyName)
+    return rcall('ZRANGE', keyName, 0, -1)
+end
+
+local function getSetItems(keyName)
+    return rcall('SMEMBERS', keyName, 0, -1)
+end
+
+local function removeKeys(parentKey, keys)
+    for i, key in ipairs(keys) do
+        if(count > maxCount) then
+            return true
+        end
+        rcall("DEL", baseKey .. key)
+        count = count + 1
+    end
+    rcall("DEL", parentKey)
+    return false
+end
+
+local function removeLockKeys(keys)
+    for i, key in ipairs(keys) do
+        if(count > maxCount) then
+            return true
+        end
+        rcall("DEL", baseKey .. key .. ':lock')
+        count = count + 1
+    end
+    return false
+end
 
 -- 1) Check if paused, if not return with error.
 if rcall("HEXISTS", KEYS[1], "paused") ~= 1 then
     return -1 -- Error, NotPaused
 end
 
 -- 2) Check if there are active jobs, if there are and not "force" return error.
-local activeKey = KEYS[2]
-local active = rcall('LRANGE', activeKey, 0, -1)
-if (#active > 0) then
-    if(ARGV[4] == "") then 
+local activeKey = baseKey .. 'active'
+local activeKeys = getListItems(activeKey)
+if (#activeKeys > 0) then
+    if(ARGV[2] == "") then 
         return -2 -- Error, ExistsActiveJobs
     end
 end
 
-local result = rcall("scan", ARGV[1], "MATCH", ARGV[2], "COUNT", ARGV[3])
-local cursor = result[1]
-local keys = result[2]
-for i, key in ipairs(keys) do
-    rcall("DEL", key)
+if(removeLockKeys(activeKeys)) then
+    return 1
 end
+
+if(removeKeys(activeKey, activeKeys)) then
+    return 1
+end
+
+local waitKey = baseKey .. 'paused'
+if(removeKeys(waitKey, getListItems(waitKey))) then
+    return 1
+end
+
+local delayedKey = baseKey .. 'delayed'
+if(removeKeys(delayedKey, getZSetItems(delayedKey))) then
+    return 1
+end
+
+local completedKey = baseKey .. 'completed'
+if(removeKeys(completedKey, getZSetItems(completedKey))) then
+    return 1
+end
+
+local failedKey = baseKey .. 'failed'
+if(removeKeys(failedKey, getZSetItems(failedKey))) then
+    return 1
+end
+
+local waitKey = baseKey .. 'wait'
+if(removeKeys(waitKey, getListItems(waitKey))) then
+    return 1
+end
+
+local waitKey = baseKey .. 'wait'
+if(removeKeys(waitKey, getListItems(waitKey))) then
+    return 1
+end
+
+rcall("DEL", baseKey .. 'priority')
+rcall("DEL", baseKey .. 'events')
+rcall("DEL", baseKey .. 'delay')
+rcall("DEL", baseKey .. 'stalled-check')
+rcall("DEL", baseKey .. 'stalled')
+rcall("DEL", baseKey .. 'id')
+rcall("DEL", baseKey .. 'meta')
+
+return 0

src/test/test_obliterate.ts

@@ -29,7 +29,7 @@ describe('Obliterate', () => {
     await queue.obliterate();
 
     const client = await queue.client;
-    const keys = await client.keys(`bull:${queue.name}*`);
+    const keys = await client.keys(`bull:${queue.name}:*`);
 
     expect(keys.length).to.be.eql(0);
   });
@@ -56,7 +56,7 @@ describe('Obliterate', () => {
 
     await queue.obliterate();
     const client = await queue.client;
-    const keys = await client.keys(`bull:${queue.name}*`);
+    const keys = await client.keys(`bull:${queue.name}:*`);
     expect(keys.length).to.be.eql(0);
 
     await worker.close();
@@ -87,7 +87,7 @@ describe('Obliterate', () => {
       await queue.obliterate();
     } catch (err) {
       const client = await queue.client;
-      const keys = await client.keys(`bull:${queue.name}*`);
+      const keys = await client.keys(`bull:${queue.name}:*`);
       expect(keys.length).to.be.not.eql(0);
 
       await worker.close();
@@ -119,7 +119,7 @@ describe('Obliterate', () => {
     await job.waitUntilFinished(queueEvents);
     await queue.obliterate({ force: true });
     const client = await queue.client;
-    const keys = await client.keys(`bull:${queue.name}*`);
+    const keys = await client.keys(`bull:${queue.name}:*`);
     expect(keys.length).to.be.eql(0);
 
     await worker.close();