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
---
 src/com/github/ivarref/yoltq/utils.clj | 154 +++++++++++++++++++++++++++++++++
 1 file changed, 154 insertions(+)
 create mode 100644 src/com/github/ivarref/yoltq/utils.clj

(limited to 'src/com/github/ivarref/yoltq/utils.clj')

diff --git a/src/com/github/ivarref/yoltq/utils.clj b/src/com/github/ivarref/yoltq/utils.clj
new file mode 100644
index 0000000..c96d1dc
--- /dev/null
+++ b/src/com/github/ivarref/yoltq/utils.clj
@@ -0,0 +1,154 @@
+(ns com.github.ivarref.yoltq.utils
+  (:require [datomic.api :as d]
+            [clojure.edn :as edn]
+            [com.github.ivarref.yoltq.ext-sys :as ext]
+            [clojure.tools.logging :as log])
+  (:import (datomic Connection)
+           (java.time Duration)))
+
+
+(def status-init :init)
+(def status-processing :processing)
+(def status-done :done)
+(def status-error :error)
+
+
+(defn duration->nanos [m]
+  (reduce-kv (fn [o k v]
+               (if (instance? Duration v)
+                 (assoc o k (.toNanos v))
+                 (assoc o k v)))
+             {}
+             m))
+
+
+(defn squuid []
+  (ext/squuid))
+
+
+(defn random-uuid []
+  (ext/random-uuid))
+
+
+(defn now-ns []
+  (ext/now-ns))
+
+
+(defn root-cause [e]
+  (if-let [root (ex-cause e)]
+    (root-cause root)
+    e))
+
+
+(defn db-error-map [^Throwable t]
+  (loop [e t]
+    (cond (nil? e) nil
+
+          (and (map? (ex-data e))
+               (contains? (ex-data e) :db/error))
+          (ex-data e)
+
+          :else
+          (recur (ex-cause e)))))
+
+
+(defn get-queue-item [db id]
+  (-> (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/bindings edn/read-string)))
+
+
+(defn prepare-processing [id queue-name old-lock old-status]
+  (let [new-lock (random-uuid)]
+    {:id         id
+     :lock       new-lock
+     :queue-name queue-name
+     :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)}]}))
+
+
+(defn get-init [{:keys [conn db init-backoff-time] :as cfg} queue-name]
+  (assert (instance? Connection conn) (str "Expected conn to be of type datomic.Connection. Was: "
+                                           (str (if (nil? conn) "nil" conn))
+                                           "\nConfig was: " (str cfg)))
+  (if-let [ids (->> (d/q '[:find ?id ?lock
+                           :in $ ?queue-name ?backoff
+                           :where
+                           [?e :com.github.ivarref.yoltq/status :init]
+                           [?e :com.github.ivarref.yoltq/queue-name ?queue-name]
+                           [?e :com.github.ivarref.yoltq/init-time ?init-time]
+                           [(>= ?backoff ?init-time)]
+                           [?e :com.github.ivarref.yoltq/id ?id]
+                           [?e :com.github.ivarref.yoltq/lock ?lock]]
+                         (or db (d/db conn))
+                         queue-name
+                         (- (now-ns) init-backoff-time))
+                    (not-empty))]
+    (let [[id old-lock] (rand-nth (into [] ids))]
+      (prepare-processing id queue-name old-lock :init))
+    (log/trace "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]
+  (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 [max-retries (get-in cfg [:handlers queue-name :max-retries] max-retries)]
+    (when-let [ids (->> (d/q '[:find ?id ?lock
+                               :in $ ?queue-name ?backoff ?max-tries
+                               :where
+                               [?e :com.github.ivarref.yoltq/status :error]
+                               [?e :com.github.ivarref.yoltq/queue-name ?queue-name]
+                               [?e :com.github.ivarref.yoltq/error-time ?time]
+                               [(>= ?backoff ?time)]
+                               [?e :com.github.ivarref.yoltq/tries ?tries]
+                               [(> ?max-tries ?tries)]
+                               [?e :com.github.ivarref.yoltq/id ?id]
+                               [?e :com.github.ivarref.yoltq/lock ?lock]]
+                             (or db (d/db conn))
+                             queue-name
+                             (- (now-ns) error-backoff-time)
+                             (inc max-retries))
+                        (not-empty))]
+      (let [[id old-lock] (rand-nth (into [] ids))]
+        (prepare-processing id queue-name old-lock :error)))))
+
+
+(defn get-hung [{:keys [conn db now hung-backoff-time max-retries] :as cfg} queue-name]
+  (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))
+        max-retries (get-in cfg [:handlers queue-name :max-retries] max-retries)]
+    (when-let [ids (->> (d/q '[:find ?id ?lock ?tries
+                               :in $ ?qname ?backoff
+                               :where
+                               [?e :com.github.ivarref.yoltq/status :processing]
+                               [?e :com.github.ivarref.yoltq/queue-name ?qname]
+                               [?e :com.github.ivarref.yoltq/processing-time ?time]
+                               [(>= ?backoff ?time)]
+                               [?e :com.github.ivarref.yoltq/tries ?tries]
+                               [?e :com.github.ivarref.yoltq/id ?id]
+                               [?e :com.github.ivarref.yoltq/lock ?lock]]
+                             (or db (d/db conn))
+                             queue-name
+                             (- now hung-backoff-time))
+                        (not-empty))]
+      (let [new-lock (random-uuid)
+            [id old-lock tries _t] (rand-nth (into [] ids))
+            to-error? (>= tries max-retries)]
+        {:id         id
+         :lock       new-lock
+         :queue-name queue-name
+         :was-hung?  true
+         :to-error?  to-error?
+         :tx         (if (not to-error?)
+                       [[: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/tries tries (inc tries)]
+                        {:db/id [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/error-time now}]
+                       [[: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/tries tries (inc tries)]
+                        [:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/status status-processing status-error]
+                        {:db/id [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/error-time now}])}))))
-- 
cgit v1.2.3


From ad8a41bd7d9e6fed77f633a75ef36410b7afbef1 Mon Sep 17 00:00:00 2001
From: Ivar Refsdal <ivar.refsdal@nsd.no>
Date: Fri, 17 Sep 2021 14:25:08 +0200
Subject: Start add bindings ...

