fixes #30 by adding delete-queue! and delete-all!

and deprecating delete! as dangerous

Signed-off-by: Sean Corfield <[email protected]>

Author: seancorfield

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 v0.2.0 in progress
 * Deprecated single-segment `durable-queue` ns; use `clj-commons.durable-queue` instead
+* Deprecate `delete!` because it is dangerous: it leaves queues in a corrupted state (because it deletes files without cleaning up in-memory state).
+* Address [#30](https://github.com/clj-commons/durable-queue/issues/30) by adding `delete-queue!` and `delete-all!` to safely delete an individual queue (and its files) and safely delete all queues (and their files).
 * Clean up lint issues; export clj-kondo config
 * Add a changelog!
 

README.md

@@ -70,6 +70,20 @@ To get a description of the current state of the queue, we can use `stats`, whic
 | `:num-slabs` | the number of underlying files which are being used to store tasks |
 | `:num-active-slabs` | the number of underlying files which are currently open and mapped into memory |
 
+As of version 0.2.0, you can delete an entire queue using `delete-queue!`, which will remove all tasks and delete all files associated with the queue:
+
+```clj
+> (dq/delete-queue! q :foo)
+{}
+```
+
+As of version 0.2.0, you can also delete all queues using `delete-all!`, which will remove all tasks and delete all files associated with all queues:
+
+```clj
+> (dq/delete-all! q)
+nil
+```
+
 ### configuring the queues
 
 `queues` can be given a number of different options, which can affect its performance and correctness.

src/clj_commons/durable_queue.clj

@@ -424,20 +424,24 @@
   (^:private mark-complete! [_ q-name])
   (^:private mark-retry! [_ q-name])
   (delete! [_]
-    "Deletes all files associated with the queues.")
+    "DEPRECATED: Deletes all files associated with all queues. Dangerous!")
+  (delete-queue! [_ q-name]
+    "Deletes the specific queue, including all associated files.")
+  (delete-all! [_]
+    "Deletes all queues, including all associated files.")
   (stats [_]
     "Returns a map of queue names onto information about the immediate state of the queue.")
   (fsync [_]
     "Forces an fsync on all modified files.")
   (take!
     [_ q-name]
     [_ q-name timeout timeout-val]
-    "A blocking dequeue from `name`.  If `timeout` is specified, returns `timeout-val` if
+    "A blocking dequeue from `q-name`.  If `timeout` is specified, returns `timeout-val` if
      no task is available within `timeout` milliseconds.")
   (put!
     [_ q-name task-descriptor]
     [_ q-name task-descriptor timeout]
-    "A blocking enqueue to `name`.  If `timeout` is specified, returns `false` if unable to
+    "A blocking enqueue to `q-name`.  If `timeout` is specified, returns `false` if unable to
      enqueue within `timeout` milliseconds, `true` otherwise."))
 
 (defn queues
@@ -613,11 +617,25 @@
 
            IQueues
 
-           (delete! [_]
+           (delete! [_] ; DEPRECATED and dangerous since it breaks the queues
              (doseq [s (->> @queue-name->slabs vals (apply concat))]
                (unmap s)
                (delete-slab s)))
 
+           (delete-queue! [_ q-name]
+             (let [q-name (munge (name q-name))]
+               (doseq [s (get @queue-name->slabs q-name)]
+                 (unmap s)
+                 (delete-slab s))
+               (.clear (queue q-name))
+               (swap! queue-name->stats dissoc q-name)
+               (swap! queue-name->slabs dissoc q-name)
+               (swap! queue-name->current-slab dissoc q-name)))
+
+           (delete-all! [this]
+             (doseq [q (keys @queue-name->stats )]
+               (delete-queue! this q)))
+
            (fsync [_]
              (doseq [slab (->> @queue-name->slabs vals (apply concat))]
                (sync! slab)))
@@ -637,15 +655,15 @@
            (stats [_]
              (let [ks (keys @queue-name->stats)]
                (zipmap ks
-                 (map
-                   (fn [q-name]
-                     (merge
-                       {:num-slabs (-> @queue-name->slabs (get q-name) count)
-                        :num-active-slabs (->> (get @queue-name->slabs q-name)
-                                            (filter mapped?)
-                                            count)}
-                       (immediate-stats (queue q-name) (get @queue-name->stats q-name))))
-                   ks))))
+                       (map
+                        (fn [q-name]
+                          (merge
+                           {:num-slabs (-> @queue-name->slabs (get q-name) count)
+                            :num-active-slabs (->> (get @queue-name->slabs q-name)
+                                                   (filter mapped?)
+                                                   count)}
+                           (immediate-stats (queue q-name) (get @queue-name->stats q-name))))
+                        ks))))
 
            (take! [_ q-name timeout timeout-val]
              (let [q-name (munge (name q-name))
@@ -663,8 +681,8 @@
                        (when-not (= slab old-slab)
                          (swap! queue-name->current-slab assoc q-name slab)
                          (doseq [s (->> (get @queue-name->slabs q-name)
-                                     butlast
-                                     (remove #(= slab %)))]
+                                        butlast
+                                        (remove #(= slab %)))]
                            (unmap s))))
 
                      (status! t :in-progress)

test/clj_commons/durable_queue_test.clj

@@ -17,8 +17,7 @@
         tasks (range 1e4)]
     (doseq [t tasks]
       (dq/put! q :foo t))
-    (is (= tasks (map deref (dq/immediate-task-seq q :foo))))
-    (dq/delete! q)))
+    (is (= tasks (map deref (dq/immediate-task-seq q :foo))))))
 
 (deftest test-partial-slab-writes
   (clear-tmp-directory)