core, storage: add storage worker that handles writes asynchronously and thread-safely

fix

3 files changed, 88 insertions, 9 deletions

diff --git a/src/weepinbell/core.clj b/src/weepinbell/core.clj
index 20c0aef..1cc233e 100644
--- a/src/weepinbell/core.clj
+++ b/src/weepinbell/core.clj
@@ -4,6 +4,7 @@
             [weepinbell.config :refer [load-config]]
             [weepinbell.webhook :as webhook]
             [weepinbell.storage.edn-storage :as edn]
+            [weepinbell.storage.worker :as worker]
             [clojure.core.async :refer [chan >!! <!!]]
             [taoensso.timbre :as timbre]
             ;; two main subsystems' facades
@@ -28,7 +29,8 @@
         config (load-config (or (System/getenv "WEEPINBELL_CONFIG") "config.edn"))
         ;; Storage middle layer, a map of data-manipulating functions.
         ;; Supports various storage as long as the map conforms storage.spec.
-        store (edn/use-store (:data-path config))
+        bare-store (edn/use-store (:data-path config))
+        [stop-worker async-store] (worker/start bare-store)
         ;; Hooks, a map of callback functions that processor calls 
         ;; on certain events. Can be used for webhooks and more.
         hooks (webhook/make-hooks config)
@@ -36,13 +38,17 @@
         stop-processor (start-processor
                         {:wmc wmc
                          :callbacks hooks
-                         :storage store
+                         :storage async-store
                          :config config})
         stop-webserver (start-webserver
                         {:wmc wmc
-                         :store store
+                         :store async-store
                          :config config})]
-    (fn [] (stop-processor) (stop-webserver) true)))
+    (fn [] 
+      (stop-processor) 
+      (stop-worker)
+      (stop-webserver) 
+      true)))
 
 ; Entry point; Terminal user interface
 ;; start up system and wait for signals to perform graceful shutdown
diff --git a/src/weepinbell/storage/worker.clj b/src/weepinbell/storage/worker.clj
new file mode 100644
index 0000000..41d4912
--- /dev/null
+++ b/src/weepinbell/storage/worker.clj
@@ -0,0 +1,45 @@
+(ns weepinbell.storage.worker

+  "Storage worker is a background loop that handles write operations 

+   to the storage, which makes write/delete/update thread-safe."

+  (:require [clojure.core.async :refer [go-loop chan put! <! close!]]

+            [taoensso.timbre :as timbre]))

+

+(defn- start-worker

+  "Asynchrnously start worker loop that takes from writer-chan 

+   and writes to storage when operation is received.

+   Returns a function that stops the worker gracefully."

+  [storage writer-chan]

+  (go-loop [operation (<! writer-chan)]

+    (let [{:keys [op args]} operation]

+      (if (not= op :stop)

+        (do ;; perform operation and recur

+          (try

+            (apply (op storage) args)

+            (catch Throwable e

+              (timbre/error "Something unexpected occrus while writing to storage.")

+              (timbre/error e)))

+          (recur (<! writer-chan)))

+        (do

+          (close! writer-chan)

+          (timbre/info "shutting down storage worker.")))))

+  ;; return a function that stops the loop

+  (fn [] (put! writer-chan {:op :stop})))

+

+(defn- put-op!

+  "Asynchronously put an operation map in a channel."

+  [channel op args]

+  (put! channel {:op op :args args}))

+

+(defn start

+  "Start worker. Wraps a base storage and returns a vector of 

+  a stopper function and a asynchronous storage vtable."

+  [storage]

+  (let [writer-chan (chan)

+        put-op (partial put-op! writer-chan)

+        stop-worker (start-worker storage writer-chan)

+        vtable {:append #(put-op :append [%])

+                :update (fn [& args] (put-op :update args))

+                :delete (fn [& args] (put-op :delete args))

+                :get (:get storage)

+                :get-by-domain (:get-by-domain storage)}]

+    [stop-worker vtable]))

diff --git a/test/weepinbell/storage_test.clj b/test/weepinbell/storage_test.clj
index 53abbb6..743a0cf 100644
--- a/test/weepinbell/storage_test.clj
+++ b/test/weepinbell/storage_test.clj
@@ -1,10 +1,23 @@
 (ns weepinbell.storage-test

   (:require [clojure.test :refer [testing is deftest]]

             [weepinbell.storage.edn-storage :as edns]

+            [weepinbell.storage.worker :as worker]

             [clojure.java.io :as io]

             [clojure.spec.alpha :as s]

             [weepinbell.storage.spec]))

 

+;; preparing storages to test

+

+(def test-edn-file "tmp/testfile")

+(when (.exists (io/file test-edn-file))

+  (io/delete-file test-edn-file))

+

+(def edn-storage (edns/use-store test-edn-file))

+(def storages [edn-storage

+               (last (worker/start edn-storage))])

+

+;; check if storages conform spec

+

 (def spec-check (partial s/valid? :weepinbell.storage.spec/vtable))

 (deftest storage-conformity-test

   (testing "edn-storage should conform storage spec."

@@ -17,10 +30,16 @@
                    nil 1 [] {}]]

       (is (not (spec-check check))))))

 

-(def test-edn-file "tmp/testfile")

-(when (.exists (io/file test-edn-file))

-  (io/delete-file test-edn-file))

-(def storages [(edns/use-store test-edn-file)])

+;; basic general test suite

+

+(defmacro wait-til! [condition-expr & body]

+  `(let [~'start (System/currentTimeMillis)]

+     (loop []

+       (if (try ~condition-expr (catch Throwable ~'_ false))

+         (do ~@body)

+         (when (< (- (System/currentTimeMillis) ~'start) 10000)

+           (Thread/sleep 10)

+           (recur))))))

 

 (defn- wm= [& wms]

   (->> (map :webmention wms)

@@ -36,9 +55,11 @@
                            :id id

                            :modified (System/currentTimeMillis)}}]

     ((:append storage) data)

+    (wait-til! ((:get storage) id))

     (is (wm= data ((:get storage) id)))

     (is (wm= data ((:get storage) source target)))

     ((:delete storage) id)

+    (wait-til! (nil? ((:get storage) id)))

     (is (nil? ((:get storage) id)))))

 

 (defn- test-append-and-get-list [storage]

@@ -54,10 +75,14 @@
         ids (map #(:id (:webmention %)) data)]

     (doseq [item data]

       ((:append storage) item))

+    (wait-til! (->> data

+                    (map (fn [item] (get-in item [:webmention :id])))

+                    (every? #((:get storage) %))))

     (is (= (count data) (count ((:get-by-domain storage) domain))))

     (is (= 1 (count ((:get-by-domain storage) domain "/b"))))

     (doseq [id ids]

       ((:delete storage) id))

+    (wait-til! (= 0 (count ((:get-by-domain storage) domain))))

     (is (= 0 (count ((:get-by-domain storage) domain))))))

 

 (defn- test-update [storage]

@@ -74,11 +99,14 @@
                                (+ (System/currentTimeMillis) 1)))

         data-neo (assoc data-new :webmention

                         (assoc data-wm :modified

-                               (+ (System/currentTimeMillis) 1)))]

+                               (+ (System/currentTimeMillis) 2)))]

     ((:append storage) data)

+    (wait-til! ((:get storage) id))

     ((:update storage) id data-new)

+    (wait-til! (wm= data-new ((:get storage) id)))

     (is (wm= data-new ((:get storage) id)))

     ((:update storage) source target data-neo)

+    (wait-til! (wm= data-neo ((:get storage) source target)))

     (is (wm= data-neo ((:get storage) source target)))

     ((:delete storage) id)))