---
 src/com/github/ivarref/yoltq.clj               | 10 +++--
 src/com/github/ivarref/yoltq/poller.clj        | 12 ++++--
 src/com/github/ivarref/yoltq/report_queue.clj  | 11 ++---
 src/com/github/ivarref/yoltq/utils.clj         | 58 ++++++++++++++++----------
 src/com/github/ivarref/yoltq/virtual_queue.clj | 23 +++++-----
 test/com/github/ivarref/yoltq/virtual_test.clj | 32 +++++++++++++-
 6 files changed, 98 insertions(+), 48 deletions(-)

(limited to 'src/com/github/ivarref/yoltq/utils.clj')

diff --git a/src/com/github/ivarref/yoltq.clj b/src/com/github/ivarref/yoltq.clj
index d3eefef..6341e41 100644
--- a/src/com/github/ivarref/yoltq.clj
+++ b/src/com/github/ivarref/yoltq.clj
@@ -58,6 +58,8 @@
        ; contain the stacktrace of the stuck threads.
        :pool-size                     4
 
+       :capture-bindings                      []
+
        ; How often should the system be polled for failed queue jobs
        :system-error-poll-delay       (Duration/ofMinutes 1)
 
@@ -159,17 +161,17 @@
       (d/create-database uri)
       (let [ok-items (atom [])
             conn (d/connect uri)
-            n 100]
+            n 1]
         (init! {:conn               conn
                 :error-backoff-time (Duration/ofSeconds 1)
                 :poll-delay         (Duration/ofSeconds 1)})
         (add-consumer! :q (fn [payload]
-                            (when (> (Math/random) 0.5)
-                              (throw (ex-info "oops" {})))
+                            #_(when (> (Math/random) 0.5)
+                                (throw (ex-info "oops" {})))
                             (if (= n (count (swap! received conj (:work payload))))
                               (log/info "... and we are done!")
                               (log/info "got payload" payload "total ok:" (count @received)))))
         (start!)
         (dotimes [x n]
-          @(d/transact conn [(put :q {:work x})]))
+          @(d/transact conn [(put :q {:work 123})]))
         nil))))
\ No newline at end of file
diff --git a/src/com/github/ivarref/yoltq/poller.clj b/src/com/github/ivarref/yoltq/poller.clj
index ad9d32a..1f4e65d 100644
--- a/src/com/github/ivarref/yoltq/poller.clj
+++ b/src/com/github/ivarref/yoltq/poller.clj
@@ -5,10 +5,14 @@
 
 
 (defn poll-once! [cfg q status]
-  (case status
-    :init (some->> (u/get-init cfg q) (i/take! cfg) (i/execute! cfg))
-    :error (some->> (u/get-error cfg q) (i/take! cfg) (i/execute! cfg))
-    :hung (some->> (u/get-hung cfg q) (i/take! cfg) (i/execute! cfg))))
+  (when-let [item (case status
+                    :init (u/get-init cfg q)
+                    :error (u/get-error cfg q)
+                    :hung (u/get-hung cfg q))]
+    (with-bindings (get item :bindings {})
+      (some->> item
+               (i/take! cfg)
+               (i/execute! cfg)))))
 
 
 (defn poll-queue! [running?
diff --git a/src/com/github/ivarref/yoltq/report_queue.clj b/src/com/github/ivarref/yoltq/report_queue.clj
index a40d29a..c6559bf 100644
--- a/src/com/github/ivarref/yoltq/report_queue.clj
+++ b/src/com/github/ivarref/yoltq/report_queue.clj
@@ -19,11 +19,12 @@
       (doseq [id new-ids]
         (consumer (fn []
                     (try
-                      (let [{:com.github.ivarref.yoltq/keys [lock id status queue-name]} (u/get-queue-item db-after id)]
-                        (some->>
-                          (u/prepare-processing id queue-name lock status)
-                          (i/take! cfg)
-                          (i/execute! cfg)))
+                      (let [{:com.github.ivarref.yoltq/keys [lock id status queue-name bindings]} (u/get-queue-item db-after id)]
+                        (with-bindings (or bindings {})
+                          (some->>
+                            (u/prepare-processing db-after id queue-name lock status)
+                            (i/take! cfg)
+                            (i/execute! cfg))))
                       (catch Throwable t
                         (log/error t "unexpected error in process-poll-result!")))))))))
 
diff --git a/src/com/github/ivarref/yoltq/utils.clj b/src/com/github/ivarref/yoltq/utils.clj
index c96d1dc..9501343 100644
--- a/src/com/github/ivarref/yoltq/utils.clj
+++ b/src/com/github/ivarref/yoltq/utils.clj
@@ -56,14 +56,22 @@
   (-> (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/bindings edn/read-string)))
+      (update :com.github.ivarref.yoltq/bindings
+              (fn [s]
+                (when s
+                  (->> s
+                       (edn/read-string)
+                       (reduce-kv (fn [o k v]
+                                    (assoc o (resolve k) v))
+                                  {})))))))
 
 
-(defn prepare-processing [id queue-name old-lock old-status]
+(defn prepare-processing [db id queue-name old-lock old-status]
   (let [new-lock (random-uuid)]
     {:id         id
      :lock       new-lock
      :queue-name queue-name
+     :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)}]}))
@@ -73,29 +81,31 @@
   (assert (instance? Connection conn) (str "Expected conn to be of type datomic.Connection. Was: "
                                            (str (if (nil? conn) "nil" conn))
                                            "\nConfig was: " (str cfg)))
