From ea40c5dbc2b47d6fd2a23236828dc9e4ab1f77dc Mon Sep 17 00:00:00 2001
From: Ivar Refsdal <ivar.refsdal@nsd.no>
Date: Sat, 4 Sep 2021 13:23:07 +0200
Subject: Initial commit

Add release script

Release 0.1.3

Use com.github.ivarref.yoltq namespace

Use com.github.ivarref.yoltq namespace
---
 test/com/github/ivarref/yoltq/test_utils.clj | 74 ++++++++++++++++++++++++++++
 1 file changed, 74 insertions(+)
 create mode 100644 test/com/github/ivarref/yoltq/test_utils.clj

(limited to 'test/com/github/ivarref/yoltq/test_utils.clj')

diff --git a/test/com/github/ivarref/yoltq/test_utils.clj b/test/com/github/ivarref/yoltq/test_utils.clj
new file mode 100644
index 0000000..dacba68
--- /dev/null
+++ b/test/com/github/ivarref/yoltq/test_utils.clj
@@ -0,0 +1,74 @@
+(ns com.github.ivarref.yoltq.test-utils
+  (:require [com.github.ivarref.yoltq.log-init :as logconfig]
+            [clojure.tools.logging :as log]
+            [com.github.ivarref.yoltq.utils :as u]
+            [com.github.ivarref.yoltq :as dq]
+            [datomic.api :as d]
+            [clojure.string :as str]
+            [com.github.ivarref.yoltq.impl :as i]
+            [clojure.edn :as edn]
+            [com.github.ivarref.yoltq.ext-sys :as ext])
+  (:import (java.util UUID)))
+
+
+(logconfig/init-logging!
+  [[#{"datomic.*" "com.datomic.*" "org.apache.*"} :warn]
+   [#{"*"} (edn/read-string
+             (System/getProperty "TAOENSSO_TIMBRE_MIN_LEVEL_EDN" ":info"))]])
+
+
+(defn empty-conn []
+  (let [uri (str "datomic:mem://hello-world-" (UUID/randomUUID))]
+    (d/delete-database uri)
+    (d/create-database uri)
+    (d/connect uri)))
+
+
+(defn break []
+  (log/info (str/join "*" (repeat 60 ""))))
+
+
+(defn clear []
+  (.print System/out "\033[H\033[2J")
+  (.flush System/out)
+  (break))
+
+
+(defn put-transact! [id payload]
+  @(d/transact (:conn @dq/*config*) [(i/put @dq/*config* id payload)]))
+
+
+(defn advance! [tp]
+  (assert (some? ext/*now-ns-atom*) "Expected to be running in test-mode!")
+  (swap! ext/*now-ns-atom* + (if (number? tp)
+                               tp
+                               (.toNanos tp))))
+
+
+(defn done-count []
+  (d/q '[:find (count ?e) .
+         :where
+         [?e :com.github.ivarref.yoltq/id _]
+         [?e :com.github.ivarref.yoltq/status :done]]
+       (d/db (:conn @dq/*config*))))
+
+
+(defn get-init [& args]
+  (apply u/get-init @dq/*config* args))
+
+
+(defn get-error [& args]
+  (apply u/get-error @dq/*config* args))
+
+
+(defn get-hung [& args]
+  (apply u/get-hung @dq/*config* args))
+
+
+(defn take! [& args]
+  (apply i/take! @dq/*config* args))
+
+
+(defn execute! [& args]
+  (apply i/execute! @dq/*config* args))
+
-- 
cgit v1.2.3


From 60f7371f4d2dd43c5b177039406eeaab00ba27cc Mon Sep 17 00:00:00 2001
From: Ivar Refsdal <ivar.refsdal@nsd.no>
Date: Thu, 23 Sep 2021 11:06:33 +0200
Subject: To test-queue namespace

---
 README.md                                      |  24 +--
 src/com/github/ivarref/yoltq/test_queue.clj    | 197 ++++++++++++++++++++++++
 src/com/github/ivarref/yoltq/virtual_queue.clj | 198 -------------------------
 test/com/github/ivarref/yoltq/test_utils.clj   |  16 +-
 test/com/github/ivarref/yoltq/virtual_test.clj | 140 ++++++++---------
 5 files changed, 287 insertions(+), 288 deletions(-)
 create mode 100644 src/com/github/ivarref/yoltq/test_queue.clj
 delete mode 100644 src/com/github/ivarref/yoltq/virtual_queue.clj

(limited to 'test/com/github/ivarref/yoltq/test_utils.clj')

diff --git a/README.md b/README.md
index 29b6a62..84dc91b 100644
--- a/README.md
+++ b/README.md
@@ -243,23 +243,23 @@ by enabling the virtual queue:
 ```clojure
 ...
 (:require [clojure.test :refer :all]
-          [com.github.ivarref.yoltq :as yq]
-          [com.github.ivarref.yoltq.virtual-queue :as vq])
+  [com.github.ivarref.yoltq :as yq]
+  [com.github.ivarref.yoltq.test-queue :as tq])
 
 ; Enables the virtual queue and disables the threadpool for each test.
 ; yq/start! and yq/stop! becomes a no-op.
-(use-fixtures :each vq/call-with-virtual-queue!)
+(use-fixtures :each tq/call-with-virtual-queue!)
 
 (deftest demo
-  (let [conn ...]
-    (dq/init! {:conn conn}) ; Setup
-    (dq/add-consumer! :q identity)
-    
-    @(d/transact conn [(yq/put :q {:work 123})]) ; Add work
-    
-    ; vq/consume! consumes one job and asserts that it succeeds.
-    ; It returns the return value of the consumer function
-    (is (= {:work 123} (vq/consume! :q)))))
+         (let [conn ...]
+           (yq/init! {:conn conn}) ; Setup
+           (yq/add-consumer! :q identity)
+
+           @(d/transact conn [(yq/put :q {:work 123})]) ; Add work
+
+           ; tq/consume! consumes one job and asserts that it succeeds.
+           ; It returns the return value of the consumer function
+           (is (= {:work 123} (tq/consume! :q)))))
 ```
 
 
diff --git a/src/com/github/ivarref/yoltq/test_queue.clj b/src/com/github/ivarref/yoltq/test_queue.clj
new file mode 100644
index 0000000..6aeb959
--- /dev/null
+++ b/src/com/github/ivarref/yoltq/test_queue.clj
@@ -0,0 +1,197 @@
+(ns com.github.ivarref.yoltq.test-queue
+  (:require [clojure.tools.logging :as log]
+            [com.github.ivarref.yoltq.report-queue :as rq]
+            [com.github.ivarref.yoltq.ext-sys :as ext]
+            [com.github.ivarref.yoltq :as yq]
+            [datomic.api :as d]
+            [com.github.ivarref.yoltq.poller :as poller]
+            [clojure.test :as test]
+            [com.github.ivarref.yoltq.utils :as u]
+            [com.github.ivarref.yoltq.impl :as i])
+  (:import (java.util.concurrent BlockingQueue TimeUnit)
+           (datomic Datom)))
+
+
+(defn bootstrap-poller! [txq running? poller-exited? bootstrapped? conn]
+  (let [ready? (promise)]
+    (future
+      (let [q (d/tx-report-queue conn)]
+        (try
+          (while @running?
+            (when-let [poll-result (.poll ^BlockingQueue q 10 TimeUnit/MILLISECONDS)]
+              (swap! txq conj poll-result))
+            (deliver ready? true)
+            (reset! bootstrapped? true))
+          (catch Throwable t
+            (log/error t "test-poller crashed: " (ex-message t)))
+          (finally
+            (try
+              (d/remove-tx-report-queue conn)
+              (catch Throwable t
+                (log/warn t "remove-tx-report-queue failed:" (ex-message t))))
+            (deliver poller-exited? true)))))
+    @ready?))
+
+
+(defmacro with-virtual-queue!
+  [& body]
+  `(let [txq# (atom [])
+         poller-exited?# (promise)
+         bootstrapped?# (atom false)
+         running?# (atom true)
+         config# (atom {:bootstrap-poller! (partial bootstrap-poller! txq# running?# poller-exited?# bootstrapped?#)
+                        :init-backoff-time 0
+                        :hung-log-level    :warn
+                        :tx-queue          txq#})]
+     (with-bindings {#'yq/*config*       config#
+                     #'yq/*running?*     (atom false)
+                     #'yq/*test-mode*    true
+                     #'ext/*now-ns-atom* (atom 0)
+                     #'ext/*random-atom* (atom 0)
+                     #'ext/*squuid-atom* (atom 0)}
+       (try
+         ~@body
+         (finally
+           (reset! running?# false)
+           (when @bootstrapped?#
+             @poller-exited?#))))))
+
+
+(defn call-with-virtual-queue!
+  [f]
+  (with-virtual-queue!
+    (f)))
+
+
+(defn run-report-queue! [min-items]
+  (let [{:keys [tx-queue conn]} @yq/*config*
+        id-ident (d/q '[:find ?e .
+                        :where [?e :db/ident :com.github.ivarref.yoltq/id]]
+                      (d/db conn))]
+    (let [timeout (+ 3000 (System/currentTimeMillis))]
+      (while (and (< (System/currentTimeMillis) timeout)
+                  (< (count @tx-queue) min-items))
+        (Thread/sleep 10)))
+    (when (< (count @tx-queue) min-items)
+      (let [msg (str "run-report-queue: timeout waiting for " min-items " items")]
+        (log/error msg)
+        (throw (ex-info msg {}))))
+    (let [res (atom [])]
+      (doseq [itm (first (swap-vals! tx-queue (constantly [])))]
+        (rq/process-poll-result!
+          @yq/*config*
+          id-ident
+          itm
+          (fn [f] (swap! res conj (f)))))
+      @res)))
+
+
+(defn run-one-report-queue! []
+  (first (run-report-queue! 1)))
+
+
+(defn run-queue-once! [q status]
+  (poller/poll-once! @yq/*config* q status))
+
+
+(defn put! [q payload]
+  @(d/transact (:conn @yq/*config*) [(yq/put q payload)]))
+
+
+(defn transact-result->maps [{:keys [tx-data db-after]}]
+  (let [m (->> tx-data
+               (group-by (fn [^Datom d] (.e d)))
+               (vals)
+               (mapv (fn [datoms]
+                       (reduce (fn [o ^Datom d]
+                                 (if (.added d)
+                                   (assoc o (d/q '[:find ?r .
+                                                   :in $ ?e
+                                                   :where [?e :db/ident ?r]]
+                                                 db-after
+                                                 (.a d))
+                                            (.v d))
+                                   o))
+                               {}
+                               datoms))))]
+    m))
+
+(defn contains-queue-job?
+  [queue-id conn {::yq/keys [id queue-name status] :as m}]
+  (when (and (= queue-id queue-name)
+             (= status :init)
+             (d/q '[:find ?e .
+                    :in $ ?id
+                    :where
+                    [?e ::yq/id ?id]
+                    [?e ::yq/status :init]]
+                  (d/db conn)
+                  id))
+    m))
+
+
+(defn get-tx-q-job [q-id]
+  (let [{:keys [tx-queue conn]} @yq/*config*]
+    (loop [timeout (+ 3000 (System/currentTimeMillis))]
+      (if-let [job (->> @tx-queue
+                        (mapcat transact-result->maps)
+                        (filter (partial contains-queue-job? q-id conn))
+                        (first))]
+        (u/get-queue-item (d/db conn) (::yq/id job))
+        (if (< (System/currentTimeMillis) timeout)
+          (do (Thread/sleep 10)
+              (recur timeout))
+          nil)))))
+
+(defmacro consume-expect! [queue-name expected-status]
+  `(if-let [job# (get-tx-q-job ~queue-name)]
+     (try
+       (with-bindings (:com.github.ivarref.yoltq/bindings job#)
+         (let [res# (some->> (u/prepare-processing (d/db (:conn @yq/*config*))
+                                                   (:com.github.ivarref.yoltq/id job#)
+                                                   ~queue-name
+                                                   (:com.github.ivarref.yoltq/lock job#)
+                                                   (:com.github.ivarref.yoltq/status job#))
+                             (i/take! @yq/*config*)
+                             (i/execute! @yq/*config*))]
+           (test/is (= ~expected-status (:com.github.ivarref.yoltq/status res#)))
+           (if (:retval res#)
+             (:retval res#)
+             (:exception res#))))
+       (catch Throwable t#
+         (log/error t# "unexpected error in consume-expect:" (ex-message t#))))
+     (test/is false (str "No job found for queue " ~queue-name))))
+
+(defmacro consume! [queue-name]
+  `(consume-expect! ~queue-name :done))
+
+
+(defn mark-fails! [{:keys [conn]}
+                   {:com.github.ivarref.yoltq/keys [id lock tries]}
+                   _]
+  (try
+    (let [tx [[:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/lock lock (u/random-uuid)]
+              [:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/tries tries (inc tries)]
+              [:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/status u/status-processing :init]]]
+      @(d/transact conn tx)
+      nil)
+    (catch Throwable t
+      (log/error t "unexpected error in mark-status!: " (ex-message t))
+      nil)))
+
+
+(defmacro force-retry! [queue-name]
+  `(if-let [job# (get-tx-q-job ~queue-name)]
+     (try
+       (with-bindings (:com.github.ivarref.yoltq/bindings job#)
+         (some->> (u/prepare-processing (d/db (:conn @yq/*config*))
+                                        (:com.github.ivarref.yoltq/id job#)
+                                        ~queue-name
+                                        (:com.github.ivarref.yoltq/lock job#)
+                                        (:com.github.ivarref.yoltq/status job#))
+                  (i/take! @yq/*config*)
+                  (i/execute! (assoc @yq/*config* :mark-status-fn! mark-fails!)))
+         (consume! ~queue-name))
+       (catch Throwable t#
+         (log/error t# "unexpected error in consume-twice!:" (ex-message t#))))
+     (test/is false (str "No job found for queue " ~queue-name))))
\ No newline at end of file
diff --git a/src/com/github/ivarref/yoltq/virtual_queue.clj b/src/com/github/ivarref/yoltq/virtual_queue.clj
deleted file mode 100644
index db429a8..0000000
--- a/src/com/github/ivarref/yoltq/virtual_queue.clj
+++ /dev/null
@@ -1,198 +0,0 @@
-(ns com.github.ivarref.yoltq.virtual-queue
-  (:require [clojure.tools.logging :as log]
-            [com.github.ivarref.yoltq.report-queue :as rq]
-            [com.github.ivarref.yoltq.ext-sys :as ext]
-            [com.github.ivarref.yoltq :as yq]
-            [datomic.api :as d]
-            [com.github.ivarref.yoltq.poller :as poller]
-            [clojure.test :as test]
-            [com.github.ivarref.yoltq.utils :as u]
-            [com.github.ivarref.yoltq.impl :as i])
-  (:import (java.util.concurrent BlockingQueue TimeUnit)
-           (datomic Datom)))
-
-
-(defn bootstrap-poller! [txq running? poller-exited? bootstrapped? conn]
-  (let [ready? (promise)]
-    (future
-      (let [q (d/tx-report-queue conn)]
-        (try
-          (while @running?
-            (when-let [poll-result (.poll ^BlockingQueue q 10 TimeUnit/MILLISECONDS)]
-              (swap! txq conj poll-result))
-            (deliver ready? true)
-            (reset! bootstrapped? true))
-          (catch Throwable t
-            (log/error t "test-poller crashed: " (ex-message t)))
-          (finally
-            (try
-              (d/remove-tx-report-queue conn)
-              (catch Throwable t
-                (log/warn t "remove-tx-report-queue failed:" (ex-message t))))
-            (deliver poller-exited? true)))))
-    @ready?))
-
-
-(defmacro with-virtual-queue!
-  [& body]
-  `(let [txq# (atom [])
-         poller-exited?# (promise)
-         bootstrapped?# (atom false)
-         running?# (atom true)
-         config# (atom {:bootstrap-poller! (partial bootstrap-poller! txq# running?# poller-exited?# bootstrapped?#)
-                        :init-backoff-time 0
-                        :hung-log-level    :warn
-                        :tx-queue          txq#})]
-     (with-bindings {#'yq/*config*       config#
-                     #'yq/*running?*     (atom false)
-                     #'yq/*test-mode*    true
-                     #'ext/*now-ns-atom* (atom 0)
-                     #'ext/*random-atom* (atom 0)
-                     #'ext/*squuid-atom* (atom 0)}
-       (try
-         ~@body
-         (finally
-           (reset! running?# false)
-           (when @bootstrapped?#
-             @poller-exited?#))))))
-
-
-(defn call-with-virtual-queue!
-  [f]
-  (with-virtual-queue!
-    (f)))
-
-
-(defn run-report-queue! [min-items]
-  (let [{:keys [tx-queue conn]} @yq/*config*
-        id-ident (d/q '[:find ?e .
-                        :where [?e :db/ident :com.github.ivarref.yoltq/id]]
-                      (d/db conn))]
-    (let [timeout (+ 3000 (System/currentTimeMillis))]
-      (while (and (< (System/currentTimeMillis) timeout)
-                  (< (count @tx-queue) min-items))
-        (Thread/sleep 10)))
-    (when (< (count @tx-queue) min-items)
-      (let [msg (str "run-report-queue: timeout waiting for " min-items " items")]
-        (log/error msg)
-        (throw (ex-info msg {}))))
-    (let [res (atom [])]
-      (doseq [itm (first (swap-vals! tx-queue (constantly [])))]
-        (rq/process-poll-result!
-          @yq/*config*
-          id-ident
-          itm
-          (fn [f] (swap! res conj (f)))))
-      @res)))
-
-
-(defn run-one-report-queue! []
-  (first (run-report-queue! 1)))
-
-
-(defn run-queue-once! [q status]
-  (poller/poll-once! @yq/*config* q status))
-
-
-(defn put! [q payload]
-  @(d/transact (:conn @yq/*config*) [(yq/put q payload)]))
-
-
-(defn transact-result->maps [{:keys [tx-data db-after]}]
-  (let [m (->> tx-data
-               (group-by (fn [^Datom d] (.e d)))
-               (vals)
-               (mapv (fn [datoms]
-                       (reduce (fn [o ^Datom d]
-                                 (if (.added d)
-                                   (assoc o (d/q '[:find ?r .
-                                                   :in $ ?e
-                                                   :where [?e :db/ident ?r]]
-                                                 db-after
-                                                 (.a d))
-                                            (.v d))
-                                   o))
-                               {}
-                               datoms))))]
-    m))
-
-(defn contains-queue-job?
-  [queue-id conn {::yq/keys [id queue-name status] :as m}]
-  (when (and (= queue-id queue-name)
-             (= status :init)
-             (d/q '[:find ?e .
-                    :in $ ?id
-                    :where
-                    [?e ::yq/id ?id]
-                    [?e ::yq/status :init]]
-                  (d/db conn)
-                  id))
-    m))
-
-
-(defn get-tx-q-job [q-id]
-  (let [{:keys [tx-queue conn]} @yq/*config*]
-    (loop [timeout (+ 3000 (System/currentTimeMillis))]
-      (if-let [job (->> @tx-queue
-                        (mapcat transact-result->maps)
-                        (filter (partial contains-queue-job? q-id conn))
-                        (first))]
-        (u/get-queue-item (d/db conn) (::yq/id job))
-        (if (< (System/currentTimeMillis) timeout)
-          (do (Thread/sleep 10)
-              (recur timeout))
-          nil)))))
-
-(defmacro consume-expect! [queue-name expected-status]
-  `(if-let [job# (get-tx-q-job ~queue-name)]
-     (try
-       (with-bindings (:com.github.ivarref.yoltq/bindings job#)
-         (let [res# (some->> (u/prepare-processing (d/db (:conn @yq/*config*))
-                                                   (:com.github.ivarref.yoltq/id job#)
-                                                   ~queue-name
-                                                   (:com.github.ivarref.yoltq/lock job#)
-                                                   (:com.github.ivarref.yoltq/status job#))
-                             (i/take! @yq/*config*)
-                             (i/execute! @yq/*config*))]
-           (test/is (= ~expected-status (:com.github.ivarref.yoltq/status res#)))
-           (if (:retval res#)
-             (:retval res#)
-             (:exception res#))))
-       (catch Throwable t#
-         (log/error t# "unexpected error in consume-expect:" (ex-message t#))))
-     (test/is false (str "No job found for queue " ~queue-name))))
-tx-spent-time!
-
-(defmacro consume! [queue-name]
-  `(consume-expect! ~queue-name :done))
-
-
-(defn mark-fails! [{:keys [conn]}
-                   {:com.github.ivarref.yoltq/keys [id lock tries]}
-                   _]
-  (try
-    (let [tx [[:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/lock lock (u/random-uuid)]
-              [:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/tries tries (inc tries)]
-              [:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/status u/status-processing :init]]]
-      @(d/transact conn tx)
-      nil)
-    (catch Throwable t
-      (log/error t "unexpected error in mark-status!: " (ex-message t))
-      nil)))
-
-
-(defmacro consume-twice! [queue-name]
-  `(if-let [job# (get-tx-q-job ~queue-name)]
-     (try
-       (with-bindings (:com.github.ivarref.yoltq/bindings job#)
-         (some->> (u/prepare-processing (d/db (:conn @yq/*config*))
-                                        (:com.github.ivarref.yoltq/id job#)
-                                        ~queue-name
-                                        (:com.github.ivarref.yoltq/lock job#)
-                                        (:com.github.ivarref.yoltq/status job#))
-                  (i/take! @yq/*config*)
-                  (i/execute! (assoc @yq/*config* :mark-status-fn! mark-fails!)))
-         (consume! ~queue-name))
-       (catch Throwable t#
-         (log/error t# "unexpected error in consume-twice!:" (ex-message t#))))
-     (test/is false (str "No job found for queue " ~queue-name))))
\ No newline at end of file
diff --git a/test/com/github/ivarref/yoltq/test_utils.clj b/test/com/github/ivarref/yoltq/test_utils.clj
index dacba68..df56460 100644
--- a/test/com/github/ivarref/yoltq/test_utils.clj
+++ b/test/com/github/ivarref/yoltq/test_utils.clj
@@ -2,7 +2,7 @@
   (:require [com.github.ivarref.yoltq.log-init :as logconfig]
             [clojure.tools.logging :as log]
             [com.github.ivarref.yoltq.utils :as u]
-            [com.github.ivarref.yoltq :as dq]
+            [com.github.ivarref.yoltq :as yq]
             [datomic.api :as d]
             [clojure.string :as str]
             [com.github.ivarref.yoltq.impl :as i]
@@ -35,7 +35,7 @@
 
 
 (defn put-transact! [id payload]
-  @(d/transact (:conn @dq/*config*) [(i/put @dq/*config* id payload)]))
+  @(d/transact (:conn @yq/*config*) [(i/put @yq/*config* id payload)]))
 
 
 (defn advance! [tp]
@@ -50,25 +50,25 @@
          :where
          [?e :com.github.ivarref.yoltq/id _]
          [?e :com.github.ivarref.yoltq/status :done]]
-       (d/db (:conn @dq/*config*))))
+       (d/db (:conn @yq/*config*))))
 
 
 (defn get-init [& args]
-  (apply u/get-init @dq/*config* args))
+  (apply u/get-init @yq/*config* args))
 
 
 (defn get-error [& args]
-  (apply u/get-error @dq/*config* args))
+  (apply u/get-error @yq/*config* args))
 
 
 (defn get-hung [& args]
-  (apply u/get-hung @dq/*config* args))
+  (apply u/get-hung @yq/*config* args))
 
 
 (defn take! [& args]
-  (apply i/take! @dq/*config* args))
+  (apply i/take! @yq/*config* args))
 
 
 (defn execute! [& args]
-  (apply i/execute! @dq/*config* args))
+  (apply i/execute! @yq/*config* args))
 
diff --git a/test/com/github/ivarref/yoltq/virtual_test.clj b/test/com/github/ivarref/yoltq/virtual_test.clj
index 5e5fc92..442acac 100644
--- a/test/com/github/ivarref/yoltq/virtual_test.clj
+++ b/test/com/github/ivarref/yoltq/virtual_test.clj
@@ -1,8 +1,7 @@
 (ns com.github.ivarref.yoltq.virtual-test
   (:require [datomic-schema.core]
             [clojure.test :refer :all]
-            [com.github.ivarref.yoltq.virtual-queue :as vq]
-            [com.github.ivarref.yoltq :as dq]
+            [com.github.ivarref.yoltq.test-queue :as tq]
             [com.github.ivarref.yoltq.test-utils :as u]
             [datomic.api :as d]
             [com.github.ivarref.yoltq.utils :as uu]
@@ -12,29 +11,29 @@
             [taoensso.timbre :as timbre]))
 
 
-(use-fixtures :each vq/call-with-virtual-queue!)
+(use-fixtures :each tq/call-with-virtual-queue!)
 
 
 (deftest happy-case-1
   (let [conn (u/empty-conn)]
-    (dq/init! {:conn conn})
-    (dq/add-consumer! :q identity)
-    @(d/transact conn [(dq/put :q {:work 123})])
-    (is (= {:work 123} (vq/consume! :q)))))
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q identity)
+    @(d/transact conn [(yq/put :q {:work 123})])
+    (is (= {:work 123} (tq/consume! :q)))))
 
 
 (deftest happy-case-tx-report-q
   (let [conn (u/empty-conn)]
-    (dq/init! {:conn conn})
-    (dq/add-consumer! :q identity)
-    @(d/transact conn [(dq/put :q {:work 123})])
-    (is (= {:work 123} (:retval (vq/run-one-report-queue!))))
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q identity)
+    @(d/transact conn [(yq/put :q {:work 123})])
+    (is (= {:work 123} (:retval (tq/run-one-report-queue!))))
     (is (= 1 (u/done-count)))))
 
 
 (deftest happy-case-poller
   (let [conn (u/empty-conn)]
-    (dq/init! {:conn     conn
+    (yq/init! {:conn     conn
                :handlers {:q {:f (fn [payload] payload)}}})
     (u/put-transact! :q {:work 123})
     (u/advance! (:init-backoff-time yq/default-opts))
@@ -47,8 +46,8 @@
 (deftest happy-case-queue-fn-allow-cas-failure
   (let [conn (u/empty-conn)
         invoke-count (atom 0)]
-    (dq/init! {:conn conn})
-    (dq/add-consumer! :q
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q
                       (fn [{:keys [id]}]
                         (swap! invoke-count inc)
                         @(d/transact conn [[:db/cas [:e/id id] :e/val nil "janei"]]))
@@ -56,14 +55,14 @@
     @(d/transact conn #d/schema [[:e/id :one :string :id]
                                  [:e/val :one :string]])
     @(d/transact conn [{:e/id "demo"}
-                       (dq/put :q {:id "demo"})])
+                       (yq/put :q {:id "demo"})])
     (u/advance! (:init-backoff-time yq/default-opts))
-    (swap! dq/*config* assoc :mark-status-fn! (fn [_ _ new-status]
+    (swap! yq/*config* assoc :mark-status-fn! (fn [_ _ new-status]
                                                 (log/info "mark-status! doing nothing for new status" new-status)))
     (is (nil? (some->> (u/get-init :q)
                        (u/take!)
                        (u/execute!))))
-    (swap! dq/*config* dissoc :mark-status-fn!)
+    (swap! yq/*config* dissoc :mark-status-fn!)
 
     ; (mark-status! :done) failed but f succeeded.
     (is (nil? (u/get-hung :q)))
@@ -78,7 +77,7 @@
 
 (deftest backoff-test
   (let [conn (u/empty-conn)]
-    (dq/init! {:conn              conn
+    (yq/init! {:conn              conn
                :init-backoff-time (:init-backoff-time yq/default-opts)
                :handlers          {:q {:f (fn [_] (throw (ex-info "janei" {})))}}})
     (u/put-transact! :q {:work 123})
@@ -102,7 +101,7 @@
 
 (deftest get-hung-test
   (let [conn (u/empty-conn)]
-    (dq/init! {:conn              conn
+    (yq/init! {:conn              conn
                :init-backoff-time (:init-backoff-time yq/default-opts)
                :handlers          {:q {:f (fn [_] (throw (ex-info "janei" {})))}}})
     (u/put-transact! :q {:work 123})
@@ -122,7 +121,7 @@
 
 (deftest basic-locking
   (let [conn (u/empty-conn)]
-    (dq/init! {:conn              conn
+    (yq/init! {:conn              conn
                :init-backoff-time (:init-backoff-time yq/default-opts)
                :cas-failures      (atom 0)
                :handlers          {:q {:f (fn [_] (throw (ex-info "janei" {})))}}})
@@ -133,12 +132,12 @@
     (let [job (u/get-init :q)]
       (is (= :processing (some->> job (u/take!) :com.github.ivarref.yoltq/status)))
       (u/take! job)
-      (is (= 1 @(:cas-failures @dq/*config*))))))
+      (is (= 1 @(:cas-failures @yq/*config*))))))
 
 
 (deftest retry-test
   (let [conn (u/empty-conn)]
-    (dq/init! {:conn              conn
+    (yq/init! {:conn              conn
                :init-backoff-time (:init-backoff-time yq/default-opts)
                :handlers          {:q {:f
                                        (let [c (atom 0)]
@@ -161,34 +160,34 @@
 (deftest max-retries-test
   (let [conn (u/empty-conn)
         call-count (atom 0)]
-    (dq/init! {:conn               conn
+    (yq/init! {:conn               conn
                :error-backoff-time 0})
-    (dq/add-consumer! :q (fn [_]
+    (yq/add-consumer! :q (fn [_]
                            (swap! call-count inc)
                            (throw (ex-info "janei" {})))
                       {:max-retries 1})
-    (vq/put! :q {:work 123})
-    (is (some? (:exception (vq/run-one-report-queue!))))
+    (tq/put! :q {:work 123})
+    (is (some? (:exception (tq/run-one-report-queue!))))
 
     (dotimes [_ 10]
-      (vq/run-queue-once! :q :error))
+      (tq/run-queue-once! :q :error))
     (is (= 2 @call-count))))
 
 
 (deftest max-retries-test-two
   (let [conn (u/empty-conn)
         call-count (atom 0)]
-    (dq/init! {:conn               conn
+    (yq/init! {:conn               conn
                :error-backoff-time 0})
-    (dq/add-consumer! :q (fn [_]
+    (yq/add-consumer! :q (fn [_]
                            (swap! call-count inc)
                            (throw (ex-info "janei" {})))
                       {:max-retries 3})
-    (vq/put! :q {:work 123})
-    (is (some? (:exception (vq/run-one-report-queue!))))
+    (tq/put! :q {:work 123})
+    (is (some? (:exception (tq/run-one-report-queue!))))
 
     (dotimes [_ 20]
-      (vq/run-queue-once! :q :error))
+      (tq/run-queue-once! :q :error))
     (is (= 4 @call-count))))
 
 
@@ -196,55 +195,55 @@
   (let [conn (u/empty-conn)
         call-count (atom 0)
         missed-mark-status (atom nil)]
-    (dq/init! {:conn conn})
-    (dq/add-consumer! :q
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q
                       (fn [_]
                         (if (= 1 (swap! call-count inc))
                           (throw (ex-info "error" {}))
                           (log/info "return OK"))))
-    (vq/put! :q {:id "demo"})
-    (vq/run-one-report-queue!)                              ; now in status :error
+    (tq/put! :q {:id "demo"})
+    (tq/run-one-report-queue!)                              ; now in status :error
 
 
-    (swap! dq/*config* assoc :mark-status-fn! (fn [_ _ new-status]
+    (swap! yq/*config* assoc :mark-status-fn! (fn [_ _ new-status]
                                                 (reset! missed-mark-status new-status)
                                                 (log/info "mark-status! doing nothing for new status" new-status)))
     (u/advance! (:error-backoff-time @yq/*config*))
-    (vq/run-queue-once! :q :error)
-    (swap! dq/*config* dissoc :mark-status-fn!)
+    (tq/run-queue-once! :q :error)
+    (swap! yq/*config* dissoc :mark-status-fn!)
     (is (= :done @missed-mark-status))
 
-    (is (nil? (uu/get-hung @dq/*config* :q)))
+    (is (nil? (uu/get-hung @yq/*config* :q)))
     (u/advance! (:hung-backoff-time @yq/*config*))
 
-    (is (some? (uu/get-hung @dq/*config* :q)))
+    (is (some? (uu/get-hung @yq/*config* :q)))
 
     (is (= 2 @call-count))
 
-    (is (true? (some->> (uu/get-hung (assoc-in @dq/*config* [:handlers :q :max-retries] 1) :q)
-                        (i/take! @dq/*config*)
-                        (i/execute! @dq/*config*)
+    (is (true? (some->> (uu/get-hung (assoc-in @yq/*config* [:handlers :q :max-retries] 1) :q)
+                        (i/take! @yq/*config*)
+                        (i/execute! @yq/*config*)
                         :failed?)))
 
     (u/advance! (:error-backoff-time @yq/*config*))
-    (is (some? (uu/get-error @dq/*config* :q)))
-    (is (nil? (uu/get-error (assoc-in @dq/*config* [:handlers :q :max-retries] 1) :q)))))
+    (is (some? (uu/get-error @yq/*config* :q)))
+    (is (nil? (uu/get-error (assoc-in @yq/*config* [:handlers :q :max-retries] 1) :q)))))
 
 
 (deftest consume-expect-test
   (let [conn (u/empty-conn)
         seen (atom #{})]
-    (dq/init! {:conn conn})
-    (dq/add-consumer! :q (fn [payload]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q (fn [payload]
                            (when (= #{1 2} (swap! seen conj payload))
                              (throw (ex-info "oops" {})))
                            payload))
 
-    @(d/transact conn [(dq/put :q 1)])
-    @(d/transact conn [(dq/put :q 2)])
+    @(d/transact conn [(yq/put :q 1)])
+    @(d/transact conn [(yq/put :q 2)])
 
-    (is (= 1 (vq/consume-expect! :q :done)))
-    (vq/consume-expect! :q :error)))
+    (is (= 1 (tq/consume-expect! :q :done)))
+    (tq/consume-expect! :q :error)))
 
 
 (def ^:dynamic *some-binding* nil)
@@ -252,34 +251,35 @@
 
 (deftest binding-test
   (let [conn (u/empty-conn)]
-    (dq/init! {:conn conn
+    (yq/init! {:conn             conn
                :capture-bindings [#'*some-binding* #'timbre/*context*]})
-    (dq/add-consumer! :q (fn [_] *some-binding*))
+    (yq/add-consumer! :q (fn [_] *some-binding*))
     (binding [timbre/*context* {:x-request-id "wooho"}]
       (binding [*some-binding* 1]
-        @(d/transact conn [(dq/put :q nil)]))
+        @(d/transact conn [(yq/put :q nil)]))
       (binding [*some-binding* 2]
-        @(d/transact conn [(dq/put :q nil)]))
-      @(d/transact conn [(dq/put :q nil)]))
+        @(d/transact conn [(yq/put :q nil)]))
+      @(d/transact conn [(yq/put :q nil)]))
 
-    (is (= 1 (vq/consume-expect! :q :done)))
-    (is (= 2 (vq/consume-expect! :q :done)))
-    (is (nil? (vq/consume-expect! :q :done)))))
+    (is (= 1 (tq/consume-expect! :q :done)))
+    (is (= 2 (tq/consume-expect! :q :done)))
+    (is (nil? (tq/consume-expect! :q :done)))))
 
 
 (deftest default-binding-test
   (let [conn (u/empty-conn)]
-    (dq/init! {:conn conn})
-    (dq/add-consumer! :q (fn [_] (:x-request-id timbre/*context*)))
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q (fn [_] (:x-request-id timbre/*context*)))
     (binding [timbre/*context* {:x-request-id "123"}]
-      @(d/transact conn [(dq/put :q nil)]))
-    (is (= "123" (vq/consume-expect! :q :done)))))
+      @(d/transact conn [(yq/put :q nil)]))
+    (is (= "123" (tq/consume-expect! :q :done)))))
 
 
-(deftest consume-twice
-  (let [conn (u/empty-conn)
-        cnt (atom 0)]
+(deftest force-retry-test
+  (let [conn (u/empty-conn)]
     (yq/init! {:conn conn})
-    (yq/add-consumer! :q (fn [_] (swap! cnt inc)))
-    @(d/transact conn [(dq/put :q nil)])
-    (is (= 2 (vq/consume-twice! :q)))))
+    (yq/add-consumer! :q (let [cnt (atom 0)]
+                           (fn [_] (swap! cnt inc))))
+    @(d/transact conn [(yq/put :q nil)])
+    (is (= 1 (tq/consume! :q)))
+    #_(is (= 2 (tq/force-retry! :q)))))
-- 
cgit v1.2.3


From dc2e14b4e1e91e6fefecc01c312a44c0033640c9 Mon Sep 17 00:00:00 2001
From: Ivar Refsdal <ivar.refsdal@nsd.no>
Date: Thu, 23 Sep 2021 13:01:23 +0200
Subject: Basic depends-on works for test queue

---
 src/com/github/ivarref/yoltq.clj               | 12 +++---
 src/com/github/ivarref/yoltq/impl.clj          | 51 +++++++++++++++++++++-----
 src/com/github/ivarref/yoltq/test_queue.clj    | 27 ++++++++------
 src/com/github/ivarref/yoltq/utils.clj         |  1 +
 test/com/github/ivarref/yoltq/test_utils.clj   |  2 +-
 test/com/github/ivarref/yoltq/virtual_test.clj | 27 ++++++++++++++
 6 files changed, 92 insertions(+), 28 deletions(-)

(limited to 'test/com/github/ivarref/yoltq/test_utils.clj')

diff --git a/src/com/github/ivarref/yoltq.clj b/src/com/github/ivarref/yoltq.clj
index 58efca1..3164020 100644
--- a/src/com/github/ivarref/yoltq.clj
+++ b/src/com/github/ivarref/yoltq.clj
@@ -94,11 +94,13 @@
    (swap! *config* (fn [old-config] (assoc-in old-config [:handlers queue-id] (merge opts {:f f}))))))
 
 
-(defn put [queue-id payload]
-  (let [{:keys [bootstrap-poller! conn] :as cfg} @*config*]
-    (when (and *test-mode* bootstrap-poller!)
-      (bootstrap-poller! conn))
-    (i/put cfg queue-id payload)))
+(defn put
+  ([queue-id payload] (put queue-id payload {}))
+  ([queue-id payload opts]
+   (let [{:keys [bootstrap-poller! conn] :as cfg} @*config*]
+     (when (and *test-mode* bootstrap-poller!)
+       (bootstrap-poller! conn))
+     (i/put cfg queue-id payload opts))))
 
 
 (defn- do-start! []
diff --git a/src/com/github/ivarref/yoltq/impl.clj b/src/com/github/ivarref/yoltq/impl.clj
index 9c95cff..9811c93 100644
--- a/src/com/github/ivarref/yoltq/impl.clj
+++ b/src/com/github/ivarref/yoltq/impl.clj
@@ -8,9 +8,11 @@
 
 (def schema
   [#:db{:ident :com.github.ivarref.yoltq/id, :cardinality :db.cardinality/one, :valueType :db.type/uuid, :unique :db.unique/identity}
+   #:db{:ident :com.github.ivarref.yoltq/ext-id, :cardinality :db.cardinality/one, :valueType :db.type/string, :unique :db.unique/value}
    #:db{:ident :com.github.ivarref.yoltq/queue-name, :cardinality :db.cardinality/one, :valueType :db.type/keyword, :index true}
    #:db{:ident :com.github.ivarref.yoltq/status, :cardinality :db.cardinality/one, :valueType :db.type/keyword, :index true}
    #:db{:ident :com.github.ivarref.yoltq/payload, :cardinality :db.cardinality/one, :valueType :db.type/string}
+   #:db{:ident :com.github.ivarref.yoltq/opts, :cardinality :db.cardinality/one, :valueType :db.type/string}
    #:db{:ident :com.github.ivarref.yoltq/bindings, :cardinality :db.cardinality/one, :valueType :db.type/string}
    #:db{:ident :com.github.ivarref.yoltq/tries, :cardinality :db.cardinality/one, :valueType :db.type/long, :noHistory true}
    #:db{:ident :com.github.ivarref.yoltq/lock, :cardinality :db.cardinality/one, :valueType :db.type/uuid, :noHistory true}
@@ -20,8 +22,10 @@
    #:db{:ident :com.github.ivarref.yoltq/error-time, :cardinality :db.cardinality/one, :valueType :db.type/long}])
 
 
-(defn put [{:keys [capture-bindings] :as config}
-           queue-name payload]
+(defn put [{:keys [capture-bindings conn] :as config}
+           queue-name
+           payload
+           opts]
   (if-let [_ (get-in config [:handlers queue-name])]
     (let [id (u/squuid)
           str-bindings (->> (reduce (fn [o k]
@@ -30,19 +34,46 @@
                                     (or capture-bindings []))
                             (pr-str))]
       (log/debug "queue item" (str id) "for queue" queue-name "is pending status" u/status-init)
-      {:com.github.ivarref.yoltq/id         id
-       :com.github.ivarref.yoltq/queue-name queue-name
-       :com.github.ivarref.yoltq/status     u/status-init
-       :com.github.ivarref.yoltq/payload    (pr-str payload)
-       :com.github.ivarref.yoltq/bindings   str-bindings
-       :com.github.ivarref.yoltq/lock       (u/random-uuid)
-       :com.github.ivarref.yoltq/tries      0
-       :com.github.ivarref.yoltq/init-time  (u/now-ns)})
+      (merge
+        {:com.github.ivarref.yoltq/id         id
+         :com.github.ivarref.yoltq/queue-name queue-name
+         :com.github.ivarref.yoltq/status     u/status-init
+         :com.github.ivarref.yoltq/payload    (pr-str payload)
+         :com.github.ivarref.yoltq/bindings   str-bindings
+         :com.github.ivarref.yoltq/opts       (pr-str (or opts {}))
+         :com.github.ivarref.yoltq/lock       (u/random-uuid)
+         :com.github.ivarref.yoltq/tries      0
+         :com.github.ivarref.yoltq/init-time  (u/now-ns)}
+        (when-let [[q ext-id] (:depends-on opts)]
+          (when-not (d/q '[:find ?e .
+                           :in $ ?ext-id
+                           :where
+                           [?e :com.github.ivarref.yoltq/ext-id ?ext-id]]
+                         (d/db conn)
+                         (pr-str [q ext-id]))
+            (throw (ex-info ":depends-on not found in database" opts))))
+        (when-let [ext-id (:id opts)]
+          {:com.github.ivarref.yoltq/ext-id (pr-str [queue-name ext-id])})))
     (do
       (log/error "Did not find registered handler for queue" queue-name)
       (throw (ex-info (str "Did not find registered handler for queue: " queue-name) {:queue queue-name})))))
 
 
+(defn depends-on-waiting? [{:keys [conn]}
+                           {:keys [id]}]
+  (let [db (d/db conn)]
+    (when-let [{:com.github.ivarref.yoltq/keys [opts]} (u/get-queue-item db id)]
+      (when-let [[q id :as depends-on] (:depends-on opts)]
+        (when-not (d/q '[:find ?e .
+                         :in $ ?ext-id
+                         :where
+                         [?e :com.github.ivarref.yoltq/ext-id ?ext-id]
+                         [?e :com.github.ivarref.yoltq/status :done]]
+                       db
+                       (pr-str [q id]))
+          {:depends-on depends-on})))))
+
+
 (defn take! [{:keys [conn cas-failures hung-log-level tx-spent-time!]
               :or   {hung-log-level :error}}
              {:keys [tx id queue-name was-hung? to-error?] :as queue-item-info}]
diff --git a/src/com/github/ivarref/yoltq/test_queue.clj b/src/com/github/ivarref/yoltq/test_queue.clj
index 4c4f903..6183216 100644
--- a/src/com/github/ivarref/yoltq/test_queue.clj
+++ b/src/com/github/ivarref/yoltq/test_queue.clj
@@ -148,18 +148,21 @@
   `(if-let [job# (get-tx-q-job ~queue-name)]
      (try
        (with-bindings (:com.github.ivarref.yoltq/bindings job#)
-         (let [res# (some->> (u/prepare-processing (d/db (:conn @yq/*config*))
-                                                   (:com.github.ivarref.yoltq/id job#)
-                                                   ~queue-name
-                                                   (:com.github.ivarref.yoltq/lock job#)
-                                                   (:com.github.ivarref.yoltq/status job#))
-                             (i/take! @yq/*config*)
-                             (i/execute! @yq/*config*))]
-           (swap! (:prev-consumed @yq/*config*) assoc ~queue-name res#)
-           (test/is (= ~expected-status (:com.github.ivarref.yoltq/status res#)))
-           (if (:retval res#)
-             (:retval res#)
-             (:exception res#))))
+         (let [prep# (u/prepare-processing (d/db (:conn @yq/*config*))
+                                           (:com.github.ivarref.yoltq/id job#)
+                                           ~queue-name
+                                           (:com.github.ivarref.yoltq/lock job#)
+                                           (:com.github.ivarref.yoltq/status job#))]
+           (if-let [depends-on# (i/depends-on-waiting? @yq/*config* prep#)]
+             depends-on#
+             (let [res# (some->> prep#
+                                 (i/take! @yq/*config*)
+                                 (i/execute! @yq/*config*))]
+               (swap! (:prev-consumed @yq/*config*) assoc ~queue-name res#)
+               (test/is (= ~expected-status (:com.github.ivarref.yoltq/status res#)))
+               (if (:retval res#)
+                 (:retval res#)
+                 (:exception res#))))))
        (catch Throwable t#
          (log/error t# "unexpected error in consume-expect:" (ex-message t#))
          (throw t#)))
diff --git a/src/com/github/ivarref/yoltq/utils.clj b/src/com/github/ivarref/yoltq/utils.clj
index 9501343..d551510 100644
--- a/src/com/github/ivarref/yoltq/utils.clj
+++ b/src/com/github/ivarref/yoltq/utils.clj
@@ -56,6 +56,7 @@
   (-> (d/pull db '[:*] [:com.github.ivarref.yoltq/id id])
       (dissoc :db/id)
       (update :com.github.ivarref.yoltq/payload edn/read-string)
+      (update :com.github.ivarref.yoltq/opts (fn [s] (or (when s (edn/read-string s)) {})))
       (update :com.github.ivarref.yoltq/bindings
               (fn [s]
                 (when s
diff --git a/test/com/github/ivarref/yoltq/test_utils.clj b/test/com/github/ivarref/yoltq/test_utils.clj
index df56460..5427ff5 100644
--- a/test/com/github/ivarref/yoltq/test_utils.clj
+++ b/test/com/github/ivarref/yoltq/test_utils.clj
@@ -35,7 +35,7 @@
 
 
 (defn put-transact! [id payload]
-  @(d/transact (:conn @yq/*config*) [(i/put @yq/*config* id payload)]))
+  @(d/transact (:conn @yq/*config*) [(i/put @yq/*config* id payload {})]))
 
 
 (defn advance! [tp]
diff --git a/test/com/github/ivarref/yoltq/virtual_test.clj b/test/com/github/ivarref/yoltq/virtual_test.clj
index 2b67e5e..789e5b4 100644
--- a/test/com/github/ivarref/yoltq/virtual_test.clj
+++ b/test/com/github/ivarref/yoltq/virtual_test.clj
@@ -284,3 +284,30 @@
     (is (= 1 (tq/consume! :q)))
     (is (= 2 (tq/force-retry! :q)))
     (is (= 3 (tq/force-retry! :q)))))
+
+
+(deftest ext-id-no-duplicates
+  (let [conn (u/empty-conn)]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q identity)
+    @(d/transact conn [(yq/put :q nil {:id "123"})])
+    (is (thrown? Exception @(d/transact conn [(yq/put :q nil {:id "123"})])))))
+
+
+(deftest depends-on
+  (let [conn (u/empty-conn)]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :a identity)
+    (yq/add-consumer! :b identity)
+    @(d/transact conn [(yq/put :a "a" {:id "1"})])
+    (is (thrown? Exception @(d/transact conn [(yq/put :b "b" {:depends-on [:a "0"]})])))
+    @(d/transact conn [(yq/put :b "b" {:depends-on [:a "1"]})])
+
+    ; can't consume :b yet:
+    (is (= {:depends-on [:a "1"]} (tq/consume! :b)))
+    (is (= {:depends-on [:a "1"]} (tq/consume! :b)))
+
+    (is (= "a" (tq/consume! :a)))
+    (is (= "b" (tq/consume! :b)))
+    (is (= "b" (tq/force-retry! :b)))))
+
-- 
cgit v1.2.3


From 6c26a3b6871286510bb8e9770ee7f7e3abf97abe Mon Sep 17 00:00:00 2001
From: Ivar Refsdal <refsdal.ivar@gmail.com>
Date: Sun, 27 Mar 2022 18:39:44 +0200
Subject: Start use current millis in the database, not nano offset

---
 .gitignore                                         |  3 +-
 README.md                                          | 38 ++++++++++++++--------
 deps.edn                                           |  2 +-
 src/com/github/ivarref/yoltq.clj                   |  8 ++---
 src/com/github/ivarref/yoltq/error_poller.clj      | 10 +++---
 src/com/github/ivarref/yoltq/ext_sys.clj           | 13 ++++----
 src/com/github/ivarref/yoltq/impl.clj              | 14 ++++----
 src/com/github/ivarref/yoltq/poller.clj            | 19 +++++++----
 .../ivarref/yoltq/slow_executor_detector.clj       |  4 +--
 src/com/github/ivarref/yoltq/test_queue.clj        |  2 +-
 src/com/github/ivarref/yoltq/utils.clj             | 19 ++++++-----
 test/com/github/ivarref/yoltq/test_utils.clj       |  9 ++---
 12 files changed, 81 insertions(+), 60 deletions(-)

(limited to 'test/com/github/ivarref/yoltq/test_utils.clj')

diff --git a/.gitignore b/.gitignore
index cb9a7ca..c82fdd7 100644
--- a/.gitignore
+++ b/.gitignore
@@ -9,4 +9,5 @@ tree.txt
 .stage-url.txt
 *.pom.asc
 *.pom
-temp/
\ No newline at end of file
+temp/
+.clj-kondo/
diff --git a/README.md b/README.md
index 45ba8c4..9c5669c 100644
--- a/README.md
+++ b/README.md
@@ -333,12 +333,21 @@ easier.
 
 ## Change log
 
-### 2022-03-27 [v0.2.41](https://github.com/ivarref/yoltq/compare/v0.2.39...v0.2.41)
+### 20..-..-.. vHEAD [diff](https://github.com/ivarref/yoltq/compare/v0.2.41...HEAD)
+* Critical bugfix.
+```
+Started using (System/currentTimeMillis) and not (System/nanoTime)
+when storing time in the database. 
+```
+
+* Bump Clojure to `1.11.0`.
+
+### 2022-03-27 v0.2.41 [diff](https://github.com/ivarref/yoltq/compare/v0.2.39...v0.2.41)
 * Added function `healthy?` that returns:
 ```
-    true if no errors
-    false if one or more errors
-    nil if error-poller is yet to be executed.
+  true if no errors
+  false if one or more errors
+  nil if error-poller is yet to be executed.
 ```
 
 * Added default functions for `:on-system-error` and `:on-system-recovery`
@@ -348,22 +357,23 @@ easier.
 * Added function `queue-stats` that returns a nicely "formatted"
   vector of queue stats, for example:
 ```
-(queue-stats)
-=>
-[{:qname :add-message-thread, :status :done, :count 10274}
- {:qname :add-message-thread, :status :init, :count 30}
- {:qname :add-message-thread, :status :processing, :count 1}
- {:qname :send-message, :status :done, :count 21106}
- {:qname :send-message, :status :init, :count 56}]
+  (queue-stats)
+  =>
+  [{:qname :add-message-thread, :status :done, :count 10274}
+   {:qname :add-message-thread, :status :init, :count 30}
+   {:qname :add-message-thread, :status :processing, :count 1}
+   {:qname :send-message, :status :done, :count 21106}
+   {:qname :send-message, :status :init, :count 56}]
 ```
 
-### 2021-09-27 [v0.2.39](https://github.com/ivarref/yoltq/compare/v0.2.37...v0.2.39)
+### 2021-09-27 v0.2.39 [diff](https://github.com/ivarref/yoltq/compare/v0.2.37...v0.2.39)
 Added `:valid-payload?` option for queue consumers.
 
-### 2021-09-27 [v0.2.37](https://github.com/ivarref/yoltq/compare/v0.2.33...v0.2.37) 
+### 2021-09-27 v0.2.37 [diff](https://github.com/ivarref/yoltq/compare/v0.2.33...v0.2.37) 
 Improved error reporting.
 
-### 2021-09-24 v0.2.33: First publicly announced release.
+### 2021-09-24 v0.2.33
+First publicly announced release.
 
 ## License
 
diff --git a/deps.edn b/deps.edn
index d0f0a26..8e769e1 100644
--- a/deps.edn
+++ b/deps.edn
@@ -1,5 +1,5 @@
 {:deps      {org.clojure/tools.logging {:mvn/version "1.1.0"}
-             org.clojure/clojure       {:mvn/version "1.10.3"}}
+             org.clojure/clojure       {:mvn/version "1.11.0"}}
 
  :paths     ["src"]
 
diff --git a/src/com/github/ivarref/yoltq.clj b/src/com/github/ivarref/yoltq.clj
index 03a364f..17aa40a 100644
--- a/src/com/github/ivarref/yoltq.clj
+++ b/src/com/github/ivarref/yoltq.clj
@@ -19,7 +19,7 @@
 
 (def default-opts
   (-> {; Default number of times a queue job will be retried before giving up
-       ; Can be overridden on a per consumer basis with
+       ; Can be overridden on a per-consumer basis with
        ; (yq/add-consumer! :q (fn [payload] ...) {:max-retries 200})
        :max-retries                   100
 
@@ -34,7 +34,7 @@
        :hung-backoff-time             (Duration/ofMinutes 30)
 
        ; Most queue jobs in init state will be consumed by the tx-report-queue listener.
-       ; However in the case where a init job was added right before the application
+       ; However, in the case where an init job was added right before the application
        ; was shut down and did not have time to be processed by the tx-report-queue listener,
        ; it will be consumer by the init poller. This init poller backs off by
        ; :init-backoff-time to avoid unnecessary compare-and-swap lock failures that could
@@ -66,7 +66,7 @@
        ; How often should the system invoke
        :system-error-callback-backoff (Duration/ofHours 1)}
 
-      u/duration->nanos))
+      u/duration->millis))
 
 
 (defn init! [{:keys [conn] :as cfg}]
@@ -83,7 +83,7 @@
                                            default-opts
                                            (if *test-mode* old-conf (select-keys old-conf [:handlers]))
                                            cfg)
-                               u/duration->nanos)))]
+                               u/duration->millis)))]
       new-cfg)))
 
 
diff --git a/src/com/github/ivarref/yoltq/error_poller.clj b/src/com/github/ivarref/yoltq/error_poller.clj
index 1268482..ee6359e 100644
--- a/src/com/github/ivarref/yoltq/error_poller.clj
+++ b/src/com/github/ivarref/yoltq/error_poller.clj
@@ -22,7 +22,7 @@
                                   state       :recovery}}
                           {:keys [system-error-min-count system-error-callback-backoff]
                            :or   {system-error-min-count 3}}
-                          now-ns
+                          now-ms
                           error-count]
   (let [new-errors (->> (conj errors error-count)
                         (take-last system-error-min-count)
@@ -50,14 +50,14 @@
           (when (and (= old-state :recovery)
                      (= new-state :error))
             {:run-callback :error
-             :last-notify  now-ns})
+             :last-notify  now-ms})
 
           (when (and (= new-state :error)
                      (= old-state :error)
-                     (> now-ns
+                     (> now-ms
                         (+ last-notify system-error-callback-backoff)))
             {:run-callback :error
-             :last-notify  now-ns})
+             :last-notify  now-ms})
 
           (when (and (= new-state :recovery)
                      (= old-state :error))
@@ -88,7 +88,7 @@
         (log/debug "poll-errors found" error-count "errors in system")
         (reset! healthy? false))
       (reset! healthy? true))
-    (let [{:keys [run-callback] :as new-state} (swap! system-error handle-error-count config (ext/now-ns) error-count)]
+    (let [{:keys [run-callback] :as new-state} (swap! system-error handle-error-count config (ext/now-ms) error-count)]
       (when run-callback
         (cond (= run-callback :error)
               (on-system-error)
diff --git a/src/com/github/ivarref/yoltq/ext_sys.clj b/src/com/github/ivarref/yoltq/ext_sys.clj
index 3480475..692b934 100644
--- a/src/com/github/ivarref/yoltq/ext_sys.clj
+++ b/src/com/github/ivarref/yoltq/ext_sys.clj
@@ -1,17 +1,18 @@
 (ns com.github.ivarref.yoltq.ext-sys
   (:require [datomic.api :as d])
+  (:refer-clojure :exclude [random-uuid])
   (:import (java.util UUID)))
 
 
-(def ^:dynamic *now-ns-atom* nil)
+(def ^:dynamic *now-ms-atom* nil)
 (def ^:dynamic *squuid-atom* nil)
 (def ^:dynamic *random-atom* nil)
 
 
-(defn now-ns []
-  (if *now-ns-atom*
-    @*now-ns-atom*
-    (System/nanoTime)))
+(defn now-ms []
+  (if *now-ms-atom*
+    @*now-ms-atom*
+    (System/currentTimeMillis)))
 
 
 (defn squuid []
@@ -23,4 +24,4 @@
 (defn random-uuid []
   (if *random-atom*
     (UUID/fromString (str "00000000-0000-0000-0000-" (format "%012d" (swap! *random-atom* inc))))
-    (UUID/randomUUID)))
\ No newline at end of file
+    (UUID/randomUUID)))
diff --git a/src/com/github/ivarref/yoltq/impl.clj b/src/com/github/ivarref/yoltq/impl.clj
index 8b75fc3..b4eef8d 100644
--- a/src/com/github/ivarref/yoltq/impl.clj
+++ b/src/com/github/ivarref/yoltq/impl.clj
@@ -63,7 +63,7 @@
          :com.github.ivarref.yoltq/opts       (pr-str-safe :opts opts)
          :com.github.ivarref.yoltq/lock       (u/random-uuid)
          :com.github.ivarref.yoltq/tries      0
-         :com.github.ivarref.yoltq/init-time  (u/now-ns)}
+         :com.github.ivarref.yoltq/init-time  (u/now-ms)}
         (when-let [[q ext-id] (:depends-on opts)]
           (when-not (d/q '[:find ?e .
                            :in $ ?ext-id
@@ -138,8 +138,8 @@
               [:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/tries tries (inc tries)]
               [:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/status u/status-processing new-status]
               (if (= new-status u/status-done)
-                {:db/id [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/done-time (u/now-ns)}
-                {:db/id [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/error-time (u/now-ns)})]
+                {:db/id [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/done-time (u/now-ms)}
+                {:db/id [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/error-time (u/now-ms)})]
           start-time (System/nanoTime)
           {:keys [db-after]} @(d/transact conn tx)]
       (when tx-spent-time! (tx-spent-time! (- (System/nanoTime) start-time)))
@@ -171,7 +171,7 @@
           (log/debug "queue item" (str id) "for queue" queue-name "is now processing")
           (let [{:keys [retval exception]}
                 (try
-                  (swap! start-execute-time assoc (Thread/currentThread) [(ext/now-ns) id queue-name])
+                  (swap! start-execute-time assoc (Thread/currentThread) [(ext/now-ms) id queue-name])
                   (let [v (f payload)]
                     {:retval v})
                   (catch Throwable t
@@ -188,7 +188,7 @@
               (when-let [q-item (mark-status-fn! cfg queue-item u/status-done)]
                 (let [{:com.github.ivarref.yoltq/keys [init-time done-time tries]} q-item]
                   (log/info (fmt id queue-name u/status-done tries (- done-time init-time)))
-                  (when collect-spent-time! (collect-spent-time! (- (u/now-ns) init-time)))
+                  (when collect-spent-time! (collect-spent-time! (- (u/now-ms) init-time)))
                   (assoc q-item :retval retval :success? true :allow-cas-failure? true)))
 
               (some? exception)
@@ -198,14 +198,14 @@
                   (log/logp level exception (fmt id queue-name u/status-error tries (- error-time init-time)))
                   (log/logp level exception "error message was:" (str \" (ex-message exception) \") "for queue-item" (str id))
                   (log/logp level exception "ex-data was:" (ex-data exception) "for queue-item" (str id))
-                  (when collect-spent-time! (collect-spent-time! (- (u/now-ns) init-time)))
+                  (when collect-spent-time! (collect-spent-time! (- (u/now-ms) init-time)))
                   (assoc q-item :exception exception)))
 
               :else
               (when-let [q-item (mark-status-fn! cfg queue-item u/status-done)]
                 (let [{:com.github.ivarref.yoltq/keys [init-time done-time tries]} q-item]
                   (log/info (fmt id queue-name u/status-done tries (- done-time init-time)))
-                  (when collect-spent-time! (collect-spent-time! (- (u/now-ns) init-time)))
+                  (when collect-spent-time! (collect-spent-time! (- (u/now-ms) init-time)))
                   (assoc q-item :retval retval :success? true))))))
         (do
           (log/error "no handler for queue" queue-name)
diff --git a/src/com/github/ivarref/yoltq/poller.clj b/src/com/github/ivarref/yoltq/poller.clj
index 28b158f..9cf81c7 100644
--- a/src/com/github/ivarref/yoltq/poller.clj
+++ b/src/com/github/ivarref/yoltq/poller.clj
@@ -25,17 +25,16 @@
       (if-not (contains? old q)
         (try
           (log/debug "polling queue" queue-name "for status" status)
-          (let [start-time (u/now-ns)
+          (let [start-time (u/now-ms)
                 last-res (loop [prev-res nil]
                            (when @running?
                              (let [res (poll-once! cfg queue-name status)]
+                               (log/debug "poll-once! returned" res)
                                (if (and res (:success? res))
                                  (recur res)
                                  prev-res))))]
-            (let [spent-ns (- (u/now-ns) start-time)]
-              (log/trace "done polling queue" q "in"
-                         (format "%.1f" (double (/ spent-ns 1e6)))
-                         "ms"))
+            (let [spent-ms (- (u/now-ms) start-time)]
+              (log/trace "done polling queue" q "in" spent-ms "ms"))
             last-res)
           (finally
             (swap! running-queues disj q)))
@@ -44,6 +43,14 @@
       (log/error t "poll-queue! crashed:" (ex-message t)))
     (finally)))
 
+(comment
+  (def cfg @com.github.ivarref.yoltq/*config*))
+
+(comment
+  (poll-queue!
+    (atom true)
+    @com.github.ivarref.yoltq/*config*
+    [:add-message-thread :init]))
 
 (defn poll-all-queues! [running? config-atom pool]
   (try
@@ -54,4 +61,4 @@
                                   [q-name status])))]
           (.execute pool (fn [] (poll-queue! running? @config-atom q))))))
     (catch Throwable t
-      (log/error t "poll-all-queues! crashed:" (ex-message t)))))
\ No newline at end of file
+      (log/error t "poll-all-queues! crashed:" (ex-message t)))))
diff --git a/src/com/github/ivarref/yoltq/slow_executor_detector.clj b/src/com/github/ivarref/yoltq/slow_executor_detector.clj
index f15ef7d..80d3718 100644
--- a/src/com/github/ivarref/yoltq/slow_executor_detector.clj
+++ b/src/com/github/ivarref/yoltq/slow_executor_detector.clj
@@ -7,7 +7,7 @@
 (defn- do-show-slow-threads [{:keys [start-execute-time
                                      max-execute-time]}]
   (doseq [[^Thread thread [start-time queue-id queue-name]] @start-execute-time]
-    (when (> (ext/now-ns) (+ start-time max-execute-time))
+    (when (> (ext/now-ms) (+ start-time max-execute-time))
       (log/error "thread" (.getName thread) "spent too much time on"
                  "queue item" (str queue-id)
                  "for queue" queue-name
@@ -25,4 +25,4 @@
       (dotimes [_ 3]
         (when @running? (Thread/sleep 1000))))
     (catch Throwable t
-      (log/error t "reap! crashed:" (ex-message t)))))
\ No newline at end of file
+      (log/error t "reap! crashed:" (ex-message t)))))
diff --git a/src/com/github/ivarref/yoltq/test_queue.clj b/src/com/github/ivarref/yoltq/test_queue.clj
index 6183216..ee9cd54 100644
--- a/src/com/github/ivarref/yoltq/test_queue.clj
+++ b/src/com/github/ivarref/yoltq/test_queue.clj
@@ -47,7 +47,7 @@
      (with-bindings {#'yq/*config*       config#
                      #'yq/*running?*     (atom false)
                      #'yq/*test-mode*    true
-                     #'ext/*now-ns-atom* (atom 0)
+                     #'ext/*now-ms-atom* (atom 0)
                      #'ext/*random-atom* (atom 0)
                      #'ext/*squuid-atom* (atom 0)}
        (try
diff --git a/src/com/github/ivarref/yoltq/utils.clj b/src/com/github/ivarref/yoltq/utils.clj
index d551510..ad2444a 100644
--- a/src/com/github/ivarref/yoltq/utils.clj
+++ b/src/com/github/ivarref/yoltq/utils.clj
@@ -3,6 +3,7 @@
             [clojure.edn :as edn]
             [com.github.ivarref.yoltq.ext-sys :as ext]
             [clojure.tools.logging :as log])
+  (:refer-clojure :exclude [random-uuid])
   (:import (datomic Connection)
            (java.time Duration)))
 
@@ -13,10 +14,10 @@
 (def status-error :error)
 
 
-(defn duration->nanos [m]
+(defn duration->millis [m]
   (reduce-kv (fn [o k v]
                (if (instance? Duration v)
-                 (assoc o k (.toNanos v))
+                 (assoc o k (.toMillis v))
                  (assoc o k v)))
              {}
              m))
@@ -30,8 +31,8 @@
   (ext/random-uuid))
 
 
-(defn now-ns []
-  (ext/now-ns))
+(defn now-ms []
+  (ext/now-ms))
 
 
 (defn root-cause [e]
@@ -75,7 +76,7 @@
      :bindings   (get (get-queue-item db id) :com.github.ivarref.yoltq/bindings {})
      :tx         [[:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/lock old-lock new-lock]
                   [:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/status old-status status-processing]
-                  {:db/id [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/processing-time (now-ns)}]}))
+                  {:db/id [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/processing-time (now-ms)}]}))
 
 
 (defn get-init [{:keys [conn db init-backoff-time] :as cfg} queue-name]
@@ -94,11 +95,11 @@
                              [?e :com.github.ivarref.yoltq/lock ?lock]]
                            db
                            queue-name
-                           (- (now-ns) init-backoff-time))
+                           (- (now-ms) init-backoff-time))
                       (not-empty))]
       (let [[id old-lock] (rand-nth (into [] ids))]
         (prepare-processing db id queue-name old-lock :init))
-      (log/trace "no new-items in :init status for queue" queue-name))))
+      (log/debug "no new-items in :init status for queue" queue-name))))
 
 
 (defn get-error [{:keys [conn db error-backoff-time max-retries] :as cfg} queue-name]
@@ -120,7 +121,7 @@
                                [?e :com.github.ivarref.yoltq/lock ?lock]]
                              db
                              queue-name
-                             (- (now-ns) error-backoff-time)
+                             (- (now-ms) error-backoff-time)
                              (inc max-retries))
                         (not-empty))]
       (let [[id old-lock] (rand-nth (into [] ids))]
@@ -131,7 +132,7 @@
   (assert (instance? Connection conn) (str "Expected conn to be of type datomic.Connection. Was: "
                                            (str (if (nil? conn) "nil" conn))
                                            "\nConfig was: " (str cfg)))
-  (let [now (or now (now-ns))
+  (let [now (or now (now-ms))
         max-retries (get-in cfg [:handlers queue-name :max-retries] max-retries)
         db (or db (d/db conn))]
     (when-let [ids (->> (d/q '[:find ?id ?lock ?tries
diff --git a/test/com/github/ivarref/yoltq/test_utils.clj b/test/com/github/ivarref/yoltq/test_utils.clj
index 5427ff5..e4151c2 100644
--- a/test/com/github/ivarref/yoltq/test_utils.clj
+++ b/test/com/github/ivarref/yoltq/test_utils.clj
@@ -8,7 +8,8 @@
             [com.github.ivarref.yoltq.impl :as i]
             [clojure.edn :as edn]
             [com.github.ivarref.yoltq.ext-sys :as ext])
-  (:import (java.util UUID)))
+  (:import (java.util UUID)
+           (java.time Duration)))
 
 
 (logconfig/init-logging!
@@ -39,10 +40,10 @@
 
 
 (defn advance! [tp]
-  (assert (some? ext/*now-ns-atom*) "Expected to be running in test-mode!")
-  (swap! ext/*now-ns-atom* + (if (number? tp)
+  (assert (some? ext/*now-ms-atom*) "Expected to be running in test-mode!")
+  (swap! ext/*now-ms-atom* + (if (number? tp)
                                tp
-                               (.toNanos tp))))
+                               (.toMillis ^Duration tp))))
 
 
 (defn done-count []
-- 
cgit v1.2.3


From 41c9e08d63176cf7c239574d1d07f2b302a2d3ec Mon Sep 17 00:00:00 2001
From: Ivar Refsdal <refsdal.ivar@gmail.com>
Date: Sun, 27 Mar 2022 21:33:00 +0200
Subject: Fix use current millis in the database, not nano offset

---
 src/com/github/ivarref/yoltq.clj               | 11 ++-
 src/com/github/ivarref/yoltq/impl.clj          |  6 +-
 src/com/github/ivarref/yoltq/migrate.clj       | 58 ++++++++++++++++
 test/com/github/ivarref/yoltq/migrate_test.clj | 92 ++++++++++++++++++++++++++
 test/com/github/ivarref/yoltq/test_utils.clj   |  7 +-
 test/com/github/ivarref/yoltq/virtual_test.clj | 14 +++-
 6 files changed, 180 insertions(+), 8 deletions(-)
 create mode 100644 src/com/github/ivarref/yoltq/migrate.clj
 create mode 100644 test/com/github/ivarref/yoltq/migrate_test.clj

(limited to 'test/com/github/ivarref/yoltq/test_utils.clj')

diff --git a/src/com/github/ivarref/yoltq.clj b/src/com/github/ivarref/yoltq.clj
index 17aa40a..1a60a45 100644
--- a/src/com/github/ivarref/yoltq.clj
+++ b/src/com/github/ivarref/yoltq.clj
@@ -6,6 +6,7 @@
             [com.github.ivarref.yoltq.poller :as poller]
             [com.github.ivarref.yoltq.error-poller :as errpoller]
             [com.github.ivarref.yoltq.slow-executor-detector :as slow-executor]
+            [com.github.ivarref.yoltq.migrate :as migrate]
             [com.github.ivarref.yoltq.utils :as u])
   (:import (datomic Connection)
            (java.util.concurrent Executors TimeUnit ExecutorService)
@@ -64,7 +65,11 @@
        :system-error-poll-delay       (Duration/ofMinutes 1)
 
        ; How often should the system invoke
-       :system-error-callback-backoff (Duration/ofHours 1)}
+       :system-error-callback-backoff (Duration/ofHours 1)
+
+       ; Should old, possibly stalled jobs be automatically be migrated
+       ; as part of `start!`?
+       :auto-migrate? true}
 
       u/duration->millis))
 
@@ -104,7 +109,9 @@
 
 
 (defn- do-start! []
-  (let [{:keys [poll-delay pool-size system-error-poll-delay]} @*config*]
+  (let [{:keys [poll-delay pool-size system-error-poll-delay auto-migrate?] :as cfg} @*config*]
+    (when auto-migrate?
+      (migrate/migrate! cfg))
     (reset! threadpool (Executors/newScheduledThreadPool (+ 2 pool-size)))
     (let [pool @threadpool
           queue-listener-ready (promise)]
diff --git a/src/com/github/ivarref/yoltq/impl.clj b/src/com/github/ivarref/yoltq/impl.clj
index b4eef8d..6b14ffc 100644
--- a/src/com/github/ivarref/yoltq/impl.clj
+++ b/src/com/github/ivarref/yoltq/impl.clj
@@ -20,7 +20,8 @@
    #:db{:ident :com.github.ivarref.yoltq/init-time, :cardinality :db.cardinality/one, :valueType :db.type/long}
    #:db{:ident :com.github.ivarref.yoltq/processing-time, :cardinality :db.cardinality/one, :valueType :db.type/long}
    #:db{:ident :com.github.ivarref.yoltq/done-time, :cardinality :db.cardinality/one, :valueType :db.type/long}
-   #:db{:ident :com.github.ivarref.yoltq/error-time, :cardinality :db.cardinality/one, :valueType :db.type/long}])
+   #:db{:ident :com.github.ivarref.yoltq/error-time, :cardinality :db.cardinality/one, :valueType :db.type/long}
+   #:db{:ident :com.github.ivarref.yoltq/version, :cardinality :db.cardinality/one, :valueType :db.type/string, :index true}])
 
 
 (defn pr-str-safe [what x]
@@ -63,7 +64,8 @@
          :com.github.ivarref.yoltq/opts       (pr-str-safe :opts opts)
          :com.github.ivarref.yoltq/lock       (u/random-uuid)
          :com.github.ivarref.yoltq/tries      0
-         :com.github.ivarref.yoltq/init-time  (u/now-ms)}
+         :com.github.ivarref.yoltq/init-time  (u/now-ms)
+         :com.github.ivarref.yoltq/version    "2"}
         (when-let [[q ext-id] (:depends-on opts)]
           (when-not (d/q '[:find ?e .
                            :in $ ?ext-id
diff --git a/src/com/github/ivarref/yoltq/migrate.clj b/src/com/github/ivarref/yoltq/migrate.clj
new file mode 100644
index 0000000..89fc286
--- /dev/null
+++ b/src/com/github/ivarref/yoltq/migrate.clj
@@ -0,0 +1,58 @@
+(ns com.github.ivarref.yoltq.migrate
+  (:require [datomic.api :as d]
+            [clojure.tools.logging :as log]))
+
+(defn to->v2-ent [{:keys [conn]} now-ms id]
+  (log/info "Migrating id" id)
+  (let [attr-val (fn [attr]
+                   (when-let [old (d/q '[:find ?time .
+                                         :in $ ?e ?a
+                                         :where
+                                         [?e ?a ?time]]
+                                       (d/db conn)
+                                       [:com.github.ivarref.yoltq/id id]
+                                       attr)]
+                     (let [now-ms (or now-ms
+                                      (.getTime (d/q '[:find (max ?txinst) .
+                                                       :in $ ?e ?a
+                                                       :where
+                                                       [?e ?a _ ?tx true]
+                                                       [?tx :db/txInstant ?txinst]]
+                                                     (d/history (d/db conn))
+                                                     [:com.github.ivarref.yoltq/id id]
+                                                     attr)))]
+                       (log/info "Updating" id attr "to" now-ms)
+                       [[:db/cas [:com.github.ivarref.yoltq/id id]
+                         attr old now-ms]])))]
+    (vec (concat [[:db/cas [:com.github.ivarref.yoltq/id id]
+                   :com.github.ivarref.yoltq/version nil "2"]]
+                 (mapcat attr-val [:com.github.ivarref.yoltq/init-time
+                                   :com.github.ivarref.yoltq/processing-time
+                                   :com.github.ivarref.yoltq/done-time
+                                   :com.github.ivarref.yoltq/error-time])))))
+
+(defn to->v2 [{:keys [conn loop? now-ms]
+               :or   {loop? true}
+               :as   cfg}]
+  (loop [tx-vec []]
+    (if-let [id (some->> (d/q '[:find [?id ...]
+                                :in $
+                                :where
+                                [?e :com.github.ivarref.yoltq/id ?id]
+                                [(missing? $ ?e :com.github.ivarref.yoltq/version)]]
+                              (d/db conn))
+                         (sort)
+                         (not-empty)
+                         (first))]
+      (let [tx (to->v2-ent cfg now-ms id)]
+        @(d/transact conn tx)
+        (if loop?
+          (recur (conj tx-vec tx))
+          tx))
+      (do
+        (log/info "No items left to migrate")
+        tx-vec))))
+
+
+(defn migrate! [cfg]
+  (to->v2 cfg))
diff --git a/test/com/github/ivarref/yoltq/migrate_test.clj b/test/com/github/ivarref/yoltq/migrate_test.clj
new file mode 100644
index 0000000..0063631
--- /dev/null
+++ b/test/com/github/ivarref/yoltq/migrate_test.clj
@@ -0,0 +1,92 @@
+(ns com.github.ivarref.yoltq.migrate-test
+  (:require [clojure.test :refer [deftest is]]
+            [com.github.ivarref.yoltq.ext-sys :as ext]
+            [com.github.ivarref.yoltq.migrate :as m]
+            [com.github.ivarref.yoltq.impl :as impl]
+            [com.github.ivarref.yoltq.test-utils :as tu]
+            [com.github.ivarref.yoltq.utils :as u]
+            [datomic.api :as d]))
+
+
+(deftest to-v2-migration
+  (with-bindings {#'ext/*squuid-atom* (atom 0)}
+    (let [conn (tu/empty-conn)]
+      @(d/transact conn impl/schema)
+      @(d/transact conn [{:com.github.ivarref.yoltq/id              (u/squuid)
+                          :com.github.ivarref.yoltq/queue-name      :dummy
+                          :com.github.ivarref.yoltq/status          u/status-processing
+                          :com.github.ivarref.yoltq/init-time       1
+                          :com.github.ivarref.yoltq/processing-time 2}])
+      @(d/transact conn [{:com.github.ivarref.yoltq/id         (u/squuid)
+                          :com.github.ivarref.yoltq/queue-name :dummy
+                          :com.github.ivarref.yoltq/status     u/status-init
+                          :com.github.ivarref.yoltq/init-time  3}])
+      (is (= [[[:db/cas
+                [:com.github.ivarref.yoltq/id
+                 #uuid "00000000-0000-0000-0000-000000000001"]
+                :com.github.ivarref.yoltq/version
+                nil
+                "2"]
+               [:db/cas
+                [:com.github.ivarref.yoltq/id
+                 #uuid "00000000-0000-0000-0000-000000000001"]
+                :com.github.ivarref.yoltq/init-time
+                1
+                1000]
+               [:db/cas
+                [:com.github.ivarref.yoltq/id
+                 #uuid "00000000-0000-0000-0000-000000000001"]
+                :com.github.ivarref.yoltq/processing-time
+                2
+                1000]]
+              [[:db/cas
+                [:com.github.ivarref.yoltq/id
+                 #uuid "00000000-0000-0000-0000-000000000002"]
+                :com.github.ivarref.yoltq/version
+                nil
+                "2"]
+               [:db/cas
+                [:com.github.ivarref.yoltq/id
+                 #uuid "00000000-0000-0000-0000-000000000002"]
+                :com.github.ivarref.yoltq/init-time
+                3
+                1000]]]
+             (m/migrate! {:conn   conn
+                          :now-ms 1000
+                          :loop?  true})))
+      (is (= []
+             (m/migrate! {:conn   conn
+                          :now-ms 1000
+                          :loop?  true}))))))
+
+
+(deftest to-v2-migration-real-time
+  (with-bindings {#'ext/*squuid-atom* (atom 0)}
+    (let [conn (tu/empty-conn)
+          id (u/squuid)]
+      @(d/transact conn impl/schema)
+      @(d/transact conn [{:com.github.ivarref.yoltq/id         id
+                          :com.github.ivarref.yoltq/queue-name :dummy
+                          :com.github.ivarref.yoltq/status     u/status-init
+                          :com.github.ivarref.yoltq/init-time  1}])
+      (Thread/sleep 100)
+      @(d/transact conn [{:com.github.ivarref.yoltq/id        id
+                          :com.github.ivarref.yoltq/init-time 2}])
+      (let [tx-times (->> (d/q '[:find [?txinst ...]
+                                 :in $ ?e
+                                 :where
+                                 [?e :com.github.ivarref.yoltq/init-time _ ?tx true]
+                                 [?tx :db/txInstant ?txinst]]
+                               (d/history (d/db conn))
+                               [:com.github.ivarref.yoltq/id id])
+                          (sort)
+                          (vec))]
+        (is (= 2 (count tx-times)))
+        (m/migrate! {:conn conn})
+        (is (= (.getTime (last tx-times))
+               (d/q '[:find ?init-time .
+                      :in $ ?e
+                      :where
+                      [?e :com.github.ivarref.yoltq/init-time ?init-time]]
+                    (d/db conn)
+                    [:com.github.ivarref.yoltq/id id])))))))
diff --git a/test/com/github/ivarref/yoltq/test_utils.clj b/test/com/github/ivarref/yoltq/test_utils.clj
index e4151c2..0c1b2f0 100644
--- a/test/com/github/ivarref/yoltq/test_utils.clj
+++ b/test/com/github/ivarref/yoltq/test_utils.clj
@@ -7,7 +7,8 @@
             [clojure.string :as str]
             [com.github.ivarref.yoltq.impl :as i]
             [clojure.edn :as edn]
-            [com.github.ivarref.yoltq.ext-sys :as ext])
+            [com.github.ivarref.yoltq.ext-sys :as ext]
+            [clojure.pprint :as pp])
   (:import (java.util UUID)
            (java.time Duration)))
 
@@ -54,6 +55,10 @@
        (d/db (:conn @yq/*config*))))
 
 
+(defn pp [x]
+  (pp/pprint x)
+  x)
+
 (defn get-init [& args]
   (apply u/get-init @yq/*config* args))
 
diff --git a/test/com/github/ivarref/yoltq/virtual_test.clj b/test/com/github/ivarref/yoltq/virtual_test.clj
index acd3eb7..34c9026 100644
--- a/test/com/github/ivarref/yoltq/virtual_test.clj
+++ b/test/com/github/ivarref/yoltq/virtual_test.clj
@@ -1,6 +1,6 @@
 (ns com.github.ivarref.yoltq.virtual-test
   (:require [datomic-schema.core]
-            [clojure.test :refer :all]
+            [clojure.test :refer [use-fixtures deftest is] :refer-macros [thrown?]]
             [com.github.ivarref.yoltq.test-queue :as tq]
             [com.github.ivarref.yoltq.test-utils :as u]
             [datomic.api :as d]
@@ -8,7 +8,8 @@
             [clojure.tools.logging :as log]
             [com.github.ivarref.yoltq.impl :as i]
             [com.github.ivarref.yoltq :as yq]
-            [taoensso.timbre :as timbre]))
+            [taoensso.timbre :as timbre]
+            [com.github.ivarref.yoltq.migrate :as migrate]))
 
 
 (use-fixtures :each tq/call-with-virtual-queue!)
@@ -21,6 +22,13 @@
     @(d/transact conn [(yq/put :q {:work 123})])
     (is (= {:work 123} (tq/consume! :q)))))
 
+(deftest happy-case-no-migration-for-new-entities
+  (let [conn (u/empty-conn)]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q identity)
+    @(d/transact conn [(yq/put :q {:work 123})])
+    (is (= {:work 123} (tq/consume! :q)))
+    (is (= [] (migrate/migrate! @yq/*config*)))))
 
 (deftest happy-case-tx-report-q
   (let [conn (u/empty-conn)]
@@ -341,4 +349,4 @@
                                          (some? id))})
     @(d/transact conn [(yq/put :q {:id "a"})])
     (timbre/with-level :fatal
-                       (is (thrown? Exception @(d/transact conn [(yq/put :q {})]))))))
\ No newline at end of file
+                       (is (thrown? Exception @(d/transact conn [(yq/put :q {})]))))))
-- 
cgit v1.2.3