be a bit more wary of overrunning boundaries on the mmapped buffers, since the JVM seems kinda segfaulty in slab->task-seq

Author: ztellman

src/durable_queue.clj

@@ -219,24 +219,28 @@
 
          ;; is there a next task, and is there space left in the buffer?
          (when (and
-                 (pos? (.remaining buf'))
+                 (< header-size (.remaining buf'))
                  (== 1 (.get buf')))
 
            (lazy-seq
              (let [status (.get buf')
                    checksum (.getLong buf')
                    size (.getInt buf')]
-               (cons
 
-                 (task
-                   slab
-                   pos
-                   (+ header-size size)
-                   (read-write-lock slab))
+               ;; this shouldn't be necessary, but let's not gratuitously
+               ;; overreach our bounds
+               (when (< size (.remaining buf'))
+                 (cons
 
-                 (slab->task-seq
-                   slab
-                   (+ pos header-size size)))))))
+                   (task
+                     slab
+                     pos
+                     (+ header-size size)
+                     (read-write-lock slab))
+
+                   (slab->task-seq
+                     slab
+                     (+ pos header-size size))))))))
        (catch Throwable e
          ;; this implies unrecoverable corruption
          nil
@@ -424,6 +428,8 @@
 (defprotocol IQueues
   (^:private mark-complete! [_ q-name])
   (^:private mark-retry! [_ q-name])
+  (delete! [_]
+    "Deletes all files associated with the queues.")
   (stats [_]
     "Returns a map of queue names onto information about the immediate state of the queue.")
   (fsync [_]
@@ -611,6 +617,11 @@
 
            IQueues
 
+           (delete! [this]
+             (doseq [s (->> @queue-name->slabs vals (apply concat))]
+               (unmap s)
+               (delete-slab s)))
+
            (fsync [_]
              (doseq [slab (->> @queue-name->slabs vals (apply concat))]
                (sync! slab)))

test/durable_queue_test.clj

@@ -17,7 +17,8 @@
         tasks (range 1e4)]
     (doseq [t tasks]
       (put! q :foo t))
-    (is (= tasks (map deref (immediate-task-seq q :foo))))))
+    (is (= tasks (map deref (immediate-task-seq q :foo))))
+    (delete! q)))
 
 (deftest test-partial-slab-writes
   (clear-tmp-directory)
@@ -45,13 +46,13 @@
       (is (= (range 5 15) (map deref tasks')))
       (doseq [t (take 5 tasks')]
         (complete! t))))
-    
+
   (with-open [q (queues "/tmp")]
     (let [tasks' (immediate-task-seq q :foo)]
       (is (= (range 10 15) (map deref tasks')))
       (doseq [t (range 15 20)]
         (put! q :foo t))))
-  
+
   (let [q (queues "/tmp" {:complete? even?})]
     (is (= (remove even? (range 10 20)) (map deref (immediate-task-seq q :foo))))))
 
@@ -128,5 +129,5 @@
     (let [s (immediate-task-seq q :stress)]
       (doseq [t s]
         (complete! t))))
-  
+
   (clear-tmp-directory))