-  (if-let [ids (->> (d/q '[:find ?id ?lock
-                           :in $ ?queue-name ?backoff
-                           :where
-                           [?e :com.github.ivarref.yoltq/status :init]
-                           [?e :com.github.ivarref.yoltq/queue-name ?queue-name]
-                           [?e :com.github.ivarref.yoltq/init-time ?init-time]
-                           [(>= ?backoff ?init-time)]
-                           [?e :com.github.ivarref.yoltq/id ?id]
-                           [?e :com.github.ivarref.yoltq/lock ?lock]]
-                         (or db (d/db conn))
-                         queue-name
-                         (- (now-ns) init-backoff-time))
-                    (not-empty))]
-    (let [[id old-lock] (rand-nth (into [] ids))]
-      (prepare-processing id queue-name old-lock :init))
-    (log/trace "no new-items in :init status for queue" queue-name)))
+  (let [db (or db (d/db conn))]
+    (if-let [ids (->> (d/q '[:find ?id ?lock
+                             :in $ ?queue-name ?backoff
+                             :where
+                             [?e :com.github.ivarref.yoltq/status :init]
+                             [?e :com.github.ivarref.yoltq/queue-name ?queue-name]
+                             [?e :com.github.ivarref.yoltq/init-time ?init-time]
+                             [(>= ?backoff ?init-time)]
+                             [?e :com.github.ivarref.yoltq/id ?id]
+                             [?e :com.github.ivarref.yoltq/lock ?lock]]
+                           db
+                           queue-name
+                           (- (now-ns) 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))))
 
 
 (defn get-error [{:keys [conn db error-backoff-time max-retries] :as cfg} queue-name]
   (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 [max-retries (get-in cfg [:handlers queue-name :max-retries] max-retries)]
+  (let [db (or db (d/db conn))
+        max-retries (get-in cfg [:handlers queue-name :max-retries] max-retries)]
     (when-let [ids (->> (d/q '[:find ?id ?lock
                                :in $ ?queue-name ?backoff ?max-tries
                                :where
@@ -107,13 +117,13 @@
                                [(> ?max-tries ?tries)]
                                [?e :com.github.ivarref.yoltq/id ?id]
                                [?e :com.github.ivarref.yoltq/lock ?lock]]
-                             (or db (d/db conn))
+                             db
                              queue-name
                              (- (now-ns) error-backoff-time)
                              (inc max-retries))
                         (not-empty))]
       (let [[id old-lock] (rand-nth (into [] ids))]
-        (prepare-processing id queue-name old-lock :error)))))
+        (prepare-processing db id queue-name old-lock :error)))))
 
 
 (defn get-hung [{:keys [conn db now hung-backoff-time max-retries] :as cfg} queue-name]
@@ -121,7 +131,8 @@
                                            (str (if (nil? conn) "nil" conn))
                                            "\nConfig was: " (str cfg)))
   (let [now (or now (now-ns))
-        max-retries (get-in cfg [:handlers queue-name :max-retries] max-retries)]
+        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
                                :in $ ?qname ?backoff
                                :where
@@ -132,7 +143,7 @@
                                [?e :com.github.ivarref.yoltq/tries ?tries]
                                [?e :com.github.ivarref.yoltq/id ?id]
                                [?e :com.github.ivarref.yoltq/lock ?lock]]
-                             (or db (d/db conn))
+                             db
                              queue-name
                              (- now hung-backoff-time))
                         (not-empty))]
@@ -144,6 +155,7 @@
          :queue-name queue-name
          :was-hung?  true
          :to-error?  to-error?
+         :bindings   (get (get-queue-item db id) :com.github.ivarref.yoltq/bindings {})
          :tx         (if (not to-error?)
                        [[: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/tries tries (inc tries)]
diff --git a/src/com/github/ivarref/yoltq/virtual_queue.clj b/src/com/github/ivarref/yoltq/virtual_queue.clj
index f133bde..71c7b6d 100644
--- a/src/com/github/ivarref/yoltq/virtual_queue.clj
+++ b/src/com/github/ivarref/yoltq/virtual_queue.clj
@@ -146,17 +146,18 @@
 (defmacro consume-expect! [queue-name expected-status]
   `(if-let [job# (get-tx-q-job ~queue-name)]
      (try
-       (let [res# (some->>
-                    (u/prepare-processing (: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#)))
+       (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 nil (str "No job found for queue " ~queue-name))))
\ No newline at end of file
diff --git a/test/com/github/ivarref/yoltq/virtual_test.clj b/test/com/github/ivarref/yoltq/virtual_test.clj
index 41d2461..575dc1b 100644
--- a/test/com/github/ivarref/yoltq/virtual_test.clj
+++ b/test/com/github/ivarref/yoltq/virtual_test.clj
@@ -8,7 +8,9 @@
             [com.github.ivarref.yoltq.utils :as uu]
             [clojure.tools.logging :as log]
             [com.github.ivarref.yoltq.impl :as i]
-            [com.github.ivarref.yoltq :as yq]))
+            [com.github.ivarref.yoltq :as yq]
+            [clojure.pprint :as pprint]
+            [clojure.edn :as edn]))
 
 
 (use-fixtures :each vq/call-with-virtual-queue!)
@@ -230,3 +232,31 @@
     (is (nil? (uu/get-error (assoc-in @dq/*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]
+                           (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)])
+
+    (is (= 1 (vq/consume-expect! :q :done)))
+    (vq/consume-expect! :q :error)))
+
+
+(def ^:dynamic *some-binding* nil)
+
+
+(deftest binding-test
+  (let [conn (u/empty-conn)]
+    (dq/init! {:conn conn
+               :bindings [#'*some-binding*]})
+    (dq/add-consumer! :q (fn [_] *some-binding*))
+    (binding [*some-binding* 1] @(d/transact conn [(dq/put :q nil)]))
+    #_(binding [*some-binding* 2] @(d/transact conn [(dq/put :q nil)]))
+    #_@(d/transact conn [(dq/put :q nil)])
+    (is (= 1 (vq/consume-expect! :q :done)))))
-- 
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 'src/com/github/ivarref/yoltq/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 'src/com/github/ivarref/yoltq/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 ce96806e63ead7c926a348842a3b466eba01190c Mon Sep 17 00:00:00 2001
From: Ivar Refsdal <refsdal.ivar@gmail.com>
Date: Sun, 27 Mar 2022 22:41:49 +0200
Subject: Auto migration should run in the background and not interfere with
 polling

---
 README.md                              |  4 ++++
 src/com/github/ivarref/yoltq.clj       |  2 +-
 src/com/github/ivarref/yoltq/utils.clj | 25 ++++++++++++++++---------
 3 files changed, 21 insertions(+), 10 deletions(-)

(limited to 'src/com/github/ivarref/yoltq/utils.clj')

diff --git a/README.md b/README.md
index 334af5d..2385d15 100644
--- a/README.md
+++ b/README.md
@@ -333,6 +333,10 @@ easier.
 
 ## Change log
 
+### 2022-03-27 v0.2.48 [diff](https://github.com/ivarref/yoltq/compare/v0.2.46...v0.2.48)
+* Auto migration is done in the background.
+* Only poll for current version of jobs, thus no races for auto migration.
+
 ### 2022-03-27 v0.2.46 [diff](https://github.com/ivarref/yoltq/compare/v0.2.41...v0.2.46)
 * Critical bugfix that in some cases can lead to stalled jobs.
 ```
diff --git a/src/com/github/ivarref/yoltq.clj b/src/com/github/ivarref/yoltq.clj
index 1a60a45..4b324c4 100644
--- a/src/com/github/ivarref/yoltq.clj
+++ b/src/com/github/ivarref/yoltq.clj
@@ -111,7 +111,7 @@
 (defn- do-start! []
   (let [{:keys [poll-delay pool-size system-error-poll-delay auto-migrate?] :as cfg} @*config*]
     (when auto-migrate?
-      (migrate/migrate! cfg))
+      (future (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/utils.clj b/src/com/github/ivarref/yoltq/utils.clj
index ad2444a..39572a9 100644
--- a/src/com/github/ivarref/yoltq/utils.clj
+++ b/src/com/github/ivarref/yoltq/utils.clj
@@ -13,6 +13,7 @@
 (def status-done :done)
 (def status-error :error)
 
+(def current-version "2")
 
 (defn duration->millis [m]
   (reduce-kv (fn [o k v]
@@ -85,17 +86,19 @@
                                            "\nConfig was: " (str cfg)))
   (let [db (or db (d/db conn))]
     (if-let [ids (->> (d/q '[:find ?id ?lock
-                             :in $ ?queue-name ?backoff
+                             :in $ ?queue-name ?backoff ?current-version
                              :where
                              [?e :com.github.ivarref.yoltq/status :init]
                              [?e :com.github.ivarref.yoltq/queue-name ?queue-name]
                              [?e :com.github.ivarref.yoltq/init-time ?init-time]
                              [(>= ?backoff ?init-time)]
                              [?e :com.github.ivarref.yoltq/id ?id]
-                             [?e :com.github.ivarref.yoltq/lock ?lock]]
+                             [?e :com.github.ivarref.yoltq/lock ?lock]
+                             [?e :com.github.ivarref.yoltq/version ?current-version]]
                            db
                            queue-name
-                           (- (now-ms) init-backoff-time))
+                           (- (now-ms) init-backoff-time)
+                           current-version)
                       (not-empty))]
       (let [[id old-lock] (rand-nth (into [] ids))]
         (prepare-processing db id queue-name old-lock :init))
@@ -109,7 +112,7 @@
   (let [db (or db (d/db conn))
         max-retries (get-in cfg [:handlers queue-name :max-retries] max-retries)]
     (when-let [ids (->> (d/q '[:find ?id ?lock
-                               :in $ ?queue-name ?backoff ?max-tries
+                               :in $ ?queue-name ?backoff ?max-tries ?current-version
                                :where
                                [?e :com.github.ivarref.yoltq/status :error]
                                [?e :com.github.ivarref.yoltq/queue-name ?queue-name]
@@ -118,11 +121,13 @@
                                [?e :com.github.ivarref.yoltq/tries ?tries]
                                [(> ?max-tries ?tries)]
                                [?e :com.github.ivarref.yoltq/id ?id]
-                               [?e :com.github.ivarref.yoltq/lock ?lock]]
+                               [?e :com.github.ivarref.yoltq/lock ?lock]
+                               [?e :com.github.ivarref.yoltq/version ?current-version]]
                              db
                              queue-name
                              (- (now-ms) error-backoff-time)
-                             (inc max-retries))
+                             (inc max-retries)
+                             current-version)
                         (not-empty))]
       (let [[id old-lock] (rand-nth (into [] ids))]
         (prepare-processing db id queue-name old-lock :error)))))
@@ -136,7 +141,7 @@
         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
-                               :in $ ?qname ?backoff
+                               :in $ ?qname ?backoff ?current-version
                                :where
                                [?e :com.github.ivarref.yoltq/status :processing]
                                [?e :com.github.ivarref.yoltq/queue-name ?qname]
@@ -144,10 +149,12 @@
                                [(>= ?backoff ?time)]
                                [?e :com.github.ivarref.yoltq/tries ?tries]
                                [?e :com.github.ivarref.yoltq/id ?id]
-                               [?e :com.github.ivarref.yoltq/lock ?lock]]
+                               [?e :com.github.ivarref.yoltq/lock ?lock]
+                               [?e :com.github.ivarref.yoltq/version ?current-version]]
                              db
                              queue-name
-                             (- now hung-backoff-time))
+                             (- now hung-backoff-time)
+                             current-version)
                         (not-empty))]
       (let [new-lock (random-uuid)
             [id old-lock tries _t] (rand-nth (into [] ids))
-- 
cgit v1.2.3


From 8f945d8c0189ad73d862c988faa511e0a7b017df Mon Sep 17 00:00:00 2001
From: Ivar Refsdal <refsdal.ivar@gmail.com>
Date: Fri, 18 Nov 2022 14:12:50 +0100
Subject: Release 0.2.63: Add support for :encode and :decode function. Add
 :partition-fn. Fixes #1

---
 README.md                                      | 65 ++++++++++++++++-
 deps.edn                                       |  6 +-
 pom.xml                                        |  9 ++-
 src/com/github/ivarref/yoltq/impl.clj          | 95 +++++++++++++++----------
 src/com/github/ivarref/yoltq/utils.clj         |  1 -
 test/com/github/ivarref/yoltq/virtual_test.clj | 98 ++++++++++++++++++++++----
 6 files changed, 218 insertions(+), 56 deletions(-)

(limited to 'src/com/github/ivarref/yoltq/utils.clj')

diff --git a/README.md b/README.md
index 63b9ad3..a914cc7 100644
--- a/README.md
+++ b/README.md
@@ -131,7 +131,8 @@ Inspecting `(yq/put :q {:work 123})]` you will see something like this:
 
 This is the queue job as it will be stored into the database. 
 You can see that the payload, i.e. the second argument of `yq/put`,
-is persisted into the database. Thus the payload must be `pr-str`-able.
+is persisted into the database. Thus the payload must be `pr-str`-able (unless you have specified
+custom `:encode` and `:decode` functions that override this).
 
 
 A queue job will initially have status `:init`.
@@ -220,6 +221,47 @@ is shut down abruptly during processing of queue jobs.
 A queue job will remain in status `:error` once `:max-retries` (default: 100) have been reached.
 Ideally this will not happen. ¯\\\_(ツ)\_/¯
 
+### Custom encoding and decoding
+
+Yoltq will use `pr-str` and `clojure.edn/read-string` by default to encode and decode data.
+You may specify `:encode` and `:decode` either globally or per queue to override this behaviour.
+The `:encode` function must return a byte array or a string.
+
+For example if you want to use [nippy](https://github.com/ptaoussanis/nippy):
+```clojure
+(require '[taoensso.nippy :as nippy])
+
+; Globally for all queues:
+(yq/init!
+  {:conn conn
+   :encode nippy/freeze
+   :decode nippy/thaw})
+
+; Or per queue:
+(yq/add-consumer! 
+  :q ; Queue to consume  
+  (fn [payload] (println "got payload:" payload)) ; Queue consumer function
+  {:encode nippy/freeze
+   :decode nippy/thaw}) ; Queue options, here with :encode and :decode
+```
+
+### Partitions
+
+Yoltq supports specifying which [partition](https://docs.datomic.com/on-prem/schema/schema.html#partitions)
+queue entities should belong to.
+The default function is:
+```clojure
+(defn default-partition-fn [_queue-name]
+  (keyword "yoltq" (str "queue_" (.getValue (java.time.Year/now)))))
+```
+This is to say that there will be a single partition per year for yoltq.
+Yoltq will take care of creating the partition if it does not exist.
+
+You may override this function, either globally or per queue, with the keyword `:partition-fn`.
+E.g.:
+```clojure
+(yq/init! {:conn conn :partition-fn (fn [_queue-name] :my-partition)})
+```
 
 ### All configuration options
 
@@ -376,8 +418,29 @@ For Redis there is [carmine](https://github.com/ptaoussanis/carmine).
 
 Note: I have not tried these libraries myself.
 
+## Other stuff
+
+If you liked this library, you may also like:
+
+* [conformity](https://github.com/avescodes/conformity): A Clojure/Datomic library for idempotently transacting norms into your database – be they schema, data, or otherwise.
+* [datomic-schema](https://github.com/ivarref/datomic-schema): Simplified writing of Datomic schemas (works with conformity).
+* [double-trouble](https://github.com/ivarref/double-trouble):  Handle duplicate Datomic transactions with ease.
+* [gen-fn](https://github.com/ivarref/gen-fn): Generate Datomic function literals from regular Clojure namespaces.
+* [rewriting-history](https://github.com/ivarref/rewriting-history): A library to rewrite Datomic history.
+
 ## Change log
 
+#### 2022-11-18 v0.2.63 [diff](https://github.com/ivarref/yoltq/compare/v0.2.62...v0.2.63)
+Added custom `:encode` and `:decode` support.
+
+Added support for specifying `:partifion-fn` to specify which partition a queue item should belong to.
+It defaults to:
+```clojure
+(defn default-partition-fn [_queue-name]
+  (keyword "yoltq" (str "queue_" (.getValue (Year/now)))))
+```
+Yoltq takes care of creating the partition if it does not exist.
+
 #### 2022-11-15 v0.2.62 [diff](https://github.com/ivarref/yoltq/compare/v0.2.61...v0.2.62)
 Added function `processing-time-stats`:
 
diff --git a/deps.edn b/deps.edn
index 6923881..e36885e 100644
--- a/deps.edn
+++ b/deps.edn
@@ -1,5 +1,6 @@
-{:deps      {org.clojure/tools.logging {:mvn/version "1.2.4"}
-             org.clojure/clojure       {:mvn/version "1.11.1"}}
+{:deps      {com.github.ivarref/double-trouble {:mvn/version "0.1.102"}
+             org.clojure/tools.logging         {:mvn/version "1.2.4"}
+             org.clojure/clojure               {:mvn/version "1.11.1"}}
 
  :paths     ["src"]
 
@@ -11,6 +12,7 @@
                                      clojure-term-colors/clojure-term-colors {:mvn/version "0.1.0"}
                                      com.datomic/datomic-pro                 {:mvn/version "1.0.6316" :exclusions [org.slf4j/slf4j-nop]}
                                      org.postgresql/postgresql               {:mvn/version "9.3-1102-jdbc41"}
+                                     com.taoensso/nippy                      {:mvn/version "3.2.0"}
                                      io.github.cognitect-labs/test-runner    {:git/tag "v0.5.0" :git/sha "b3fd0d2"}}
                        :jvm-opts    ["-DDISABLE_SPY=true"
                                      "-DTAOENSSO_TIMBRE_MIN_LEVEL_EDN=:error"]
diff --git a/pom.xml b/pom.xml
index 2c11984..463899d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -4,7 +4,7 @@
   <packaging>jar</packaging>
   <groupId>com.github.ivarref</groupId>
   <artifactId>yoltq</artifactId>
-  <version>0.2.62</version>
+  <version>0.2.63</version>
   <name>yoltq</name>
   <dependencies>
     <dependency>
@@ -12,6 +12,11 @@
       <artifactId>clojure</artifactId>
       <version>1.11.1</version>
     </dependency>
+    <dependency>
+      <groupId>com.github.ivarref</groupId>
+      <artifactId>double-trouble</artifactId>
+      <version>0.1.102</version>
+    </dependency>
     <dependency>
       <groupId>org.clojure</groupId>
       <artifactId>tools.logging</artifactId>
@@ -30,7 +35,7 @@
   <scm>
     <connection>scm:git:git://github.com/ivarref/yoltq.git</connection>
     <developerConnection>scm:git:ssh://git@github.com/ivarref/yoltq.git</developerConnection>
-    <tag>v0.2.62</tag>
+    <tag>v0.2.63</tag>
     <url>https://github.com/ivarref/yoltq</url>
   </scm>
 </project>
\ No newline at end of file
diff --git a/src/com/github/ivarref/yoltq/impl.clj b/src/com/github/ivarref/yoltq/impl.clj
index c37b0e6..ac573d1 100644
--- a/src/com/github/ivarref/yoltq/impl.clj
+++ b/src/com/github/ivarref/yoltq/impl.clj
@@ -1,11 +1,12 @@
 (ns com.github.ivarref.yoltq.impl
-  (:require [datomic.api :as d]
-            [clojure.tools.logging :as log]
+  (:require [clojure.edn :as edn]
             [clojure.string :as str]
-            [com.github.ivarref.yoltq.utils :as u]
+            [clojure.tools.logging :as log]
+            [com.github.ivarref.double-trouble :as dt]
             [com.github.ivarref.yoltq.ext-sys :as ext]
-            [clojure.edn :as edn]))
-
+            [com.github.ivarref.yoltq.utils :as u]
+            [datomic.api :as d])
+  (:import (java.time Year)))
 
 (def schema
   [#:db{:ident :com.github.ivarref.yoltq/id, :cardinality :db.cardinality/one, :valueType :db.type/uuid, :unique :db.unique/identity}
@@ -13,6 +14,7 @@
    #: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/payload-bytes, :cardinality :db.cardinality/one, :valueType :db.type/bytes}
    #: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}
@@ -41,13 +43,22 @@
       (log/error "could not read-string" what ":" (ex-message e))
       (throw e))))
 
+(defn default-partition-fn [_queue-keyword]
+  (keyword "yoltq" (str "queue_" (.getValue (Year/now)))))
 
-(defn put [{:keys [capture-bindings conn] :as config}
+(defn put [{:keys [capture-bindings conn encode partition-fn]
+            :or   {partition-fn default-partition-fn
+                   encode       (partial pr-str-safe :payload)}
+            :as   config}
            queue-name
            payload
            opts]
   (if-let [q-config (get-in config [:handlers queue-name])]
     (let [id (u/squuid)
+          encode (get q-config :encode encode)
+          partition-fn (get q-config :partition-fn partition-fn)
+          partition (partition-fn queue-name)
+          _ (assert (keyword? partition) "Partition must be a keyword")
           depends-on (get q-config :depends-on (fn [_] nil))
           valid-payload? (get q-config :valid-payload? (fn [_] true))
           opts (merge
@@ -58,32 +69,41 @@
                                       (assoc o (symbol k) (deref k)))
                                     {}
                                     (or capture-bindings []))
-                            (pr-str-safe :capture-bindings))]
-      (when-not (valid-payload? payload)
-        (log/error "Payload was not valid. Payload was:" payload)
-        (throw (ex-info (str "Payload was not valid: " payload) {:payload payload})))
+                            (pr-str-safe :capture-bindings))
+          _ (when-not (valid-payload? payload)
+              (log/error "Payload was not valid. Payload was:" payload)
+              (throw (ex-info (str "Payload was not valid: " payload) {:payload payload})))
+          encoded (encode payload)
+          _ (when (not (or (bytes? encoded) (string? encoded)))
+              (log/error "Payload must be encoded to either a string or a byte array")
+              (throw (ex-info (str "Payload must be encoded to a string or a byte array. Payload: " payload) {:payload payload})))]
       (log/debug "queue item" (str id) "for queue" queue-name "is pending status" u/status-init)
-      (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-safe :payload payload)
-         :com.github.ivarref.yoltq/bindings   str-bindings
-         :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/version    "2"}
-        (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-safe :depends-on [q ext-id]))
-            (throw (ex-info (str ":depends-on not found in database. Queue: " q ", id: " ext-id) opts))))
-        (when-let [ext-id (:id opts)]
-          {:com.github.ivarref.yoltq/ext-id (pr-str-safe :id [queue-name ext-id])})))
+      (do
+        (dt/ensure-partition! conn partition)
+        (merge
+          (if (bytes? encoded)
+            {:com.github.ivarref.yoltq/payload-bytes encoded}
+            {:com.github.ivarref.yoltq/payload encoded})
+          {:db/id                               (d/tempid partition)
+           :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/bindings   str-bindings
+           :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/version    "2"}
+          (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-safe :depends-on [q ext-id]))
+              (throw (ex-info (str ":depends-on not found in database. Queue: " q ", id: " ext-id) opts))))
+          (when-let [ext-id (:id opts)]
+            {:com.github.ivarref.yoltq/ext-id (pr-str-safe :id [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})))))
@@ -169,20 +189,23 @@
                  "in" (format "%.1f" (double (/ spent-ns 1e6))) "ms"]))
 
 
-(defn execute! [{:keys [handlers mark-status-fn! start-execute-time collect-spent-time!]
-                 :or   {mark-status-fn! mark-status!}
+(defn execute! [{:keys [decode handlers mark-status-fn! start-execute-time collect-spent-time!]
+                 :or   {mark-status-fn! mark-status!
+                        decode          edn/read-string}
                  :as   cfg}
-                {:com.github.ivarref.yoltq/keys [status id queue-name payload] :as queue-item}]
+                {:com.github.ivarref.yoltq/keys [status id queue-name payload payload-bytes] :as queue-item}]
   (when queue-item
     (if (= :error status)
       (assoc queue-item :failed? true)
       (if-let [queue (get handlers queue-name)]
-        (let [{:keys [f allow-cas-failure?]} queue]
+        (let [{:keys [f allow-cas-failure?]} queue
+              decode (get queue :decode decode)]
           (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-ms) id queue-name])
-                  (let [v (f payload)]
+                  (let [payload (decode (or payload payload-bytes))
+                        v (f payload)]
                     {:retval v})
                   (catch Throwable t
                     {:exception t})
diff --git a/src/com/github/ivarref/yoltq/utils.clj b/src/com/github/ivarref/yoltq/utils.clj
index 39572a9..7665b6d 100644
--- a/src/com/github/ivarref/yoltq/utils.clj
+++ b/src/com/github/ivarref/yoltq/utils.clj
@@ -57,7 +57,6 @@
 (defn get-queue-item [db id]
   (-> (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]
diff --git a/test/com/github/ivarref/yoltq/virtual_test.clj b/test/com/github/ivarref/yoltq/virtual_test.clj
index 996792e..2800c21 100644
--- a/test/com/github/ivarref/yoltq/virtual_test.clj
+++ b/test/com/github/ivarref/yoltq/virtual_test.clj
@@ -1,18 +1,21 @@
 (ns com.github.ivarref.yoltq.virtual-test
-  (:require [clojure.test :refer [deftest is use-fixtures] :refer-macros [thrown?]]
-            [clojure.tools.logging :as log]
-            [com.github.ivarref.yoltq :as yq]
-            [com.github.ivarref.yoltq.error-poller :as error-poller]
-            [com.github.ivarref.yoltq.ext-sys :as ext]
-            [com.github.ivarref.yoltq.impl :as i]
-            [com.github.ivarref.yoltq.migrate :as migrate]
-            [com.github.ivarref.yoltq.test-queue :as tq]
-            [com.github.ivarref.yoltq.test-utils :as u]
-            [com.github.ivarref.yoltq.utils :as uu]
-            [datomic-schema.core]
-            [datomic.api :as d]
-            [taoensso.timbre :as timbre])
-  (:import (java.time Duration)))
+  (:require
+    [clojure.string :as str]
+    [clojure.test :refer [deftest is use-fixtures] :refer-macros [thrown?]]
+    [clojure.tools.logging :as log]
+    [com.github.ivarref.yoltq :as yq]
+    [com.github.ivarref.yoltq.error-poller :as error-poller]
+    [com.github.ivarref.yoltq.ext-sys :as ext]
+    [com.github.ivarref.yoltq.impl :as i]
+    [com.github.ivarref.yoltq.migrate :as migrate]
+    [com.github.ivarref.yoltq.test-queue :as tq]
+    [com.github.ivarref.yoltq.test-utils :as u]
+    [com.github.ivarref.yoltq.utils :as uu]
+    [datomic-schema.core]
+    [datomic.api :as d]
+    [taoensso.nippy :as nippy]
+    [taoensso.timbre :as timbre])
+  (:import (java.time Duration LocalDateTime)))
 
 
 (use-fixtures :each tq/call-with-virtual-queue!)
@@ -380,3 +383,70 @@
     (is (= 0 (error-poller/do-poll-errors @yq/*config* (ext/now-ms))))
     (is (= 0 (error-poller/do-poll-errors @yq/*config* (+ (dec (.toMillis (Duration/ofMinutes 15))) (ext/now-ms)))))
     (is (= 1 (error-poller/do-poll-errors @yq/*config* (+ (.toMillis (Duration/ofMinutes 15)) (ext/now-ms)))))))
+
+(deftest global-encode-decode
+  (let [conn (u/empty-conn)
+        ldt (LocalDateTime/now)
+        got-work (atom nil)]
+    (yq/init! {:conn conn
+               :encode nippy/freeze
+               :decode nippy/thaw})
+    (yq/add-consumer! :q (fn [work] (reset! got-work work)))
+    @(d/transact conn [(yq/put :q {:work ldt})])
+    (tq/consume! :q)
+    (is (= @got-work {:work ldt}))))
+
+(deftest queue-encode-decode
+  (let [conn (u/empty-conn)
+        ldt (LocalDateTime/now)
+        got-work (atom nil)]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q (fn [work] (reset! got-work work))
+                      {:encode nippy/freeze
+                       :decode nippy/thaw})
+    @(d/transact conn [(yq/put :q {:work ldt})])
+    (tq/consume! :q)
+    (is (= @got-work {:work ldt}))))
+
+(deftest global-partition
+  (let [conn (u/empty-conn)
+        got-work (atom nil)]
+    (yq/init! {:conn conn
+               :partition-fn (fn [_queue-name] :my-part)})
+    (yq/add-consumer! :q (fn [work] (reset! got-work work)))
+    @(d/transact conn [(yq/put :q {:work 123})])
+    (tq/consume! :q)
+    (is (some? (d/q '[:find ?e .
+                      :in $ ?part
+                      :where
+                      [?e :db/ident ?part]]
+                    (d/db conn)
+                    :my-part)))
+    (is (= @got-work {:work 123}))))
+
+(deftest partition-per-queue
+  (let [conn (u/empty-conn)
+        got-work (atom nil)]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q (fn [work] (reset! got-work work))
+                      {:partition-fn (fn [_queue-name] :my-part)})
+    @(d/transact conn [(yq/put :q {:work 123})])
+    (tq/consume! :q)
+    (is (some? (d/q '[:find ?e .
+                      :in $ ?part
+                      :where
+                      [?e :db/ident ?part]]
+                    (d/db conn)
+                    :my-part)))
+    (is (= @got-work {:work 123}))))
+
+(deftest string-encode-decode
+  (let [conn (u/empty-conn)
+        got-work (atom nil)]
+    (yq/init! {:conn   conn
+               :encode (fn [x] (str/join (reverse x)))
+               :decode (fn [x] (str/join (reverse x)))})
+    (yq/add-consumer! :q (fn [work] (reset! got-work work)))
+    @(d/transact conn [(yq/put :q "asdf")])
+    (tq/consume! :q)
+    (is (= @got-work "asdf"))))
-- 
cgit v1.2.3


From e848610ac341db31b804644a7dfaaf98389469d5 Mon Sep 17 00:00:00 2001
From: Ivar Refsdal <refsdal.ivar@gmail.com>
Date: Mon, 20 Mar 2023 12:24:00 +0100
Subject: Release 0.2.64: Allow for infinitive retries

---
 README.md                              | 16 +++++++++++++---
 pom.xml                                |  4 ++--
 src/com/github/ivarref/yoltq.clj       |  7 +++++--
 src/com/github/ivarref/yoltq/utils.clj | 17 +++++++++++------
 4 files changed, 31 insertions(+), 13 deletions(-)

(limited to 'src/com/github/ivarref/yoltq/utils.clj')

diff --git a/README.md b/README.md
index a914cc7..c5f2bdb 100644
--- a/README.md
+++ b/README.md
@@ -156,7 +156,9 @@ the payload. It can be added like this:
                             ; consumer function to ensure idempotence.
    :valid-payload? (fn [payload] (some? (:id payload))) ; Function that verifies payload. Should return truthy for valid payloads.
                                                         ; The default function always returns true.
-   :max-retries 10})        ; Specify maximum number of times an item will be retried. Default: 100
+   :max-retries 10})        ; Specify maximum number of times an item will be retried. Default: 10000.
+                            ; If :max-retries is given as 0, the job will ~always be retried, i.e.
+                            ; 9223372036854775807 times (Long/MAX_VALUE).
 ```
 
 The `payload` will be deserialized from the database using `clojure.edn/read-string` before invocation, i.e.
@@ -218,8 +220,10 @@ is shut down abruptly during processing of queue jobs.
 
 ### Giving up
 
-A queue job will remain in status `:error` once `:max-retries` (default: 100) have been reached.
-Ideally this will not happen. ¯\\\_(ツ)\_/¯
+A queue job will remain in status `:error` once `:max-retries` (default: 10000) have been reached.
+If `:max-retries` is given as `0`, the job will be retried 9223372036854775807 times before
+giving up.
+Ideally this should not happen. ¯\\\_(ツ)\_/¯
 
 ### Custom encoding and decoding
 
@@ -430,6 +434,12 @@ If you liked this library, you may also like:
 
 ## Change log
 
+#### 2023-03-20 v0.2.64 [diff](https://github.com/ivarref/yoltq/compare/v0.2.63...v0.2.64)
+Added support for `max-retries` being `0`, meaning the job should be retried forever
+(or at least 9223372036854775807 times).
+
+Changed the default for `max-retries` from `100` to `10000`.
+
 #### 2022-11-18 v0.2.63 [diff](https://github.com/ivarref/yoltq/compare/v0.2.62...v0.2.63)
 Added custom `:encode` and `:decode` support.
 
diff --git a/pom.xml b/pom.xml
index 463899d..466f47a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -4,7 +4,7 @@
   <packaging>jar</packaging>
   <groupId>com.github.ivarref</groupId>
   <artifactId>yoltq</artifactId>
-  <version>0.2.63</version>
+  <version>0.2.64</version>
   <name>yoltq</name>
   <dependencies>
     <dependency>
@@ -35,7 +35,7 @@
   <scm>
     <connection>scm:git:git://github.com/ivarref/yoltq.git</connection>
     <developerConnection>scm:git:ssh://git@github.com/ivarref/yoltq.git</developerConnection>
-    <tag>v0.2.63</tag>
+    <tag>v0.2.64</tag>
     <url>https://github.com/ivarref/yoltq</url>
   </scm>
 </project>
\ No newline at end of file
diff --git a/src/com/github/ivarref/yoltq.clj b/src/com/github/ivarref/yoltq.clj
index 9ffb3ad..379d701 100644
--- a/src/com/github/ivarref/yoltq.clj
+++ b/src/com/github/ivarref/yoltq.clj
@@ -24,7 +24,10 @@
   (-> {; Default number of times a queue job will be retried before giving up
        ; Can be overridden on a per-consumer basis with
        ; (yq/add-consumer! :q (fn [payload] ...) {:max-retries 200})
-       :max-retries                   100
+       ; If you want no limit on the number of retries, specify
+       ; the value `0`. That will set the effective retry limit to
+       ; 9223372036854775807 times.
+       :max-retries                   10000
 
        ; Minimum amount of time to wait before a failed queue job is retried
        :error-backoff-time            (Duration/ofSeconds 5)
@@ -244,7 +247,7 @@
 (defn retry-one-error! [qname]
   (let [{:keys [handlers] :as cfg} @*config*
         _ (assert (contains? handlers qname) "Queue not found")
-        cfg (assoc-in cfg [:handlers qname :max-retries] Integer/MAX_VALUE)]
+        cfg (assoc-in cfg [:handlers qname :max-retries] Long/MAX_VALUE)]
     (poller/poll-once! cfg qname :error)))
 
 (defn retry-stats
diff --git a/src/com/github/ivarref/yoltq/utils.clj b/src/com/github/ivarref/yoltq/utils.clj
index 7665b6d..9defd0e 100644
--- a/src/com/github/ivarref/yoltq/utils.clj
+++ b/src/com/github/ivarref/yoltq/utils.clj
@@ -103,13 +103,18 @@
         (prepare-processing db id queue-name old-lock :init))
       (log/debug "no new-items in :init status for queue" queue-name))))
 
+(defn- get-max-retries [cfg queue-name]
+  (let [v (get-in cfg [:handlers queue-name :max-retries] (:max-retries cfg))]
+    (if (and (number? v) (pos-int? v))
+      v
+      Long/MAX_VALUE)))
 
-(defn get-error [{:keys [conn db error-backoff-time max-retries] :as cfg} queue-name]
+(defn get-error [{:keys [conn db error-backoff-time] :as cfg} queue-name]
   (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 [db (or db (d/db conn))
-        max-retries (get-in cfg [:handlers queue-name :max-retries] max-retries)]
+        max-retries (get-max-retries cfg queue-name)]
     (when-let [ids (->> (d/q '[:find ?id ?lock
                                :in $ ?queue-name ?backoff ?max-tries ?current-version
                                :where
@@ -118,26 +123,26 @@
                                [?e :com.github.ivarref.yoltq/error-time ?time]
                                [(>= ?backoff ?time)]
                                [?e :com.github.ivarref.yoltq/tries ?tries]
-                               [(> ?max-tries ?tries)]
+                               [(>= ?max-tries ?tries)]
                                [?e :com.github.ivarref.yoltq/id ?id]
                                [?e :com.github.ivarref.yoltq/lock ?lock]
                                [?e :com.github.ivarref.yoltq/version ?current-version]]
                              db
                              queue-name
                              (- (now-ms) error-backoff-time)
-                             (inc max-retries)
+                             max-retries
                              current-version)
                         (not-empty))]
       (let [[id old-lock] (rand-nth (into [] ids))]
         (prepare-processing db id queue-name old-lock :error)))))
 
 
-(defn get-hung [{:keys [conn db now hung-backoff-time max-retries] :as cfg} queue-name]
+(defn get-hung [{:keys [conn db now hung-backoff-time] :as cfg} queue-name]
   (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-ms))
-        max-retries (get-in cfg [:handlers queue-name :max-retries] max-retries)
+        max-retries (get-max-retries cfg queue-name)
         db (or db (d/db conn))]
     (when-let [ids (->> (d/q '[:find ?id ?lock ?tries
                                :in $ ?qname ?backoff ?current-version
-- 
cgit v1.2.3