Merge remote-tracking branch 'upstream/main', where "upstream" is yoltq.

23 files changed, 3954 insertions, 0 deletions

diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
new file mode 100644
index 0000000..1350854
--- /dev/null
+++ b/.github/workflows/release.yml
@@ -0,0 +1,52 @@
+name: Tag and Release
+on: workflow_dispatch
+
+jobs:
+  tag-and-release:
+    runs-on: ubuntu-22.04
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          # NOTE: Fetch all for counting commits
+          fetch-depth: 0
+      - uses: actions/setup-java@v3
+        with:
+          distribution: 'adopt'
+          java-version: 21
+      - uses: DeLaGuardo/setup-clojure@13.4
+        with:
+          cli: 1.12.0.1530
+
+      - name: Show versions
+        run: |
+          java -version
+          clojure --version
+
+      - name: deploy to clojars
+        # NOTE: Specify ID to refer outputs from other steps
+        id: deploy
+        run: |
+          clojure -T:build deploy
+        env:
+          CLOJARS_PASSWORD: ${{secrets.CLOJARS_PASSWORD}}
+          CLOJARS_USERNAME: ${{secrets.CLOJARS_USERNAME}}
+
+      - uses: actions/create-release@v1
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          # NOTE: Refer outputs
+          tag_name: ${{ steps.deploy.outputs.version }}
+          release_name: ${{ steps.deploy.outputs.version }}
+          body: released
+          draft: false
+          prerelease: false
+
+      - run: |
+          clojure -T:build update-documents
+          git diff
+          git config --global user.email "github-actions@example.com"
+          git config --global user.name "github-actions"
+          git add -A
+          git commit -m "Update for release"
+          git push
\ No newline at end of file
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..707e1be
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,14 @@
+.idea/
+*.iml
+.cpcache/
+.nrepl-port
+target/
+.connkey
+tree.txt
+.db.url
+.stage-url.txt
+*.pom.asc
+*.pom
+temp/
+.clj-kondo/
+.rebel_readline_history
\ No newline at end of file
diff --git a/LICENSE b/LICENSE
new file mode 100644
index 0000000..d3087e4
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,277 @@
+Eclipse Public License - v 2.0
+
+    THE ACCOMPANYING PROGRAM IS PROVIDED UNDER THE TERMS OF THIS ECLIPSE
+    PUBLIC LICENSE ("AGREEMENT"). ANY USE, REPRODUCTION OR DISTRIBUTION
+    OF THE PROGRAM CONSTITUTES RECIPIENT'S ACCEPTANCE OF THIS AGREEMENT.
+
+1. DEFINITIONS
+
+"Contribution" means:
+
+  a) in the case of the initial Contributor, the initial content
+     Distributed under this Agreement, and
+
+  b) in the case of each subsequent Contributor:
+     i) changes to the Program, and
+     ii) additions to the Program;
+  where such changes and/or additions to the Program originate from
+  and are Distributed by that particular Contributor. A Contribution
+  "originates" from a Contributor if it was added to the Program by
+  such Contributor itself or anyone acting on such Contributor's behalf.
+  Contributions do not include changes or additions to the Program that
+  are not Modified Works.
+
+"Contributor" means any person or entity that Distributes the Program.
+
+"Licensed Patents" mean patent claims licensable by a Contributor which
+are necessarily infringed by the use or sale of its Contribution alone
+or when combined with the Program.
+
+"Program" means the Contributions Distributed in accordance with this
+Agreement.
+
+"Recipient" means anyone who receives the Program under this Agreement
+or any Secondary License (as applicable), including Contributors.
+
+"Derivative Works" shall mean any work, whether in Source Code or other
+form, that is based on (or derived from) the Program and for which the
+editorial revisions, annotations, elaborations, or other modifications
+represent, as a whole, an original work of authorship.
+
+"Modified Works" shall mean any work in Source Code or other form that
+results from an addition to, deletion from, or modification of the
+contents of the Program, including, for purposes of clarity any new file
+in Source Code form that contains any contents of the Program. Modified
+Works shall not include works that contain only declarations,
+interfaces, types, classes, structures, or files of the Program solely
+in each case in order to link to, bind by name, or subclass the Program
+or Modified Works thereof.
+
+"Distribute" means the acts of a) distributing or b) making available
+in any manner that enables the transfer of a copy.
+
+"Source Code" means the form of a Program preferred for making
+modifications, including but not limited to software source code,
+documentation source, and configuration files.
+
+"Secondary License" means either the GNU General Public License,
+Version 2.0, or any later versions of that license, including any
+exceptions or additional permissions as identified by the initial
+Contributor.
+
+2. GRANT OF RIGHTS
+
+  a) Subject to the terms of this Agreement, each Contributor hereby
+  grants Recipient a non-exclusive, worldwide, royalty-free copyright
+  license to reproduce, prepare Derivative Works of, publicly display,
+  publicly perform, Distribute and sublicense the Contribution of such
+  Contributor, if any, and such Derivative Works.
+
+  b) Subject to the terms of this Agreement, each Contributor hereby
+  grants Recipient a non-exclusive, worldwide, royalty-free patent
+  license under Licensed Patents to make, use, sell, offer to sell,
+  import and otherwise transfer the Contribution of such Contributor,
+  if any, in Source Code or other form. This patent license shall
+  apply to the combination of the Contribution and the Program if, at
+  the time the Contribution is added by the Contributor, such addition
+  of the Contribution causes such combination to be covered by the
+  Licensed Patents. The patent license shall not apply to any other
+  combinations which include the Contribution. No hardware per se is
+  licensed hereunder.
+
+  c) Recipient understands that although each Contributor grants the
+  licenses to its Contributions set forth herein, no assurances are
+  provided by any Contributor that the Program does not infringe the
+  patent or other intellectual property rights of any other entity.
+  Each Contributor disclaims any liability to Recipient for claims
+  brought by any other entity based on infringement of intellectual
+  property rights or otherwise. As a condition to exercising the
+  rights and licenses granted hereunder, each Recipient hereby
+  assumes sole responsibility to secure any other intellectual
+  property rights needed, if any. For example, if a third party
+  patent license is required to allow Recipient to Distribute the
+  Program, it is Recipient's responsibility to acquire that license
+  before distributing the Program.
+
+  d) Each Contributor represents that to its knowledge it has
+  sufficient copyright rights in its Contribution, if any, to grant
+  the copyright license set forth in this Agreement.
+
+  e) Notwithstanding the terms of any Secondary License, no
+  Contributor makes additional grants to any Recipient (other than
+  those set forth in this Agreement) as a result of such Recipient's
+  receipt of the Program under the terms of a Secondary License
+  (if permitted under the terms of Section 3).
+
+3. REQUIREMENTS
+
+3.1 If a Contributor Distributes the Program in any form, then:
+
+  a) the Program must also be made available as Source Code, in
+  accordance with section 3.2, and the Contributor must accompany
+  the Program with a statement that the Source Code for the Program
+  is available under this Agreement, and informs Recipients how to
+  obtain it in a reasonable manner on or through a medium customarily
+  used for software exchange; and
+
+  b) the Contributor may Distribute the Program under a license
+  different than this Agreement, provided that such license:
+     i) effectively disclaims on behalf of all other Contributors all
+     warranties and conditions, express and implied, including
+     warranties or conditions of title and non-infringement, and
+     implied warranties or conditions of merchantability and fitness
+     for a particular purpose;
+
+     ii) effectively excludes on behalf of all other Contributors all
+     liability for damages, including direct, indirect, special,
+     incidental and consequential damages, such as lost profits;
+
+     iii) does not attempt to limit or alter the recipients' rights
+     in the Source Code under section 3.2; and
+
+     iv) requires any subsequent distribution of the Program by any
+     party to be under a license that satisfies the requirements
+     of this section 3.
+
+3.2 When the Program is Distributed as Source Code:
+
+  a) it must be made available under this Agreement, or if the
+  Program (i) is combined with other material in a separate file or
+  files made available under a Secondary License, and (ii) the initial
+  Contributor attached to the Source Code the notice described in
+  Exhibit A of this Agreement, then the Program may be made available
+  under the terms of such Secondary Licenses, and
+
+  b) a copy of this Agreement must be included with each copy of
+  the Program.
+
+3.3 Contributors may not remove or alter any copyright, patent,
+trademark, attribution notices, disclaimers of warranty, or limitations
+of liability ("notices") contained within the Program from any copy of
+the Program which they Distribute, provided that Contributors may add
+their own appropriate notices.
+
+4. COMMERCIAL DISTRIBUTION
+
+Commercial distributors of software may accept certain responsibilities
+with respect to end users, business partners and the like. While this
+license is intended to facilitate the commercial use of the Program,
+the Contributor who includes the Program in a commercial product
+offering should do so in a manner which does not create potential
+liability for other Contributors. Therefore, if a Contributor includes
+the Program in a commercial product offering, such Contributor
+("Commercial Contributor") hereby agrees to defend and indemnify every
+other Contributor ("Indemnified Contributor") against any losses,
+damages and costs (collectively "Losses") arising from claims, lawsuits
+and other legal actions brought by a third party against the Indemnified
+Contributor to the extent caused by the acts or omissions of such
+Commercial Contributor in connection with its distribution of the Program
+in a commercial product offering. The obligations in this section do not
+apply to any claims or Losses relating to any actual or alleged
+intellectual property infringement. In order to qualify, an Indemnified
+Contributor must: a) promptly notify the Commercial Contributor in
+writing of such claim, and b) allow the Commercial Contributor to control,
+and cooperate with the Commercial Contributor in, the defense and any
+related settlement negotiations. The Indemnified Contributor may
+participate in any such claim at its own expense.
+
+For example, a Contributor might include the Program in a commercial
+product offering, Product X. That Contributor is then a Commercial
+Contributor. If that Commercial Contributor then makes performance
+claims, or offers warranties related to Product X, those performance
+claims and warranties are such Commercial Contributor's responsibility
+alone. Under this section, the Commercial Contributor would have to
+defend claims against the other Contributors related to those performance
+claims and warranties, and if a court requires any other Contributor to
+pay any damages as a result, the Commercial Contributor must pay
+those damages.
+
+5. NO WARRANTY
+
+EXCEPT AS EXPRESSLY SET FORTH IN THIS AGREEMENT, AND TO THE EXTENT
+PERMITTED BY APPLICABLE LAW, THE PROGRAM IS PROVIDED ON AN "AS IS"
+BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, EITHER EXPRESS OR
+IMPLIED INCLUDING, WITHOUT LIMITATION, ANY WARRANTIES OR CONDITIONS OF
+TITLE, NON-INFRINGEMENT, MERCHANTABILITY OR FITNESS FOR A PARTICULAR
+PURPOSE. Each Recipient is solely responsible for determining the
+appropriateness of using and distributing the Program and assumes all
+risks associated with its exercise of rights under this Agreement,
+including but not limited to the risks and costs of program errors,
+compliance with applicable laws, damage to or loss of data, programs
+or equipment, and unavailability or interruption of operations.
+
+6. DISCLAIMER OF LIABILITY
+
+EXCEPT AS EXPRESSLY SET FORTH IN THIS AGREEMENT, AND TO THE EXTENT
+PERMITTED BY APPLICABLE LAW, NEITHER RECIPIENT NOR ANY CONTRIBUTORS
+SHALL HAVE ANY LIABILITY FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING WITHOUT LIMITATION LOST
+PROFITS), HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ARISING IN ANY WAY OUT OF THE USE OR DISTRIBUTION OF THE PROGRAM OR THE
+EXERCISE OF ANY RIGHTS GRANTED HEREUNDER, EVEN IF ADVISED OF THE
+POSSIBILITY OF SUCH DAMAGES.
+
+7. GENERAL
+
+If any provision of this Agreement is invalid or unenforceable under
+applicable law, it shall not affect the validity or enforceability of
+the remainder of the terms of this Agreement, and without further
+action by the parties hereto, such provision shall be reformed to the
+minimum extent necessary to make such provision valid and enforceable.
+
+If Recipient institutes patent litigation against any entity
+(including a cross-claim or counterclaim in a lawsuit) alleging that the
+Program itself (excluding combinations of the Program with other software
+or hardware) infringes such Recipient's patent(s), then such Recipient's
+rights granted under Section 2(b) shall terminate as of the date such
+litigation is filed.
+
+All Recipient's rights under this Agreement shall terminate if it
+fails to comply with any of the material terms or conditions of this
+Agreement and does not cure such failure in a reasonable period of
+time after becoming aware of such noncompliance. If all Recipient's
+rights under this Agreement terminate, Recipient agrees to cease use
+and distribution of the Program as soon as reasonably practicable.
+However, Recipient's obligations under this Agreement and any licenses
+granted by Recipient relating to the Program shall continue and survive.
+
+Everyone is permitted to copy and distribute copies of this Agreement,
+but in order to avoid inconsistency the Agreement is copyrighted and
+may only be modified in the following manner. The Agreement Steward
+reserves the right to publish new versions (including revisions) of
+this Agreement from time to time. No one other than the Agreement
+Steward has the right to modify this Agreement. The Eclipse Foundation
+is the initial Agreement Steward. The Eclipse Foundation may assign the
+responsibility to serve as the Agreement Steward to a suitable separate
+entity. Each new version of the Agreement will be given a distinguishing
+version number. The Program (including Contributions) may always be
+Distributed subject to the version of the Agreement under which it was
+received. In addition, after a new version of the Agreement is published,
+Contributor may elect to Distribute the Program (including its
+Contributions) under the new version.
+
+Except as expressly stated in Sections 2(a) and 2(b) above, Recipient
+receives no rights or licenses to the intellectual property of any
+Contributor under this Agreement, whether expressly, by implication,
+estoppel or otherwise. All rights in the Program not expressly granted
+under this Agreement are reserved. Nothing in this Agreement is intended
+to be enforceable by any entity that is not a Contributor or Recipient.
+No third-party beneficiary rights are created under this Agreement.
+
+Exhibit A - Form of Secondary Licenses Notice
+
+"This Source Code may also be made available under the following 
+Secondary Licenses when the conditions for such availability set forth 
+in the Eclipse Public License, v. 2.0 are satisfied: {name license(s),
+version(s), and exceptions or additional permissions here}."
+
+  Simply including a copy of this Agreement, including this Exhibit A
+  is not sufficient to license the Source Code under Secondary Licenses.
+
+  If it is not possible or desirable to put the notice in a particular
+  file, then You may include the notice in a location (such as a LICENSE
+  file in a relevant directory) where a recipient would be likely to
+  look for such a notice.
+
+  You may add additional accurate notices of copyright ownership.
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..b64ed12
--- /dev/null
+++ b/README.md
@@ -0,0 +1,769 @@
+# yoltq
+
+An opinionated Datomic queue for building (more) reliable systems.
+Implements the
+[transactional outbox](https://microservices.io/patterns/data/transactional-outbox.html)
+pattern.
+Supports retries, backoff, ordering and more.
+On-prem only.
+
+## Installation
+
+[![Clojars Project](https://img.shields.io/clojars/v/com.github.ivarref/yoltq.svg)](https://clojars.org/com.github.ivarref/yoltq)
+
+## 1-minute example
+
+```clojure
+(require '[com.github.ivarref.yoltq :as yq])
+
+(def conn (datomic.api/connect "..."))
+
+; Initialize system
+(yq/init! {:conn conn})
+
+; Add a queue consumer that will intentionally fail on the first attempt
+(yq/add-consumer! :q
+                 (let [cnt (atom 0)]
+                   (fn [payload]
+                     (when (= 1 (swap! cnt inc))
+                       ; A consumer throwing an exception is considered a queue job failure
+                       (throw (ex-info "failed" {})))
+                     ; Anything else than a throwing exception is considered a queue job success
+                     ; This includes nil, false and everything else.
+                     (log/info "got payload" payload))))
+
+; Start threadpool that picks up queue jobs
+(yq/start!)
+
+; Queue a job
+@(d/transact conn [(yq/put :q {:work 123})])
+
+; On your console you will see something like this:
+; 17:29:54.598 DEBUG queue item 613... for queue :q is pending status :init
+; 17:29:54.602 DEBUG queue item 613... for queue :q now has status :processing
+; 17:29:54.603 DEBUG queue item 613... for queue :q is now processing
+; 17:29:54.605 WARN  queue-item 613... for queue :q now has status :error after 1 try in 4.8 ms
+; 17:29:54.607 WARN  error message was: "failed" for queue-item 613...
+; 17:29:54.615 WARN  ex-data was: {} for queue-item 613...
+; The item is so far failed...
+
+; But after approximately 10 seconds have elapsed, the item will be retried:
+; 17:30:05.596 DEBUG queue item 613... for queue :q now has status :processing
+; 17:30:05.597 DEBUG queue item 613... for queue :q is now processing
+; 17:30:05.597 INFO  got payload {:work 123}
+; 17:30:05.599 INFO  queue-item 613... for queue :q now has status :done after 2 tries in 5999.3 ms
+; And then it has succeeded.
+```
+
+## Rationale
+
+Integrating with external systems that may be unavailable can be tricky.
+Imagine the following code:
+
+```clojure
+(defn post-handler [user-input]
+  (let [db-item (process user-input)
+        ext-ref (clj-http.client/post ext-service {:connection-timeout 3000 ; milliseconds 
+                                                   :socket-timeout     10000 ; milliseconds
+                                                   ...})] ; may throw exception
+    @(d/transact conn [(assoc db-item :some/ext-ref ext-ref)])))
+```
+
+What if the POST request fails? Should it be retried? For how long?
+Should it be allowed to fail? How do you then process failures later?
+
+PS: If you do not set connection/socket-timeout, there is a chance that
+clj-http/client will wait for all eternity in the case of a dropped TCP connection.
+
+The queue way to solve this would be:
+
+```clojure
+(defn get-ext-ref [{:keys [id]}]
+  (let [ext-ref (clj-http.client/post ext-service {:connection-timeout 3000  ; milliseconds 
+                                                   :socket-timeout     10000 ; milliseconds
+                                                   ...})] ; may throw exception
+    @(d/transact conn [[:db/cas [:some/id id]
+                        :some/ext-ref
+                        nil
+                        ext-ref]])))
+
+(yq/add-consumer! :get-ext-ref get-ext-ref {:allow-cas-failure? true})
+
+(defn post-handler [user-input]
+  (let [{:some/keys [id] :as db-item} (process user-input)]
+    @(d/transact conn [db-item
+                       (yq/put :get-ext-ref {:id id})])))
+```
+
+Here `post-handler` will always succeed as long as the transaction commits.
+
+`get-ext-ref` may fail multiple times if `ext-service` is down.
+This is fine as long as it eventually succeeds.
+
+There is a special case where `get-ext-ref` succeeds, but 
+saving the new queue job status to the database fails.
+Thus `get-ext-ref` and any queue consumer should tolerate to 
+be executed successfully several times.
+
+For `get-ext-ref` this is solved by using
+the database function
+[:db/cas (compare-and-swap)](https://docs.datomic.com/on-prem/transactions/transaction-functions.html#dbfn-cas)
+to achieve a write-once behaviour.
+The yoltq system treats cas failures as job successes
+when a consumer has `:allow-cas-failure?` set to `true` in its options.
+
+## How it works
+
+### Queue jobs
+
+Creating queue jobs is done by `@(d/transact conn [...other data... (yq/put :q {:work 123})])`.
+Inspecting `(yq/put :q {:work 123})]` you will see something like this:
+
+```clojure
+#:com.github.ivarref.yoltq{:id #uuid"614232a8-e031-45bb-8660-be146eaa32a2", ; Queue job id 
+                           :queue-name :q, ; Destination queue                                 
+                           :status :init, ; Status
+                           :payload "{:work 123}", ; Payload persisted to the database with pr-str
+                           :bindings "{}", ; Bindings that will be applied before executing consumer function
+                           :lock #uuid"037d7da1-5158-4243-8f72-feb1e47e15ca", ; Lock to protect from multiple consumers
+                           :tries 0, ; How many times the job has been executed
+                           :init-time 4305758012289 ; Time of initialization (System/nanoTime)
+                           }
+```
+
+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 (unless you have specified
+custom `:encode` and `:decode` functions that override this).
+
+
+A queue job will initially have status `:init`.
+It will then transition to the following statuses:
+
+* `:processing`: When the queue job begins processing in the queue consumer function.
+* `:done`: If the queue consumer function returns normally.
+* `:error`: If the queue consumer function throws an exception.
+
+### Queue consumers
+
+Queue jobs will be consumed by queue consumers. A consumer is a function taking a single argument,
+the payload. It can be added like this:
+
+```clojure
+(yq/add-consumer! 
+  :q ; Queue to consume  
+  (fn [payload] (println "got payload:" payload)) ; Queue consumer function
+  ; An optional map of queue opts
+  {:allow-cas-failure? true ; Treat [:db.cas ...] failures as success. This is one way for the
+                            ; 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: 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. you will get back what you put into `yq/put`.
+
+The yoltq system treats a queue consumer function invocation as successful if it does not throw
+an exception. Any return value, be it `nil`, `false`, `true`, etc. is considered a success.
+
+### Listening for queue jobs
+
+When `(yq/start!)` is invoked, a threadpool is started.
+
+One thread is permanently allocated for listening to the 
+[tx-report-queue](https://docs.datomic.com/on-prem/clojure/index.html#datomic.api/tx-report-queue)
+and responding to changes. This means that yoltq will respond 
+and process newly created queue jobs fairly quickly.
+This also means that queue jobs in status `:init` will almost always be processed without
+any type of backoff.
+
+The threadpool also schedules polling jobs that will check for various statuses regularly:
+
+* Jobs in status `:error` that have waited for at least `:error-backoff-time` (default: 5 seconds) will be retried.
+* Jobs that have been in `:processing` for at least `:hung-backoff-time` (default: 30 minutes) will be considered hung and retried.
+* Old `:init-backoff-time` (default: 1 minute) `:init` jobs that have not been processed. Queue jobs can be left in status `:init` during application restart/upgrade, and thus the need for this strategy.
+
+
+### Retry and backoff strategy
+
+Yoltq assumes that if a queue consumer throws an exception for one item, it
+will also do the same for another item in the immediate future, 
+assuming the remote system that the queue consumer represents is still down.
+Thus if there are ten failures for queue `:q`, it does not make sense to
+retry all of them at once.
+
+The retry polling job that runs regularly (`:poll-delay`, default: every 10 seconds)
+thus stops at the first failure.
+Each queue have their own polling job, so if one queue is down, it will *not* stop
+other queues from retrying.
+
+The retry polling job will continue to eagerly process queue jobs as long as it 
+encounters only successes.
+
+While the `:error-backoff-time` of default 5 seconds may seem short, in practice
+if there is a lot of failed items and the external system is still down,
+the actual backoff time will be longer.
+
+
+### Stuck threads and stale jobs
+
+A single thread is dedicated to monitoring how much time a queue consumer 
+spends on a single job. If this exceeds `:max-execute-time` (default: 5 minutes)
+the stack trace of the offending consumer will be logged as `:ERROR`.
+
+If a job is found stale, that is if the database spent time exceeds 
+`:hung-backoff-time` (default: 30 minutes),
+the job will either be retried or marked as `:error`. This case may happen if the application
+is shut down abruptly during processing of queue jobs.
+
+
+### Giving up
+
+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
+
+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
+
+For an exhaustive list of all configuration options,
+see
+[yq/default-opts](https://github.com/ivarref/yoltq/blob/main/src/com/github/ivarref/yoltq.clj#L21).
+
+# Groups of Jobs
+
+Yoltq supports grouping jobs in a queue, and tracking the progress of such a
+group of jobs. Consider this example: your system is used by the marketing
+department to send emails to groups of users. Multiple colleagues in the
+marketing department could potentially do this at the same time, but they want
+to see the progress of their _own_ campagne, not that of _all_ emails being
+sent. When adding the jobs to the queue, you can specify the `job-group`
+parameter, in this case indicate which marketeer is running the jobs:
+
+```clojure
+(doseq [uid user-ids]
+  @(d/transact conn [(yq/put :send-mail
+                             ; Payload:
+                             {:user-id uid :from ... :to ... :body ...}
+                             ; Job options:
+                             {:job-group :mail-campagne/for-marketeer-42})]))
+```
+
+When you want to know the progress of that specific job group, and display it in
+your user interface, you can use `job-group-progress`, which returns a structure
+similar to `queue-stats`:
+
+```clojure
+(yq/job-group-progress :send-mail :mail-campagne/for-marketeer-42)
+;; => [{:qname :send-mail
+;;      :job-group :mail-campagne/for-marketeer-42
+;;      :status :init
+;;      :count 78}
+;;     {:qname :send-mail
+;;      :job-group :mail-campagne/for-marketeer-42
+;;      :status :done
+;;      :count 24}]
+```
+
+# Regular and REPL usage
+
+For a regular system and/or REPL session you'll want to do:
+
+```clojure
+(require '[com.github.ivarref.yoltq :as yq])
+
+(yq/init! {:conn conn})
+
+(yq/add-consumer! :q-one ...)
+(yq/add-consumer! :q-two ...)
+
+; Start yoltq system
+(yq/start!)
+
+; Oops I need another consumer. This works fine:
+(yq/add-consumer! :q-three ...)
+
+; When the application is shutting down:
+(yq/stop!)
+```
+
+You may invoke `yq/add-consumer!` and `yq/init!` on a live system as you like.
+
+If you change `:pool-size` or `:poll-delay` you will have to `(yq/stop!)` and
+`(yq/start!)` to make changes take effect.
+
+## Queue job dependencies and ordering
+
+It is possible to specify that one queue job must wait for another queue
+job to complete before it will be executed:
+
+```clojure
+@(d/transact conn [(yq/put :a 
+                           ; Payload:
+                           {:id "a1"}
+                           ; Job options:
+                           {:id "a1"})])
+
+@(d/transact conn [(yq/put :b 
+                           ; Payload:
+                           {:id "b1" :a-ref "a1"}
+                           ; Jobs options:
+                           {:depends-on [:a "a1"]})])
+
+; depends-on may also be specified as a function of the payload when 
+; adding the consumer:
+(yq/add-consumer! :b 
+                  (fn [payload] ...)
+                  {:depends-on (fn [payload]
+                                 [:a (:a-ref payload)])})
+```
+
+Here queue job `b1` will not execute before `a1` is `:done`.
+
+Note that queue-name plus `:id` in job options must be an unique value.
+In the example above that means `:a` plus `a1` must be unique.
+
+When specifying `:depends-on`, the referred job must at least exist in the database,
+otherwise `yq/put` will throw an exception.
+
+Other than this there is no attempt at ordering the execution of queue jobs.
+In fact the opposite is done in the poller to guard against the case that a single failing queue job
+could effectively take down the entire retry polling job.
+
+## Retrying jobs in the REPL
+
+```clojure
+(require '[com.github.ivarref.yoltq :as yq])
+
+; List jobs that are in state error:
+(yq/get-errors :q)
+
+; This will retry a single job that is in error, regardless 
+; of how many times it has been retried earlier.
+; If the job fails, you will get the full stacktrace on the REPL.
+(yq/retry-one-error! :q)
+; Returns a map containing the new state of the job.
+; Returns nil if there are no (more) jobs in state error for this queue.
+```
+
+# Testing
+
+For testing you will probably want determinism over an extra threadpool
+by using the test queue:
+
+```clojure
+...
+(:require [clojure.test :refer :all]
+  [com.github.ivarref.yoltq :as yq]
+  [com.github.ivarref.yoltq.test-queue :as tq])
+
+; Enables the test queue and disables the threadpool for each test.
+; yq/start! and yq/stop! becomes a no-op.
+(use-fixtures :each tq/call-with-virtual-queue!)
+
+(deftest demo
+         (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)))
+           
+           ; If you want to test the idempotence of your function, 
+           ; you may force retry a consumer function:
+           ; This may for example be useful to verify that the
+           ; :db.cas logic is correct.
+           (is (= {:work 123} (tq/force-retry! :q)))))
+```
+
+## Logging and capturing bindings
+
+Yoltq can capture and restore dynamic bindings.
+It will capture during `yq/put` and restore them when the consumer function
+is invoked. This is specified in the `:capture-bindings` setting.
+It defaults to `['#taoensso.timbre/*context*]`, 
+i.e. the [timbre](https://github.com/ptaoussanis/timbre) log context,
+if available, otherwise an empty vector.
+
+These dynamic bindings will be in place when yoltq logs errors, warnings
+etc. about failing consumer functions, possibly making troubleshooting
+easier.
+
+## Limitations
+
+Datomic does not have anything like `for update skip locked`.
+Thus consuming a queue should be limited to a single JVM process.
+This library will take queue jobs by compare-and-swapping a lock+state,
+process the item and then compare-and-swapping the lock+new-state.
+It does so eagerly, thus if you have multiple JVM consumers you will
+most likely get many locking conflicts. It should work, but it's far
+from optimal.
+
+## Alternatives
+
+I did not find any alternatives for Datomic.
+
+If I were using PostgreSQL or any other database that supports
+`for update skip locked`, I'd use a queue that uses this.
+For Clojure there is [proletarian](https://github.com/msolli/proletarian).
+
+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
+
+#### [Unreleased]
+
+#### [0.2.94] - 2025-09-22
+
+Added support for [groups of jobs](#groups-of-jobs).
+Thanks [Stefan van den Oord](https://github.com/svdo)!
+
+#### [0.2.85] - 2025-07-29
+
+Same as v0.2.82, but without the `v` prefix.
+
+#### [v0.2.82] - 2025-06-18
+
+Added support for specifying `tx-report-queue` as a keyword in `init!`. Yoltq will
+then not grab the datomic report queue, but use the one provided:
+
+```clojure
+(require '[com.github.ivarref.yoltq :as yq])
+(yq/init! {:conn            conn
+           :tx-report-queue (yq/get-tx-report-queue-multicast! conn :yoltq)
+                            ; ^^ can be any `java.util.concurrent.BlockingQueue` value
+                            })
+
+(another-tx-report-consumer! (yq/get-tx-report-queue-multicast! conn :another-consumer-id))
+
+```
+
+Added multicast support for `datomic.api/tx-report-queue`:
+```clojure
+(require '[com.github.ivarref.yoltq :as yq])
+(def my-q1 (yq/get-tx-report-queue-multicast! conn :q-id-1))
+; ^^ consume my-q1 just like you would do `datomic.api/tx-report-queue`
+
+(def my-q2 (yq/get-tx-report-queue-multicast! conn :q-id-2))
+; Both my-q1 and my-q2 will receive everything from `datomic.api/tx-report-queue`
+; for the given `conn`
+
+(def my-q3 (yq/get-tx-report-queue-multicast! conn :q-id-3 true))
+; my-q3 sets the optional third argument, `send-end-token?`, to true.
+; The queue will then receive `:end` if the queue is stopped.
+; This can enable simpler consuming of queues:
+(future
+  (loop []
+    (let [q-item (.take ^java.util.concurrent.BlockingQueue my-q3)]
+      (if (= q-item :end)
+        (println "Time to exit. Goodbye!")
+        (do
+          (println "Processing q-item" q-item)
+          (recur))))))
+
+; The default value for `send-end-token?` is `false`, i.e. the behaviour will be
+; identical to that of datomic.api/tx-report-queue.
+
+@(d/transact conn [{:db/doc "new-data"}])
+
+; Stop the queue:
+(yq/stop-multicast-consumer-id! conn :q-id-3)
+=> true
+; The multicaster thread will send `:end`.
+; The consumer thread will then print "Time to exit. Goodbye!".
+
+; if the queue is already stopped (or never was started), the `stop-multicaster...` 
+; functions will return false:
+(yq/stop-multicast-consumer-id! conn :already-stopped-queue-or-typo)
+=> false
+
+; Stop all queues for all connections:
+(yq/stop-all-multicasters!)
+```
+
+`yq/get-tx-report-queue-multicast!` returns, like
+`datomic.api/tx-report-queue`,
+`java.util.concurrent.BlockingQueue` and starts a background thread that does 
+the multicasting as needed. Identical calls to `yq/get-tx-report-queue-multicast!`
+returns the same `BlockingQueue`.
+
+Changed the default for `max-retries` from `10000` to `9223372036854775807`.
+
+Fixed reflection warnings.
+
+#### 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.
+
+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`:
+
+```clojure
+(ns com.github.ivarref.yoltq)
+
+(defn processing-time-stats
+  "Gather processing time statistics.
+
+  Optional keyword arguments:
+  * :age-days —  last number of days to look at data from. Defaults to 30.
+                 Use nil to have no limit.
+
+  * :queue-name — only gather statistics for this queue name. Defaults to nil, meaning all queues.
+
+  * :duration->long - Specify what unit should be used for values.
+                      Must take a java.time.Duration as input and return a long.
+
+                      Defaults to (fn [duration] (.toSeconds duration).
+                      I.e. the default unit is seconds.
+
+  Example return value:
+  {:queue-a {:avg 1
+             :max 10
+             :min 0
+             :p50 ...
+             :p90 ...
+             :p95 ...
+             :p99 ...}}"
+ [{:keys [age-days queue-name now db duration->long]
+  :or   {age-days 30
+         now      (ZonedDateTime/now ZoneOffset/UTC)
+         duration->long (fn [duration] (.toSeconds duration))}}]
+  ...)
+```
+
+#### 2022-09-07 v0.2.61 [diff](https://github.com/ivarref/yoltq/compare/v0.2.60...v0.2.61)
+Added function `retry-stats`:
+
+```clojure
+(ns com.github.ivarref.yoltq)
+
+(defn retry-stats
+  "Gather retry statistics.
+
+  Optional keyword arguments:
+  * :age-days —  last number of days to look at data from. Defaults to 30.
+  * :queue-name — only gather statistics for this queue name. Defaults to nil, meaning all queues.
+
+  Example return value:
+  {:queue-a {:ok 100, :retries 2, :retry-percentage 2.0}
+   :queue-b {:ok 100, :retries 75, :retry-percentage 75.0}}
+
+  From the example value above, we can see that :queue-b fails at a much higher rate than :queue-a.
+  Assuming that the queue consumers are correctly implemented, this means that the service representing :queue-b
+  is much more unstable than the one representing :queue-a. This again implies
+  that you will probably want to fix the downstream service of :queue-b, if that is possible.
+  "
+  [{:keys [age-days queue-name now]
+    :or   {age-days 30
+           now      (ZonedDateTime/now ZoneOffset/UTC)}}]
+  ...)
+```
+
+#### 2022-08-18 v0.2.60 [diff](https://github.com/ivarref/yoltq/compare/v0.2.59...v0.2.60)
+Improved: Added config option `:healthy-allowed-error-time`:
+```
+    ; If you are dealing with a flaky downstream service, you may not want
+    ; yoltq to mark itself as unhealthy on the first failure encounter with
+    ; the downstream service. Change this setting to let yoltq mark itself
+    ; as healthy even though a queue item has been failing for some time.
+    :healthy-allowed-error-time    (Duration/ofMinutes 15)
+```
+
+#### 2022-08-15 v0.2.59 [diff](https://github.com/ivarref/yoltq/compare/v0.2.58...v0.2.59)
+Fixed:
+* Race condition that made the following possible: `stop!` would terminate the slow thread 
+watcher, and a stuck thread could keep `stop!` from completing! 
+
+#### 2022-06-30 v0.2.58 [diff](https://github.com/ivarref/yoltq/compare/v0.2.57...v0.2.58)
+Slightly more safe EDN printing and parsing.
+Recommended reading:
+[Pitfalls and bumps in Clojure's Extensible Data Notation (EDN)](https://nitor.com/en/articles/pitfalls-and-bumps-clojures-extensible-data-notation-edn)
+
+#### 2022-06-29 v0.2.57 [diff](https://github.com/ivarref/yoltq/compare/v0.2.56...v0.2.57)
+Added `(get-errors qname)` and `(retry-one-error! qname)`.
+
+Improved:
+`unhealthy?` will return `false` for the first 10 minutes of the application lifetime.
+This was done in order to push new code while a queue was in error in an earlier
+version of the code. In this way rolling upgrades are possible regardless if there
+are queue errors.
+Can you tell that this issue hit me? ¯\\\_(ツ)\_/¯
+
+#### 2022-06-22 v0.2.56 [diff](https://github.com/ivarref/yoltq/compare/v0.2.55...v0.2.56)
+Added support for `:yoltq/queue-id` metadata on functions. I.e. it's possible to write
+the following:
+```clojure
+(defn my-consumer
+  {:yoltq/queue-id :some-queue}
+  [payload]
+  :work-work-work)
+
+(yq/add-consumer! #'my-consumer ; <-- will resolve to :some-queue 
+                  my-consumer)
+
+@(d/transact conn [(yq/put #'my-consumer ; <-- will resolve to :some-queue
+                           {:id "a"})])
+```
+
+The idea here is that it is simpler to jump to var definitions than going via keywords,
+which essentially refers to a var/function anyway. 
+
+#### 2022-03-29 v0.2.55 [diff](https://github.com/ivarref/yoltq/compare/v0.2.54...v0.2.55)
+Added: `unhealthy?` function which returns `true` if there are queues in error,
+or `false` otherwise.
+
+#### 2022-03-28 v0.2.54 [diff](https://github.com/ivarref/yoltq/compare/v0.2.51...v0.2.54)
+Fixed: Schedules should now be using milliseconds and not nanoseconds.
+
+#### 2022-03-28 v0.2.51 [diff](https://github.com/ivarref/yoltq/compare/v0.2.48...v0.2.51)
+* Don't OOM on migrating large amounts of data. 
+* Respect `:auto-migrate? false`.
+
+#### 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.
+```
+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.
+```
+
+* Added default functions for `:on-system-error` and `:on-system-recovery`
+  that simply logs that the system is in error (ERROR level) or has 
+  recovered (INFO level). 
+  
+* 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}]
+```
+
+#### 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 [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.
+
+## Making a new release
+
+Go to https://github.com/ivarref/yoltq/actions/workflows/release.yml and press `Run workflow`.
+
+## License
+
+Copyright © 2021-2022 Ivar Refsdal
+
+This program and the accompanying materials are made available under the
+terms of the Eclipse Public License 2.0 which is available at
+http://www.eclipse.org/legal/epl-2.0.
+
+This Source Code may also be made available under the following Secondary
+Licenses when the conditions for such availability set forth in the Eclipse
+Public License, v. 2.0 are satisfied: GNU General Public License as published by
+the Free Software Foundation, either version 2 of the License, or (at your
+option) any later version, with the GNU Classpath Exception which is available
+at https://www.gnu.org/software/classpath/license.html.
diff --git a/build.edn b/build.edn
new file mode 100644
index 0000000..3e1f016
--- /dev/null
+++ b/build.edn
@@ -0,0 +1,34 @@
+{:lib
+ com.github.ivarref/yoltq
+
+ :version
+ "0.2.{{git/commit-count}}"
+
+ :github-actions?
+ true
+
+ :scm
+ {:connection          "scm:git:git://github.com/ivarref/yoltq.git"
+  :developerConnection "scm:git:ssh://git@github.com/ivarref/yoltq.git"
+  :url                 "https://github.com/ivarref/yoltq"}
+
+ :documents
+ [{:file          "README.md"
+   :match-exactly "#### [Unreleased]"
+   :action        :append-after
+   :text          "\n#### [{{version}}] - {{now/yyyy}}-{{now/mm}}-{{now/dd}}"}
+  {:file          "README.md"
+   :match-exactly "com.github.ivarref/yoltq {:git/tag"
+   :action        :replace
+   :keep-indent?  true
+   :text
+   "com.github.ivarref/yoltq {:git/tag \"{{version}}\" :git/sha \"{{git/head-long-sha}}\"}"}
+  {:file          "README.md"
+   :match-exactly "com.github.ivarref/yoltq {:mvn/version"
+   :action        :replace
+   :keep-indent?  true
+   :text          "com.github.ivarref/yoltq {:mvn/version \"{{version}}\"}"}]
+
+ :licenses
+ [{:name "Eclipse Public License - v 2.0"
+   :url  "https://www.eclipse.org/legal/epl-2.0/"}]}
\ No newline at end of file
diff --git a/deps.edn b/deps.edn
new file mode 100644
index 0000000..f0488fa
--- /dev/null
+++ b/deps.edn
@@ -0,0 +1,40 @@
+{: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"}
+  com.datomic/peer                  {:mvn/version "1.0.7364"}}
+
+ :paths
+ ["src"]
+
+ :aliases
+ {:test
+  {:extra-paths ["test"]
+   :extra-deps  {ivarref/datomic-schema                  {:mvn/version "0.2.0"}
+                 com.taoensso/timbre                     {:mvn/version "5.2.1"}
+                 com.fzakaria/slf4j-timbre               {:mvn/version "0.3.21"}
+                 clojure-term-colors/clojure-term-colors {:mvn/version "0.1.0"}
+                 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"}}
+   :exec-fn     cognitect.test-runner.api/test
+   :jvm-opts    ["-DDISABLE_SPY=true"
+                 "-DTAOENSSO_TIMBRE_MIN_LEVEL_EDN=:error"]
+   :main-opts   ["--report" "stderr" "-m" "cognitect.test-runner"]}
+
+  :repl
+  {:extra-paths ["test"]
+   :extra-deps  {com.bhauman/rebel-readline              {:mvn/version "0.1.5"}
+                 ivarref/datomic-schema                  {:mvn/version "0.2.0"}
+                 com.taoensso/timbre                     {:mvn/version "5.2.1"}
+                 com.fzakaria/slf4j-timbre               {:mvn/version "0.3.21"}
+                 clojure-term-colors/clojure-term-colors {:mvn/version "0.1.0"}
+                 org.postgresql/postgresql               {:mvn/version "9.3-1102-jdbc41"}
+                 com.taoensso/nippy                      {:mvn/version "3.2.0"}}
+   :exec-fn     rebel-readline.tool/repl
+   :exec-args   {}
+   :main-opts   ["-m" "rebel-readline.main"]}
+
+  :build
+  {:deps       {com.github.liquidz/build.edn {:mvn/version "0.11.241"}}
+   :ns-default build-edn.main}}}
diff --git a/src/com/github/ivarref/yoltq.clj b/src/com/github/ivarref/yoltq.clj
new file mode 100644
index 0000000..8c8ca7a
--- /dev/null
+++ b/src/com/github/ivarref/yoltq.clj
@@ -0,0 +1,539 @@
+(ns com.github.ivarref.yoltq
+  (:require
+    [clojure.tools.logging :as log]
+    [com.github.ivarref.yoltq.error-poller :as errpoller]
+    [com.github.ivarref.yoltq.impl :as i]
+    [com.github.ivarref.yoltq.migrate :as migrate]
+    [com.github.ivarref.yoltq.poller :as poller]
+    [com.github.ivarref.yoltq.report-queue :as rq]
+    [com.github.ivarref.yoltq.slow-executor-detector :as slow-executor]
+    [com.github.ivarref.yoltq.utils :as u]
+    [datomic.api :as d])
+  (:import (datomic Connection)
+           (java.lang.management ManagementFactory)
+           (java.time Duration Instant ZoneOffset ZonedDateTime)
+           (java.util.concurrent BlockingQueue ExecutorService Executors ScheduledExecutorService TimeUnit)))
+
+(defonce ^:dynamic *config* (atom nil))
+(defonce threadpool (atom nil))
+(defonce ^:dynamic *running?* (atom false))
+(defonce ^:dynamic *test-mode* false)
+
+(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
+       ; (yq/add-consumer! :q (fn [payload] ...) {:max-retries 200})
+       ; 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                   9223372036854775807
+
+       ; Minimum amount of time to wait before a failed queue job is retried
+       :error-backoff-time            (Duration/ofSeconds 5)
+
+       ; Max time a queue job can execute before an error is logged
+       :max-execute-time              (Duration/ofMinutes 5)
+
+       ; Amount of time an in progress queue job can run before it is considered failed
+       ; and will be marked as such.
+       :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 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
+       ; otherwise occur if competing with the tx-report-queue listener.
+       :init-backoff-time             (Duration/ofSeconds 60)
+
+       ; If you are dealing with a flaky downstream service, you may not want
+       ; yoltq to mark itself as unhealthy on the first failure encounter with
+       ; the downstream service. Change this setting to let yoltq mark itself
+       ; as healthy even though a queue item has been failing for some time.
+       :healthy-allowed-error-time    (Duration/ofMinutes 15)
+
+       ; How frequent polling for init, error and hung jobs should be done.
+       :poll-delay                    (Duration/ofSeconds 10)
+
+       ; Specifies the number of threads available for executing queue and polling jobs.
+       ; The final thread pool will be this size + 2.
+       ;
+       ; One thread is permanently allocated for listening to the
+       ; tx-report-queue.
+       ;
+       ; Another thread is permanently allocated for checking :max-execute-time.
+       ; This means that if all executing queue jobs are stuck and the thread pool is unavailable
+       ; as such, at least an error will be logged about this. The log entry will
+       ; contain the stacktrace of the stuck threads.
+       :pool-size                     4
+
+       :capture-bindings              (if-let [s (resolve (symbol "taoensso.timbre/*context*"))]
+                                        [s]
+                                        [])
+
+       ; How often should the system be polled for failed queue jobs
+       :system-error-poll-delay       (Duration/ofMinutes 1)
+
+       ; How often should the system invoke
+       :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))
+
+
+(defn init! [{:keys [conn tx-report-queue] :as cfg}]
+  (assert (instance? Connection conn) (str "Expected :conn to be of type datomic Connection. Was: " (or (some-> conn class str) "nil")))
+  (when (some? tx-report-queue)
+    (assert (instance? BlockingQueue tx-report-queue)
+            (str "Expected :tx-report-queue to be of type java.util.concurrent.BlockingQueue")))
+  (locking threadpool
+    @(d/transact conn i/schema)
+    (let [new-cfg (swap! *config*
+                         (fn [old-conf]
+                           (-> (merge-with (fn [_ b] b)
+                                           {:running-queues            (atom #{})
+                                            :start-execute-time        (atom {})
+                                            :system-error              (atom {})
+                                            :healthy?                  (atom nil)
+                                            :slow?                     (atom nil)
+                                            :slow-thread-watcher-done? (promise)}
+                                           default-opts
+                                           (if *test-mode* old-conf (select-keys old-conf [:handlers]))
+                                           cfg)
+                               u/duration->millis)))]
+      new-cfg)))
+
+
+(defn get-queue-id
+  [queue-id-or-var]
+  (cond (and (var? queue-id-or-var)
+             (keyword? (:yoltq/queue-id (meta queue-id-or-var))))
+        (:yoltq/queue-id (meta queue-id-or-var))
+
+        (keyword? queue-id-or-var)
+        queue-id-or-var
+
+        :else
+        (throw (ex-info (str "Could not get queue-id for " queue-id-or-var) {:queue-id queue-id-or-var}))))
+
+(defn add-consumer!
+  ([queue-id f]
+   (add-consumer! queue-id f {}))
+  ([queue-id f opts]
+   (swap! *config* (fn [old-config] (assoc-in old-config [:handlers (get-queue-id queue-id)] (merge opts {:f f}))))))
+
+
+(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 (get-queue-id queue-id) payload opts))))
+
+
+(defn- do-start! []
+  (let [{:keys [poll-delay pool-size system-error-poll-delay auto-migrate? slow-thread-watcher-done?] :as cfg} @*config*]
+    (when auto-migrate?
+      (future (migrate/migrate! cfg)))
+    (let [pool (reset! threadpool (Executors/newScheduledThreadPool (+ 1 pool-size)))
+          queue-listener-ready (promise)]
+      (reset! *running?* true)
+      (.scheduleAtFixedRate ^ScheduledExecutorService pool (fn [] (poller/poll-all-queues! *running?* *config* pool)) 0 poll-delay TimeUnit/MILLISECONDS)
+      (.scheduleAtFixedRate ^ScheduledExecutorService pool (fn [] (errpoller/poll-errors *running?* *config*)) 0 system-error-poll-delay TimeUnit/MILLISECONDS)
+      (.execute ^ScheduledExecutorService pool
+                (fn []
+                  (try
+                    (log/debug "report-queue-listener starting")
+                    (rq/report-queue-listener *running?* queue-listener-ready pool *config*)
+                    (finally
+                      (log/debug "report-queue-listener exiting")
+                      (deliver queue-listener-ready :finally)))))
+      (future (try
+                (slow-executor/show-slow-threads pool *config*)
+                (finally
+                  (deliver slow-thread-watcher-done? :done))))
+      (let [q-listener-retval (deref queue-listener-ready 30000 :timeout)]
+        (cond (= :timeout q-listener-retval)
+              (do
+                (log/error "Timed out waiting for report-queue-listener to start")
+                (throw (IllegalStateException. "Timed out waiting for report-queue-listener to start")))
+
+              (= :finally q-listener-retval)
+              (do
+                (log/error "report-queue-listener did not start")
+                (throw (IllegalStateException. "report-queue-listener did not start")))
+
+              (= :ready q-listener-retval)
+              (do
+                (log/debug "report-queue-listener is ready"))
+
+              :else
+              (do
+                (log/error (str "Unexpected queue-listener-retval: " (pr-str q-listener-retval)))
+                (throw (IllegalStateException. (str "Unexpected queue-listener-retval: " (pr-str q-listener-retval))))))))))
+
+
+(defn start! []
+  (locking threadpool
+    (cond (true? *test-mode*)
+          (log/info "test mode enabled, doing nothing for start!")
+
+          (true? @*running?*)
+          nil
+
+          (false? @*running?*)
+          (do-start!))))
+
+
+(defn stop! []
+  (locking threadpool
+    (cond (true? *test-mode*)
+          (log/info "test mode enabled, doing nothing for stop!")
+
+          (false? @*running?*)
+          nil
+
+          (true? @*running?*)
+          (do
+            (reset! *running?* false)
+            (when-let [^ExecutorService tp @threadpool]
+              (log/debug "shutting down threadpool")
+              (.shutdown tp)
+              (while (not (.awaitTermination tp 1 TimeUnit/SECONDS))
+                (log/trace "waiting for threadpool to stop"))
+              (log/debug "stopped!")
+              (reset! threadpool nil))
+            (when-let [wait-slow-threads (some->> *config* deref :slow-thread-watcher-done?)]
+              (log/debug "waiting for slow-thread-watcher to stop ...")
+              @wait-slow-threads
+              (log/debug "waiting for slow-thread-watcher to stop ... OK"))))))
+
+
+(defn healthy? []
+  (cond
+    (< (.toMinutes (Duration/ofMillis (.getUptime (ManagementFactory/getRuntimeMXBean)))) 10)
+    true
+
+    (false? (some->> @*config*
+                     :healthy?
+                     (deref)))
+    false
+
+    (true? (some->> @*config*
+                    :slow?
+                    (deref)))
+    false
+
+    :else
+    true))
+
+(defn unhealthy?
+  "Returns `true` if there are queues in error or a thread is slow, and the application has been up for over 10 minutes, otherwise `false`."
+  []
+  (false? (healthy?)))
+
+(defn queue-stats []
+  (let [{:keys [conn]} @*config*
+        db (d/db conn)]
+    (->> (d/q '[:find ?e ?qname ?status
+                :in $
+                :where
+                [?e :com.github.ivarref.yoltq/queue-name ?qname]
+                [?e :com.github.ivarref.yoltq/status ?status]]
+              db)
+         (mapv (partial zipmap [:e :qname :status]))
+         (mapv #(select-keys % [:qname :status]))
+         (mapv (fn [qitem] {qitem 1}))
+         (reduce (partial merge-with +) {})
+         (mapv (fn [[{:keys [qname status]} v]]
+                 (array-map
+                   :qname qname
+                   :status status
+                   :count v)))
+         (sort-by (juxt :qname :status))
+         (vec))))
+
+(defn job-group-progress [queue-name job-group]
+  (let [{:keys [conn]} @*config*
+        db (d/db conn)]
+    (->> (d/q '[:find ?e ?qname ?job-group ?status
+                :keys :e :qname :job-group :status
+                :in $ ?qname ?job-group
+                :where
+                [?e :com.github.ivarref.yoltq/queue-name ?qname]
+                [?e :com.github.ivarref.yoltq/job-group ?job-group]
+                [?e :com.github.ivarref.yoltq/status ?status]]
+              db queue-name job-group)
+         (mapv #(select-keys % [:qname :job-group :status]))
+         (mapv (fn [qitem] {qitem 1}))
+         (reduce (partial merge-with +) {})
+         (mapv (fn [[{:keys [qname job-group status]} v]]
+                 (array-map
+                  :qname qname
+                  :job-group job-group
+                  :status status
+                  :count v)))
+         (sort-by (juxt :qname :job-group :status))
+         (vec))))
+
+(defn get-errors [qname]
+  (let [{:keys [conn]} @*config*
+        db (d/db conn)]
+    (->> (d/q '[:find [?id ...]
+                :in $ ?qname ?status
+                :where
+                [?e :com.github.ivarref.yoltq/queue-name ?qname]
+                [?e :com.github.ivarref.yoltq/status ?status]
+                [?e :com.github.ivarref.yoltq/id ?id]]
+              db
+              qname
+              :error)
+         (mapv (partial u/get-queue-item db)))))
+
+(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] Long/MAX_VALUE)]
+    (poller/poll-once! cfg qname :error)))
+
+(defn retry-stats
+  "Gather retry statistics.
+
+  Optional keyword arguments:
+  * :age-days —  last number of days to look at data from. Defaults to 30.
+  * :queue-name — only gather statistics for this queue name. Defaults to nil, meaning all queues.
+
+  Example return value:
+  {:queue-a {:ok 100, :retries 2, :retry-percentage 2.0}
+   :queue-b {:ok 100, :retries 75, :retry-percentage 75.0}}
+
+  From the example value above, we can see that :queue-b fails at a much higher rate than :queue-a.
+  Assuming that the queue consumers are correctly implemented, this means that the service representing :queue-b
+  is much more unstable than the one representing :queue-a. This again implies
+  that you will probably want to fix the downstream service of :queue-b, if that is possible.
+  "
+  [{:keys [age-days queue-name now db]
+    :or   {age-days 30
+           now      (ZonedDateTime/now ZoneOffset/UTC)}}]
+  (let [{:keys [conn]} @*config*
+        db (or db (d/db conn))]
+    (->> (d/query {:query {:find  '[?qname ?status ?tries ?init-time]
+                           :in    (into '[$] (when queue-name '[?qname]))
+                           :where '[[?e :com.github.ivarref.yoltq/queue-name ?qname]
+                                    [?e :com.github.ivarref.yoltq/status ?status]
+                                    [?e :com.github.ivarref.yoltq/tries ?tries]
+                                    [?e :com.github.ivarref.yoltq/init-time ?init-time]]}
+                   :args  (remove nil? [db queue-name])})
+         (mapv (partial zipmap [:qname :status :tries :init-time]))
+         (mapv #(update % :init-time (fn [init-time] (.atZone (Instant/ofEpochMilli init-time) ZoneOffset/UTC))))
+         (mapv #(assoc % :age-days (.toDays (Duration/between (:init-time %) now))))
+         (filter #(<= (:age-days %) age-days))
+         (group-by :qname)
+         (mapv (fn [[q values]]
+                 {q (let [{:keys [ok retries] :as m} (->> values
+                                                          (mapv (fn [{:keys [tries status]}]
+                                                                  (condp = status
+                                                                    u/status-init {}
+                                                                    u/status-processing {:processing 1 :retries (dec tries)}
+                                                                    u/status-done {:ok 1 :retries (dec tries)}
+                                                                    u/status-error {:error 1 :retries (dec tries)})))
+                                                          (reduce (partial merge-with +) {}))]
+                      (into (sorted-map) (merge m
+                                                (when (pos-int? ok)
+                                                  {:retry-percentage (double (* 100 (/ retries ok)))}))))}))
+         (into (sorted-map)))))
+
+(defn- percentile [n values]
+  (let [idx (int (Math/floor (* (count values) (/ n 100))))]
+    (nth values idx)))
+
+(defn processing-time-stats
+  "Gather processing time statistics. Default unit is seconds.
+
+  Optional keyword arguments:
+  * :age-days —  last number of days to look at data from. Defaults to 30.
+                 Use nil to have no limit.
+
+  * :queue-name — only gather statistics for this queue name. Defaults to nil, meaning all queues.
+
+  * :duration->long - Specify what unit should be used for values.
+                      Must take a java.time.Duration as input and return a long.
+
+                      Defaults to (fn [duration] (.toSeconds duration).
+                      I.e. the default unit is seconds.
+
+  Example return value:
+  {:queue-a {:avg 1
+             :max 10
+             :min 0
+             :p50 ...
+             :p90 ...
+             :p95 ...
+             :p99 ...}}"
+  [{:keys [age-days queue-name now db duration->long]
+    :or   {age-days       30
+           now            (ZonedDateTime/now ZoneOffset/UTC)
+           duration->long (fn [duration] (.toSeconds ^Duration duration))}}]
+  (let [{:keys [conn]} @*config*
+        db (or db (d/db conn))
+        ->zdt #(.atZone (Instant/ofEpochMilli %) ZoneOffset/UTC)]
+    (->> (d/query {:query {:find  '[?qname ?status ?init-time ?done-time]
+                           :in    (into '[$ ?status] (when queue-name '[?qname]))
+                           :where '[[?e :com.github.ivarref.yoltq/queue-name ?qname]
+                                    [?e :com.github.ivarref.yoltq/status ?status]
+                                    [?e :com.github.ivarref.yoltq/init-time ?init-time]
+                                    [?e :com.github.ivarref.yoltq/done-time ?done-time]]}
+                   :args  (vec (remove nil? [db u/status-done queue-name]))})
+         (mapv (partial zipmap [:qname :status :init-time :done-time]))
+         (mapv #(update % :init-time ->zdt))
+         (mapv #(update % :done-time ->zdt))
+         (mapv #(assoc % :age-days (.toDays (Duration/between (:init-time %) now))))
+         (mapv #(assoc % :spent-time (duration->long (Duration/between (:init-time %) (:done-time %)))))
+         (filter #(or (nil? age-days) (<= (:age-days %) age-days)))
+         (group-by :qname)
+         (mapv (fn [[q values]]
+                 (let [values (vec (sort (mapv :spent-time values)))]
+                   {q (sorted-map
+                        :max (apply max values)
+                        :avg (int (Math/floor (/ (reduce + 0 values) (count values))))
+                        :p50 (percentile 50 values)
+                        :p90 (percentile 90 values)
+                        :p95 (percentile 95 values)
+                        :p99 (percentile 99 values)
+                        :min (apply min values))})))
+         (into (sorted-map)))))
+
+(defn get-tx-report-queue-multicast!
+  "Multicast the datomic.api/tx-report-queue to different consumers.
+   A multicaster thread is started on demand per connection. `conn` and `id` identifies the consumer.
+   Repeated calls using the same `conn` and `id` returns the same queue.
+
+   The optional third parameter, `send-end-token?`, if set to `true`, instructs the multicaster thread
+   to send `:end` if the queue is stopped.
+   The default value for `send-end-token?` is `false`.
+
+   A queue may be stopped using `stop-multicaster-id!`, `stop-multicaster!` or `stop-all-multicasters!`.
+
+   Returns a `java.util.concurrent.BlockingQueue` like `datomic.api/tx-report-queue`."
+  ([conn id]
+   (get-tx-report-queue-multicast! conn id false))
+  ([conn id send-end-token?]
+   (assert (instance? Connection conn))
+   (assert (boolean? send-end-token?))
+   (rq/get-tx-report-queue-multicast! conn id send-end-token?)))
+
+(defn stop-multicast-consumer-id!
+  "Stop forwarding reports from datomic.api/tx-report-queue to the queue identified by `conn` and `id`.
+  If this is the last report destination for the given `conn`, the multicaster thread will exit.
+  Repeated calls are no-op.
+
+  The multicaster thread will send `:end` if `send-end-token?` was `true` when `get-tx-report-queue-multicast!`
+  was called.
+
+  Returns `true` if the queue was stopped.
+  Return `false` if the queue does not exist."
+  [conn id]
+  (assert (instance? Connection conn))
+  (rq/stop-multicast-consumer-id! conn id))
+
+(defn stop-multicaster!
+  "Stop forwarding reports from datomic.api/tx-report-queue to any queues belonging to `conn`.
+  The multicaster thread will exit.
+  Repeated calls are no-op.
+
+  The multicaster thread will send `:end` if `send-end-token?` was `true` when `get-tx-report-queue-multicast!`
+  was called.
+
+  Returns `true` if any queue belonging to `conn` was stopped.
+  Returns `false` is `conn` did not have any associated queues."
+  [conn]
+  (assert (instance? Connection conn))
+  (rq/stop-multicaster! conn))
+
+(defn stop-all-multicasters!
+  "Stop forwarding all reports from datomic.api/tx-report-queue for any `conn`.
+  All multicaster threads will exit.
+  Repeated calls are no-op.
+
+  The multicaster thread will send `:end` if `send-end-token?` was `true` when `get-tx-report-queue-multicast!`
+  was called.
+
+  Returns `true` if any queue was stopped.
+  Returns `false` if no queues existed."
+  []
+  (rq/stop-all-multicasters!))
+
+(comment
+  (do
+    (require 'com.github.ivarref.yoltq.log-init)
+    (com.github.ivarref.yoltq.log-init/init-logging!
+      [[#{"datomic.*" "com.datomic.*" "org.apache.*"} :warn]
+       [#{"ivarref.yoltq.report-queue"} :info]
+       [#{"ivarref.yoltq.poller"} :info]
+       [#{"com.github.ivarref.yoltq"} :debug]
+       ;[#{"ivarref.yoltq*"} :info]
+       [#{"*"} :info]])
+    (stop!)
+    (future (let [received (atom [])
+                  uri (str "datomic:mem://demo")]
+              (d/delete-database uri)
+              (d/create-database uri)
+              (let [conn (d/connect uri)
+                    started-consuming? (promise)
+                    n 1]
+                (init! {:conn                         conn
+                        :error-backoff-time           (Duration/ofSeconds 1)
+                        :poll-delay                   (Duration/ofSeconds 1)
+                        :max-execute-time             (Duration/ofSeconds 3)
+                        :slow-thread-show-stacktrace? false})
+                (add-consumer! :q (fn [_]
+                                    (deliver started-consuming? true)
+                                    (log/info "sleeping...")
+                                    (Thread/sleep (.toMillis (Duration/ofSeconds 60)))
+                                    (log/info "done sleeping")))
+                (start!)
+                @(d/transact conn [(put :q {:work 123})])
+                @started-consuming?
+                (stop!)
+                nil)))))
+
+(comment
+  (do
+    (require 'com.github.ivarref.yoltq.log-init)
+    (com.github.ivarref.yoltq.log-init/init-logging!
+      [[#{"datomic.*" "com.datomic.*" "org.apache.*"} :warn]
+       [#{"com.github.ivarref.yoltq.report-queue"} :debug]
+       [#{"com.github.ivarref.yoltq.poller"} :info]
+       [#{"com.github.ivarref.yoltq.migrate"} :warn]
+       [#{"com.github.ivarref.yoltq"} :debug]
+       ;[#{"ivarref.yoltq*"} :info]
+       [#{"*"} :info]])
+    (stop!)
+    (let [received (atom [])
+          uri (str "datomic:mem://demo")]
+      (d/delete-database uri)
+      (d/create-database uri)
+      (let [conn (d/connect uri)
+            started-consuming? (promise)
+            n 1]
+        (init! {:conn                         conn
+                :tx-report-queue              (get-tx-report-queue-multicast! conn :yoltq true)
+                :slow-thread-show-stacktrace? false})
+        (add-consumer! :q (fn [_]
+                            (deliver started-consuming? true)))
+        (log/info "begin start! ...")
+        (start!)
+        (log/info "begin start! ... Done")
+        (Thread/sleep 2000)
+        (log/info "*******************************************")
+        @(d/transact conn [(put :q {:work 123})])
+        @started-consuming?
+        (stop-multicaster! conn)
+        (log/info "*******************************************")
+        (stop!)
+        (log/info "stop! done")
+        nil))))
\ No newline at end of file
diff --git a/src/com/github/ivarref/yoltq/error_poller.clj b/src/com/github/ivarref/yoltq/error_poller.clj
new file mode 100644
index 0000000..dffff28
--- /dev/null
+++ b/src/com/github/ivarref/yoltq/error_poller.clj
@@ -0,0 +1,124 @@
+(ns com.github.ivarref.yoltq.error-poller
+  (:require [clojure.tools.logging :as log]
+            [com.github.ivarref.yoltq.ext-sys :as ext]
+            [com.github.ivarref.yoltq.utils :as u]
+            [datomic.api :as d]))
+
+
+(defn get-state [v]
+  (case v
+    [:error :none] :recovery
+    [:error :some] :error
+    [:error :all] :error
+    [:recovery :none] :recovery
+    [:recovery :some] :recovery
+    [:recovery :all] :error
+    nil))
+
+
+(defn handle-error-count [{:keys [errors last-notify state]
+                           :or   {errors      []
+                                  last-notify 0
+                                  state       :recovery}}
+                          {:keys [system-error-min-count system-error-callback-backoff]
+                           :or   {system-error-min-count 3}}
+                          now-ms
+                          error-count]
+  (let [new-errors (->> (conj errors error-count)
+                        (take-last system-error-min-count)
+                        (vec))
+        classify (fn [coll]
+                   (cond
+                     (not= system-error-min-count (count coll))
+                     :missing
+
+                     (every? pos-int? coll)
+                     :all
+
+                     (every? zero? coll)
+                     :none
+
+                     :else
+                     :some))
+        old-state state]
+    (merge
+      {:errors      new-errors
+       :last-notify last-notify}
+      (when-let [new-state (get-state [old-state (classify new-errors)])]
+        (merge
+          {:state new-state}
+          (when (and (= old-state :recovery)
+                     (= new-state :error))
+            {:run-callback :error
+             :last-notify  now-ms})
+
+          (when (and (= new-state :error)
+                     (= old-state :error)
+                     (> now-ms
+                        (+ last-notify system-error-callback-backoff)))
+            {:run-callback :error
+             :last-notify  now-ms})
+
+          (when (and (= new-state :recovery)
+                     (= old-state :error))
+            {:run-callback :recovery}))))))
+
+
+(defn do-poll-errors [{:keys [conn
+                              system-error
+                              on-system-error
+                              on-system-recovery
+                              healthy?
+                              healthy-allowed-error-time]
+                       :or   {on-system-error    (fn []
+                                                   (log/error "There are yoltq queues which have errors")
+                                                   nil)
+                              on-system-recovery (fn []
+                                                   (log/info "Yoltq recovered"))}
+                       :as   config}
+                      now-ms]
+  (assert (some? conn) "expected :conn to be present")
+  (assert (some? system-error) "expected :system-error to be present")
+  (assert (nat-int? healthy-allowed-error-time) "expected :healthy-allowed-error-time to be present")
+  (let [max-init-time (- now-ms healthy-allowed-error-time)
+        error-count (or (d/q '[:find (count ?e) .
+                               :in $ ?status ?max-init-time
+                               :where
+                               [?e :com.github.ivarref.yoltq/status ?status]
+                               [?e :com.github.ivarref.yoltq/init-time ?init-time]
+                               [(<= ?init-time ?max-init-time)]]
+                             (d/db conn)
+                             u/status-error
+                             max-init-time)
+                        0)]
+    (if (pos-int? error-count)
+      (do
+        (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 now-ms error-count)]
+      (when run-callback
+        (cond (= run-callback :error)
+              (on-system-error)
+
+              (= run-callback :recovery)
+              (on-system-recovery)
+
+              :else
+              (log/error "unhandled callback-type" run-callback))
+        (log/debug "run-callback is" run-callback))
+      error-count)))
+
+
+(defn poll-errors [running? config-atom]
+  (try
+    (when @running?
+      (do-poll-errors @config-atom (ext/now-ms)))
+    (catch Throwable t
+      (log/error t "unexpected error in poll-errors:" (ex-message t))
+      nil)))
+
+
+(comment
+  (do-poll-errors @com.github.ivarref.yoltq/*config* (ext/now-ms)))
+
diff --git a/src/com/github/ivarref/yoltq/ext_sys.clj b/src/com/github/ivarref/yoltq/ext_sys.clj
new file mode 100644
index 0000000..692b934
--- /dev/null
+++ b/src/com/github/ivarref/yoltq/ext_sys.clj
@@ -0,0 +1,27 @@
+(ns com.github.ivarref.yoltq.ext-sys
+  (:require [datomic.api :as d])
+  (:refer-clojure :exclude [random-uuid])
+  (:import (java.util UUID)))
+
+
+(def ^:dynamic *now-ms-atom* nil)
+(def ^:dynamic *squuid-atom* nil)
+(def ^:dynamic *random-atom* nil)
+
+
+(defn now-ms []
+  (if *now-ms-atom*
+    @*now-ms-atom*
+    (System/currentTimeMillis)))
+
+
+(defn squuid []
+  (if *squuid-atom*
+    (UUID/fromString (str "00000000-0000-0000-0000-" (format "%012d" (swap! *squuid-atom* inc))))
+    (d/squuid)))
+
+
+(defn random-uuid []
+  (if *random-atom*
+    (UUID/fromString (str "00000000-0000-0000-0000-" (format "%012d" (swap! *random-atom* inc))))
+    (UUID/randomUUID)))
diff --git a/src/com/github/ivarref/yoltq/impl.clj b/src/com/github/ivarref/yoltq/impl.clj
new file mode 100644
index 0000000..ffb1ad8
--- /dev/null
+++ b/src/com/github/ivarref/yoltq/impl.clj
@@ -0,0 +1,248 @@
+(ns com.github.ivarref.yoltq.impl
+  (:require [clojure.edn :as edn]
+            [clojure.string :as str]
+            [clojure.tools.logging :as log]
+            [com.github.ivarref.double-trouble :as dt]
+            [com.github.ivarref.yoltq.ext-sys :as ext]
+            [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}
+   #: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/job-group, :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}
+   #:db{:ident :com.github.ivarref.yoltq/lock, :cardinality :db.cardinality/one, :valueType :db.type/uuid, :noHistory true}
+   #: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/version, :cardinality :db.cardinality/one, :valueType :db.type/string, :index true}])
+
+(defn pr-str-inner [x]
+  (binding [*print-dup* false
+            *print-meta* false
+            *print-readably* true
+            *print-length* nil
+            *print-level* nil
+            *print-namespace-maps* false]
+    (pr-str x)))
+
+(defn pr-str-safe [what x]
+  (try
+    (if (= x (edn/read-string (pr-str-inner x)))
+      (pr-str-inner x)
+      (throw (ex-info (str "Could not read-string " what) {:input x})))
+    (catch Exception e
+      (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 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
+                 (when-let [deps (depends-on payload)]
+                   {:depends-on deps})
+                 (or opts {}))
+          str-bindings (->> (reduce (fn [o k]
+                                      (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})))
+          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)
+      (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])})
+          (when-let [job-group (:job-group opts)]
+            {:com.github.ivarref.yoltq/job-group job-group}))))
+    (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]}
+                           q-item]
+  (let [db (d/db conn)]
+    (when-let [{:com.github.ivarref.yoltq/keys [opts]} (u/get-queue-item db (:id q-item))]
+      (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]))
+          (log/info "queue item" (str (:id q-item)) "is waiting on" depends-on)
+          {: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}]
+  (when queue-item-info
+    (try
+      (cond to-error?
+            (log/logp hung-log-level "queue-item" (str id) "was hung and retried too many times. Giving up!")
+
+            was-hung?
+            (log/logp hung-log-level "queue-item" (str id) "was hung, retrying ...")
+
+            :else
+            nil)
+      (let [start-time (System/nanoTime)
+            {:keys [db-after]} @(d/transact conn tx)
+            _ (when tx-spent-time! (tx-spent-time! (- (System/nanoTime) start-time)))
+            {:com.github.ivarref.yoltq/keys [status] :as q-item} (u/get-queue-item db-after id)]
+        (log/debug "queue item" (str id) "for queue" queue-name "now has status" status)
+        q-item)
+      (catch Throwable t
+        (let [{:db/keys [error] :as m} (u/db-error-map t)]
+          (cond
+            (= :db.error/cas-failed error)
+            (do
+              (log/info "take! :db.error/cas-failed for queue item" (str id) "and attribute" (:a m))
+              (when cas-failures
+                (swap! cas-failures inc))
+              nil)
+
+            :else
+            (do
+              (log/error t "Unexpected failure for queue item" (str id) ":" (ex-message t))
+              nil)))))))
+
+
+(defn mark-status! [{:keys [conn tx-spent-time!]}
+                    {:com.github.ivarref.yoltq/keys [id lock tries]}
+                    new-status]
+  (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 new-status]
+              (if (= new-status u/status-done)
+                {: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)))
+      (u/get-queue-item db-after id))
+    (catch Throwable t
+      (log/error t "unexpected error in mark-status!: " (ex-message t))
+      nil)))
+
+
+(defn fmt [id queue-name new-status tries spent-ns]
+  (str/join " " ["queue-item" (str id)
+                 "for queue" queue-name
+                 "now has status" new-status
+                 "after" tries (if (= 1 tries)
+                                 "try"
+                                 "tries")
+                 "in" (format "%.1f" (double (/ spent-ns 1e6))) "ms"]))
+
+
+(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 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
+              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 [payload (decode (or payload payload-bytes))
+                        v (f payload)]
+                    {:retval v})
+                  (catch Throwable t
+                    {:exception t})
+                  (finally
+                    (swap! start-execute-time dissoc (Thread/currentThread))))
+                {:db/keys [error] :as m} (u/db-error-map exception)]
+            (cond
+              (and (some? exception)
+                   allow-cas-failure?
+                   (= :db.error/cas-failed error)
+                   (or (true? allow-cas-failure?)
+                       (allow-cas-failure? (:a m))))
+              (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-ms) init-time)))
+                  (assoc q-item :retval retval :success? true :allow-cas-failure? true)))
+
+              (some? exception)
+              (when-let [q-item (mark-status-fn! cfg queue-item u/status-error)]
+                (let [{:com.github.ivarref.yoltq/keys [init-time error-time tries]} q-item
+                      level (if (>= tries 3) :error :warn)]
+                  (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-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-ms) init-time)))
+                  (assoc q-item :retval retval :success? true))))))
+        (do
+          (log/error "no handler for queue" queue-name)
+          nil)))))
diff --git a/src/com/github/ivarref/yoltq/migrate.clj b/src/com/github/ivarref/yoltq/migrate.clj
new file mode 100644
index 0000000..c97f679
--- /dev/null
+++ b/src/com/github/ivarref/yoltq/migrate.clj
@@ -0,0 +1,61 @@
+(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 (vec (take 10 (conj tx-vec tx))))
+          tx))
+      (do
+        (log/info "No items left to migrate")
+        tx-vec))))
+
+
+(defn migrate! [cfg]
+  (to->v2 cfg))
+
+(comment
+  (migrate! @com.github.ivarref.yoltq/*config*))
diff --git a/src/com/github/ivarref/yoltq/poller.clj b/src/com/github/ivarref/yoltq/poller.clj
new file mode 100644
index 0000000..9cf81c7
--- /dev/null
+++ b/src/com/github/ivarref/yoltq/poller.clj
@@ -0,0 +1,64 @@
+(ns com.github.ivarref.yoltq.poller
+  (:require [com.github.ivarref.yoltq.utils :as u]
+            [com.github.ivarref.yoltq.impl :as i]
+            [clojure.tools.logging :as log]))
+
+
+(defn poll-once! [cfg q status]
+  (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 {})
+      (if (i/depends-on-waiting? cfg item)
+        nil
+        (some->> item
+                 (i/take! cfg)
+                 (i/execute! cfg))))))
+
+
+(defn poll-queue! [running?
+                   {:keys [running-queues] :as cfg}
+                   [queue-name status :as q]]
+  (try
+    (let [[old _] (swap-vals! running-queues conj q)]
+      (if-not (contains? old q)
+        (try
+          (log/debug "polling queue" queue-name "for status" status)
+          (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-ms (- (u/now-ms) start-time)]
+              (log/trace "done polling queue" q "in" spent-ms "ms"))
+            last-res)
+          (finally
+            (swap! running-queues disj q)))
+        (log/debug "queue" q "is already being polled, doing nothing...")))
+    (catch Throwable t
+      (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
+    (when @running?
+      (let [{:keys [handlers]} @config-atom]
+        (doseq [q (shuffle (vec (for [q-name (keys handlers)
+                                      status [:init :error :hung]]
+                                  [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)))))
diff --git a/src/com/github/ivarref/yoltq/report_queue.clj b/src/com/github/ivarref/yoltq/report_queue.clj
new file mode 100644
index 0000000..f83e3ba
--- /dev/null
+++ b/src/com/github/ivarref/yoltq/report_queue.clj
@@ -0,0 +1,459 @@
+(ns com.github.ivarref.yoltq.report-queue
+  (:require [com.github.ivarref.yoltq.utils :as u]
+            [com.github.ivarref.yoltq.impl :as i]
+            [datomic.api :as d]
+            [clojure.tools.logging :as log])
+  (:import (datomic Connection Datom)
+           (java.util.concurrent LinkedBlockingQueue ScheduledExecutorService BlockingQueue TimeUnit)))
+
+; Private API, subject to change!
+
+(defn process-poll-result! [cfg id-ident poll-result consumer]
+  (let [{:keys [tx-data db-after]} poll-result]
+    (when-let [new-ids (->> tx-data
+                            (filter (fn [^Datom datom] (and
+                                                         (= (.a datom) id-ident)
+                                                         (.added datom))))
+                            (mapv (fn [^Datom datom] (.v datom)))
+                            (into [])
+                            (not-empty))]
+      (doseq [id new-ids]
+        (consumer (fn []
+                    (try
+                      (let [{:com.github.ivarref.yoltq/keys [lock id status queue-name bindings]} (u/get-queue-item db-after id)]
+                        (with-bindings (or bindings {})
+                          (if (i/depends-on-waiting? cfg {:id id})
+                            nil
+                            (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!")))))))))
+
+(defn report-queue-listener [running?
+                             ready?
+                             ^ScheduledExecutorService pool
+                             config-atom]
+  (let [cfg @config-atom
+        conn (:conn cfg)
+        tx-report-queue-given (contains? cfg :tx-report-queue)
+        ^BlockingQueue q (if tx-report-queue-given
+                           (get cfg :tx-report-queue)
+                           (d/tx-report-queue conn))
+        id-ident (d/q '[:find ?e .
+                        :where [?e :db/ident :com.github.ivarref.yoltq/id]]
+                      (d/db conn))]
+    (assert (instance? BlockingQueue q))
+    (log/debug "tx-report-queue-given:" tx-report-queue-given)
+    (try
+      (let [running-local? (atom true)]
+        (while (and @running? @running-local?)
+          (when-let [poll-result (.poll ^BlockingQueue q 1 TimeUnit/SECONDS)]
+            (if (= poll-result :end)
+              (do
+                (log/debug "report-queue-listener received :end token. Exiting")
+                (reset! running-local? false))
+              ;(log/warn "yoltq report-queue-listener received :end token. If the rest of the system is kept running, it will result in a partially broken system."))
+              (process-poll-result! @config-atom
+                                    id-ident
+                                    poll-result
+                                    (fn [f]
+                                      (when @running?
+                                        (.execute ^ScheduledExecutorService pool f))))))
+          (deliver ready? :ready)))
+      (catch Throwable t
+        (log/error t "Unexpected error in report-queue-listener:" (.getMessage t)))
+      (finally
+        (if tx-report-queue-given
+          (log/debug "Remove tx-report-queue handled elsewhere")
+          (do
+            (log/debug "Remove tx-report-queue")
+            (d/remove-tx-report-queue conn)))))))
+
+; https://stackoverflow.com/a/14488425
+(defn- dissoc-in
+  "Dissociates an entry from a nested associative structure returning a new
+  nested structure. keys is a sequence of keys. Any empty maps that result
+  will not be present in the new structure."
+  [m [k & ks :as keys]]
+  (if ks
+    (if-let [nextmap (get m k)]
+      (let [newmap (dissoc-in nextmap ks)]
+        (if (seq newmap)
+          (assoc m k newmap)
+          (dissoc m k)))
+      m)
+    (dissoc m k)))
+
+(defn- queues-to-shutdown [old-state new-state]
+  (assert (map? old-state))
+  (assert (map? new-state))
+  (doseq [x (vals new-state)]
+    (assert (vector? x)))
+  (doseq [x (vals old-state)]
+    (assert (vector? x)))
+  (let [new-qs (into #{} (mapv second (vals new-state)))]
+    (reduce
+      (fn [o [send-end-token? old-q]]
+        ;(assert (boolean? send-end-token?))
+        ;(assert (instance? BlockingQueue old-q))
+        (if (contains? new-qs old-q)
+          o
+          (conj o [send-end-token? old-q])))
+      []
+      (vals old-state))))
+
+(comment
+  (queues-to-shutdown {:a [true 999] :b [false 777]}
+                      {:a [true 123] :b [true 777]}))
+
+(defn- multicast-once [conn work-item old-state new-state]
+  (assert (map? old-state))
+  (assert (map? new-state))
+  (doseq [[send-end-token? q-to-shutdown] (queues-to-shutdown old-state new-state)]
+    (if send-end-token?
+      (do
+        #_(log/debug "offering :end token")
+        (if (.offer ^BlockingQueue q-to-shutdown :end 1 TimeUnit/MICROSECONDS)
+          (log/debug "Multicaster sent :end token")
+          (log/debug "Multicaster failed to send :end token")))
+      (do
+        (log/debug "Multicaster not sending :end token"))))
+  (when (seq new-state)
+    (if (some? work-item)
+      (reduce-kv
+        (fn [m id [send-end-token? q]]
+          (let [ok-offer (.offer ^BlockingQueue q work-item 1 TimeUnit/MICROSECONDS)]
+            (if (true? ok-offer)
+              (assoc m id [send-end-token? q])
+              (log/error "Multicaster failed to offer item for connection" conn "and queue id" id))))
+        {}
+        new-state)
+      new-state)))
+
+(defonce ^:private multicast-state-lock (Object.))
+(defonce ^:private consumer-state-lock (Object.))
+(defonce ^:private multicast-state (atom {}))
+(defonce ^:private thread-count (atom 0))
+
+(defn- multicaster-loop [init-state conn ready?]
+  (assert (instance? Connection conn))
+  (let [input-queue (d/tx-report-queue conn)]
+    (deliver ready? true)
+    (loop [old-state init-state]
+      (let [work-item (.poll ^BlockingQueue input-queue 16 TimeUnit/MILLISECONDS)
+            new-state (locking multicast-state-lock
+                        ; writer to `multicast-state` must be protected by `multicast-state-lock`
+                        ; it should block minimally / spend minimum amount of time
+                        (swap! multicast-state (fn [old-state] (update-in old-state [:iter-count conn] (fnil inc 0))))
+                        (if-let [new-state (multicast-once conn work-item old-state (get-in @multicast-state [:queues conn] {}))]
+                          new-state
+                          (do (swap! multicast-state (fn [old-state] (dissoc-in old-state [:queues conn])))
+                              (swap! multicast-state (fn [old-state] (update-in old-state [:thread-count conn] dec)))
+                              (d/remove-tx-report-queue conn)
+                              (log/debug "Multicaster removed tx-report-queue for conn" conn)
+                              nil)))]
+        (if new-state
+          (recur new-state)
+          nil)))))
+
+(defn- start-multicaster! [conn]
+  (assert (instance? Connection conn))
+  (let [ready? (promise)]
+    (future
+      (log/debug "Multicaster starting for conn" conn)
+      (try
+        (swap! thread-count inc)
+        (let [new-state (swap! multicast-state (fn [old-state] (update-in old-state [:thread-count conn] (fnil inc 0))))]
+          (assert (= 1 (get-in new-state [:thread-count conn])))
+          ; "parent" thread holds `multicast-state-lock` and
+          ; waits for `ready?` promise, so effectively this new thread also holds
+          ; the lock until `ready?` is delivered. That is: it is safe
+          ; for this thread to modify multicast-state regardless of what other threads are doing
+          (multicaster-loop (get-in new-state [:queues conn]) conn ready?))
+        (catch Throwable t
+          (log/error t "Unexpected error in multicaster:" (.getMessage t))
+          (log/error "Multicaster exiting for conn"))
+        (finally
+          (swap! thread-count dec)
+          (log/debug "Multicaster exiting for conn" conn))))
+    (when (= :timeout (deref ready? 30000 :timeout))
+      (throw (RuntimeException. "Timed out waiting for multicaster to start")))))
+
+(defn- wait-multicast-thread-step
+  [conn state]
+  ; `get-tx-report-queue-multicast!` should return only when the multicaster thread
+  ; has picked up the new queue.
+  ;
+  ; Otherwise the following could happen:
+  ; 1. multicast thread is sleeping
+  ; 2: user-thread calls get-tx-report-queue-multicast! with `send-end-token?` `true`
+  ; 3: user-thread (or somebody else) calls `stop-multicaster`.
+  ;    The multicast-state atom is now identical as it was in step 1.
+  ; ,  Step 2 and 3 happened while the multicast thread was sleeping.
+  ; 4: The multicast thread is scheduled and does _not_ detect any state change.
+  ;    Therefore the multicast thread does _not_ send out an :end token as one would expect.
+  ;
+  ; The new queue is written to memory at this point. No other thread can remove it because
+  ; we are still, and have been during the modification of multicast-state, holding consumer-state-lock.
+  ; This means that the multicast thread cannot exit at this point. Also, because we hold the consumer-state-lock,
+  ; we can be sure that no other thread changes or has changed the state.
+  ;
+  ; Once [:iter-count conn] has changed, we know that the multicaster thread
+  ; has seen the new queue. This means that we can be sure that the queue
+  ; will receive the `:end` token if the queue is stopped.
+  (let [start-ms (System/currentTimeMillis)
+        iter-count (get-in state [:iter-count conn] -1)]
+    (loop [spin-count 0]
+      (if (not= iter-count (locking multicast-state-lock
+                             (get-in @multicast-state [:iter-count conn] -1)))
+        nil
+        (let [spent-ms (- (System/currentTimeMillis) start-ms)]
+          (if (> spent-ms 30000)
+            (throw (RuntimeException. "Timed out waiting for multicaster thread"))
+            (do
+              (Thread/sleep 16)
+              (recur (inc spin-count)))))))))
+
+(defn get-tx-report-queue-multicast!
+  "Multicast the datomic.api/tx-report-queue to different consumers.
+   A multicaster thread is started on demand per connection. `conn` and `id` identifies the consumer.
+   Repeated calls using the same `conn` and `id` returns the same queue.
+
+   The optional third parameter, `send-end-token?`, if set to `true`, instructs the multicaster thread
+   to send `:end` if the queue is stopped. The default value is `false`.
+
+   A queue may be stopped using `stop-multicaster-id!`, `stop-multicaster!` or `stop-all-multicasters!`.
+
+   Returns a `java.util.concurrent.BlockingQueue` like `datomic.api/tx-report-queue`."
+  ([conn id]
+   (get-tx-report-queue-multicast! conn id false))
+  ([conn id send-end-token?]
+   (assert (instance? Connection conn))
+   (locking consumer-state-lock
+     (let [[new-state the-q]
+           (locking multicast-state-lock
+             (assert (map? @multicast-state))
+             (if-let [existing-q (get-in @multicast-state [:queues conn id])]
+               (do
+                 (let [new-state (swap! multicast-state
+                                        (fn [old-state]
+                                          (update-in old-state [:queues conn id] (fn [[end-token? q]]
+                                                                                   (if (not= end-token? send-end-token?)
+                                                                                     (log/debug "flipped `send-end-token?`")
+                                                                                     (log/debug "identical `send-end-token?`"))
+                                                                                   [send-end-token? q]))))]
+                   (log/debug "Returning existing queue for id" id)
+                   (assert (instance? BlockingQueue (second existing-q)))
+                   [new-state (second existing-q)]))
+               (let [needs-multicaster? (nil? (get-in @multicast-state [:queues conn]))
+                     new-q (LinkedBlockingQueue.)
+                     new-state (swap! multicast-state (fn [old-state] (assoc-in old-state [:queues conn id] [send-end-token? new-q])))]
+                 (if needs-multicaster?
+                   (do
+                     (start-multicaster! conn)
+                     (log/debug "Returning new queue for id" id "(multicaster thread started)")
+                     [new-state new-q])
+                   (do
+                     (log/debug "Returning new queue for id" id "(multicaster thread already running)")
+                     [new-state new-q])))))]
+       ; wait for multicaster thread to pick up current Queue
+       (wait-multicast-thread-step conn new-state)
+       the-q))))
+
+(defn- wait-multicast-threads-exit [[old-state new-state]]
+  (assert (map? old-state))
+  (assert (map? new-state))
+  (assert (map? (get old-state :queues {})))
+  (assert (map? (get new-state :queues {})))
+  (assert (map? (get old-state :thread-count {})))
+  (assert (map? (get new-state :thread-count {})))
+  (locking consumer-state-lock
+    ; No new multicast threads will be launched inside this block.
+    ; The lock is already held by parent function.
+    ;
+    ; Why do we need to _wait_ for multicaster thread(s) to exit after
+    ; removing all queue ids for a given connection?
+    ; Otherwise the following could happen:
+    ; 1. multicaster thread is sleeping
+    ; 2. user calls stop-multicaster!
+    ;    One would expect that multicaster thread would exit, but it is still sleeping
+    ; 3. user calls get-tx-report-queue-multicast! with the same conn
+    ;    The state is now empty, so a new multicaster thread is spawned.
+    ; 4. Now there is two multicaster threads for the same connection!
+    ;    ... and since the datomic report queue can be shared between threads
+    ;    it will seemingly work, but when the end event is sent, it will be
+    ;    sent by multiple threads.
+    (let [old-conns (into #{} (keys (get old-state :queues {})))
+          new-conns (into #{} (keys (get new-state :queues {})))]
+      (assert (every?
+                (fn [x] (instance? Connection x))
+                old-conns))
+      (assert (every?
+                (fn [x] (instance? Connection x))
+                new-conns))
+      (doseq [old-conn old-conns]
+        (when-not (contains? new-conns old-conn)
+          (let [old-threadcount (get-in old-state [:thread-count old-conn] nil)]
+            (assert (= 1 old-threadcount))
+            (let [start-ms (System/currentTimeMillis)]
+              (loop []
+                (if (= 0 (get-in @multicast-state [:thread-count old-conn]))
+                  :ok
+                  (do
+                    (let [spent-ms (- (System/currentTimeMillis) start-ms)]
+                      (if (> spent-ms 30000)
+                        (throw (RuntimeException. "Timed out waiting for multicaster thread to exit"))
+                        (do
+                          (Thread/sleep 16)
+                          (recur))))))))))))))
+
+(defn- all-queues [state]
+  (->> (mapcat (fn [[conn qmap]]
+                 (mapv (fn [q-id] [conn q-id])
+                       (keys qmap)))
+               (seq (get state :queues {})))
+       (into #{})))
+
+(comment
+  (do
+    (assert (= #{}
+               (all-queues {})))
+    (assert (= #{}
+               (all-queues {:queues {}})))
+    (assert (= #{[:conn-a :q-id]}
+               (all-queues {:queues {:conn-a {:q-id 1}}})))
+    (assert (= #{[:conn-a :q-id] [:conn-a :q-id-2]}
+               (all-queues {:queues {:conn-a {:q-id 1 :q-id-2 2}}})))
+    (assert (= #{[:conn-a :q-id-2] [:conn-b :q-id-3] [:conn-a :q-id]}
+               (all-queues {:queues {:conn-a {:q-id 1 :q-id-2 2}
+                                     :conn-b {:q-id-3 3}}})))))
+
+(defn- removed-queues? [old new]
+  (not= (all-queues old)
+        (all-queues new)))
+
+(defn stop-multicast-consumer-id! [conn id]
+  (assert (instance? Connection conn))
+  (let [did-remove? (atom nil)]
+    (locking consumer-state-lock
+      (wait-multicast-threads-exit
+        (locking multicast-state-lock
+          (let [[old new] (swap-vals! multicast-state (fn [old-state]
+                                                        (let [new-state (dissoc-in old-state [:queues conn id])]
+                                                          (if (= {} (get-in new-state [:queues conn]))
+                                                            (dissoc-in old-state [:queues conn])
+                                                            new-state))))]
+              (reset! did-remove? (removed-queues? old new))
+            [old new]))))
+    @did-remove?))
+
+(defn stop-multicaster! [conn]
+  (assert (instance? Connection conn))
+  (let [did-remove? (atom nil)]
+    (locking consumer-state-lock
+      (wait-multicast-threads-exit
+        (locking multicast-state-lock
+          (let [[old new] (swap-vals! multicast-state (fn [old-state] (dissoc-in old-state [:queues conn])))]
+            (reset! did-remove? (removed-queues? old new))
+            [old new]))))
+    @did-remove?))
+
+(defn stop-all-multicasters! []
+  (let [did-remove? (atom nil)]
+    (locking consumer-state-lock
+      (wait-multicast-threads-exit
+        (locking multicast-state-lock
+          (let [[old new] (swap-vals! multicast-state (fn [old-state] (assoc old-state :queues {})))]
+            (reset! did-remove? (removed-queues? old new))
+            [old new]))))
+    @did-remove?))
+
+(comment
+  (do
+    (require 'com.github.ivarref.yoltq.log-init)
+    (require '[datomic.api :as d])
+    (com.github.ivarref.yoltq.log-init/init-logging!
+      [[#{"datomic.*" "com.datomic.*" "org.apache.*"} :warn]
+       [#{"com.github.ivarref.yoltq.report-queue"} :debug]
+       [#{"com.github.ivarref.yoltq.poller"} :info]
+       [#{"com.github.ivarref.yoltq"} :debug]
+       ;[#{"ivarref.yoltq*"} :info]
+       [#{"*"} :info]])
+    (defonce conn (let [uri (str "datomic:mem://demo")
+                        _ (d/delete-database uri)
+                        _ (d/create-database uri)
+                        conn (d/connect uri)]
+                    conn))))
+
+(comment
+  (do
+    (require 'com.github.ivarref.yoltq.log-init)
+    (defn drain! [^BlockingQueue q]
+      (loop [items []]
+        (if-let [elem (.poll q 100 TimeUnit/MILLISECONDS)]
+          (recur (conj items elem))
+          items)))
+    (com.github.ivarref.yoltq.log-init/init-logging!
+      [[#{"datomic.*" "com.datomic.*" "org.apache.*"} :warn]
+       [#{"com.github.ivarref.yoltq.report-queue"} :debug]
+       [#{"com.github.ivarref.yoltq.poller"} :info]
+       [#{"com.github.ivarref.yoltq"} :debug]
+       ;[#{"ivarref.yoltq*"} :info]
+       [#{"*"} :info]])
+    (log/info "********************************")
+    (defonce conn (let [uri (str "datomic:mem://demo")
+                        _ (d/delete-database uri)
+                        _ (d/create-database uri)
+                        conn (d/connect uri)]
+                    conn))
+    (log/info "stop-all!")
+    (stop-all-multicasters!)
+    (assert (= 0 @thread-count))
+    (let [q1 (get-tx-report-queue-multicast! conn :q1 false)
+          q2 (get-tx-report-queue-multicast! conn :q2 false)
+          _ (get-tx-report-queue-multicast! conn :q1 true)]
+      @(d/transact conn [{:db/doc "demo"}])
+      @(d/transact conn [{:db/doc "demo"}])
+      @(d/transact conn [{:db/doc "demo"}])
+      (log/info "begin drain q1")
+      (stop-multicast-consumer-id! conn :q1)
+      (stop-multicast-consumer-id! conn :q1)
+      (println "thread count" @thread-count)
+      (let [qitems-2 (drain! q2)
+            qitems-1 (drain! q1)]
+        (assert (= :end (last qitems-1)))
+        (println "drain count q1:" (count qitems-1))
+        (println "drain count q2:" (count qitems-2))))))
+
+(comment
+  (do
+    (let [q (get-tx-report-queue-multicast! conn :q1 true)]
+      (log/debug "stopping id :q1")
+      (stop-multicaster-id! conn :q1)
+      (let [drained (drain! q)]
+        (println "drained:" drained)
+        (assert (= [:end] drained)))
+      @multicast-state)))
+
+(comment
+  (stop-all-multicasters!))
+
+(comment
+  (do
+    (let [q (get-tx-report-queue-multicast! conn :q2 false)]
+      (println "drain count:" (count (drain! q)))
+      @multicast-state
+      nil)))
+
+(comment
+  (get-tx-report-queue-multicast! conn :q1 false)
+  (get-tx-report-queue-multicast! conn :q1 true))
+
+(comment
+  (do
+    @(d/transact conn [{:db/doc "demo"}])
+    @(d/transact conn [{:db/doc "demo"}])
+    @(d/transact conn [{:db/doc "demo"}])
+    :yay))
\ No newline at end of file
diff --git a/src/com/github/ivarref/yoltq/slow_executor_detector.clj b/src/com/github/ivarref/yoltq/slow_executor_detector.clj
new file mode 100644
index 0000000..53dfe89
--- /dev/null
+++ b/src/com/github/ivarref/yoltq/slow_executor_detector.clj
@@ -0,0 +1,36 @@
+(ns com.github.ivarref.yoltq.slow-executor-detector
+  (:require [clojure.string :as str]
+            [clojure.tools.logging :as log]
+            [com.github.ivarref.yoltq.ext-sys :as ext])
+  (:import (java.util.concurrent ExecutorService)))
+
+(defn- do-show-slow-threads [{:keys [start-execute-time
+                                     max-execute-time
+                                     slow?
+                                     slow-thread-show-stacktrace?]
+                              :or {slow-thread-show-stacktrace? true}}]
+  (let [new-slow-val (atom false)]
+    (doseq [[^Thread thread [start-time queue-id queue-name]] @start-execute-time]
+      (when (> (ext/now-ms) (+ start-time max-execute-time))
+        (reset! new-slow-val true)
+        (log/error "thread" (.getName thread) "spent too much time on"
+                   "queue item" (str queue-id)
+                   "for queue" queue-name
+                   (if slow-thread-show-stacktrace?
+                       (str "stacktrace: \n" (str/join "\n" (mapv str (seq (.getStackTrace thread)))))
+                       ""))))
+    (reset! slow? @new-slow-val)))
+
+(defn show-slow-threads [^ExecutorService pool config-atom]
+  (try
+    (while (not (.isTerminated pool))
+      (try
+        (do-show-slow-threads @config-atom)
+        (catch Throwable t
+          (log/error t "do-show-slow-threads crashed:" (ex-message t))))
+      (dotimes [_ 3]
+        (when (not (.isTerminated pool))
+          (Thread/sleep 1000))))
+    (log/debug "show-slow-threads exiting")
+    (catch Throwable t
+      (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
new file mode 100644
index 0000000..ee9cd54
--- /dev/null
+++ b/src/com/github/ivarref/yoltq/test_queue.clj
@@ -0,0 +1,191 @@
+(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
+                        :prev-consumed     (atom {})
+                        :tx-queue          txq#})]
+     (with-bindings {#'yq/*config*       config#
+                     #'yq/*running?*     (atom false)
+                     #'yq/*test-mode*    true
+                     #'ext/*now-ms-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 [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#)))
+     (test/is false (str "No job found for queue " ~queue-name))))
+
+(defmacro consume! [queue-name]
+  `(consume-expect! ~queue-name :done))
+
+
+(defmacro force-retry! [queue-name]
+  `(if-let [job# (some-> @yq/*config* :prev-consumed deref (get ~queue-name))]
+     (let [db-res# @(d/transact (:conn @yq/*config*) [{:com.github.ivarref.yoltq/id     (:com.github.ivarref.yoltq/id job#)
+                                                       :com.github.ivarref.yoltq/status :init}])
+           res# (some->> (u/prepare-processing (:db-after db-res#)
+                                               (:com.github.ivarref.yoltq/id job#)
+                                               ~queue-name
+                                               (:com.github.ivarref.yoltq/lock job#)
+                                               :init)
+                         (i/take! @yq/*config*)
+                         (i/execute! @yq/*config*))]
+       (swap! (:prev-consumed @yq/*config*) assoc ~queue-name res#)
+       (test/is (= :done (:com.github.ivarref.yoltq/status res#)))
+       (if (:retval res#)
+         (:retval res#)
+         (:exception res#)))
+     (test/is false "Expected to have previously consumed something. Was nil.")))
diff --git a/src/com/github/ivarref/yoltq/utils.clj b/src/com/github/ivarref/yoltq/utils.clj
new file mode 100644
index 0000000..9defd0e
--- /dev/null
+++ b/src/com/github/ivarref/yoltq/utils.clj
@@ -0,0 +1,179 @@
+(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])
+  (:refer-clojure :exclude [random-uuid])
+  (:import (datomic Connection)
+           (java.time Duration)))
+
+
+(def status-init :init)
+(def status-processing :processing)
+(def status-done :done)
+(def status-error :error)
+
+(def current-version "2")
+
+(defn duration->millis [m]
+  (reduce-kv (fn [o k v]
+               (if (instance? Duration v)
+                 (assoc o k (.toMillis v))
+                 (assoc o k v)))
+             {}
+             m))
+
+
+(defn squuid []
+  (ext/squuid))
+
+
+(defn random-uuid []
+  (ext/random-uuid))
+
+
+(defn now-ms []
+  (ext/now-ms))
+
+
+(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/opts (fn [s] (or (when s (edn/read-string s)) {})))
+      (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 [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-ms)}]}))
+
+
+(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)))
+  (let [db (or db (d/db conn))]
+    (if-let [ids (->> (d/q '[:find ?id ?lock
+                             :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/version ?current-version]]
+                           db
+                           queue-name
+                           (- (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))
+      (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] :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-max-retries cfg queue-name)]
+    (when-let [ids (->> (d/q '[:find ?id ?lock
+                               :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]
+                               [?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]
+                               [?e :com.github.ivarref.yoltq/version ?current-version]]
+                             db
+                             queue-name
+                             (- (now-ms) error-backoff-time)
+                             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] :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-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
+                               :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]
+                               [?e :com.github.ivarref.yoltq/version ?current-version]]
+                             db
+                             queue-name
+                             (- now hung-backoff-time)
+                             current-version)
+                        (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?
+         :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)]
+                        {: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}])}))))
diff --git a/test/com/github/ivarref/yoltq/error_poller_test.clj b/test/com/github/ivarref/yoltq/error_poller_test.clj
new file mode 100644
index 0000000..4d92b81
--- /dev/null
+++ b/test/com/github/ivarref/yoltq/error_poller_test.clj
@@ -0,0 +1,35 @@
+(ns com.github.ivarref.yoltq.error-poller-test
+  (:require [clojure.edn :as edn]
+            [clojure.test :refer [deftest is]]
+            [clojure.tools.logging :as log]
+            [com.github.ivarref.yoltq.error-poller :as ep]
+            [com.github.ivarref.yoltq.log-init :as logconfig]))
+
+
+(deftest error-poller
+  (logconfig/init-logging!
+    [[#{"datomic.*" "com.datomic.*" "org.apache.*"} :warn]
+     [#{"*"} (edn/read-string
+               (System/getProperty "TAOENSSO_TIMBRE_MIN_LEVEL_EDN" ":info"))]])
+  (let [cfg {:system-error-callback-backoff 100}
+        time (atom 0)
+        tick! (fn [& [amount]]
+                (swap! time + (or amount 1)))
+        verify (fn [state now-ns error-count expected-callback]
+                 (let [{:keys [errors state run-callback] :as res} (ep/handle-error-count state cfg now-ns error-count)]
+                   (log/info errors "=>" state "::" run-callback)
+                   (is (= expected-callback run-callback))
+                   res))]
+    (-> {}
+        (verify (tick!) 0 nil)
+        (verify (tick!) 1 nil)
+        (verify (tick!) 1 nil)
+        (verify (tick!) 1 :error)
+        (verify (tick! 100) 0 nil)
+        (verify (tick!) 0 :error)
+        (verify (tick!) 0 :recovery)
+        (verify (tick!) 1 nil)
+        (verify (tick!) 1 nil)
+        (verify (tick!) 1 :error)
+        (verify (tick! 100) 1 nil)
+        (verify (tick!) 1 :error))))
diff --git a/test/com/github/ivarref/yoltq/http_hang_demo.clj b/test/com/github/ivarref/yoltq/http_hang_demo.clj
new file mode 100644
index 0000000..06d877b
--- /dev/null
+++ b/test/com/github/ivarref/yoltq/http_hang_demo.clj
@@ -0,0 +1,45 @@
+(ns com.github.ivarref.yoltq.http-hang-demo
+  (:require [datomic.api :as d]
+            [com.github.ivarref.yoltq :as yq]
+            [com.github.ivarref.yoltq.log-init])
+  (:import (java.net.http HttpClient HttpRequest HttpResponse$BodyHandlers)))
+
+(comment
+  (do
+    (com.github.ivarref.yoltq.log-init/init-logging!
+      [[#{"datomic.*" "com.datomic.*" "org.apache.*"} :warn]
+       [#{"com.github.ivarref.yoltq.report-queue"} :debug]
+       [#{"com.github.ivarref.yoltq.poller"} :info]
+       [#{"com.github.ivarref.yoltq*"} :debug]
+       [#{"*"} :info]])
+    (yq/stop!)
+    (let [received (atom [])
+          uri (str "datomic:mem://hello-world-" (java.util.UUID/randomUUID))]
+      (d/delete-database uri)
+      (d/create-database uri)
+      (let [conn (d/connect uri)]
+        (init! {:conn                          conn
+                :error-backoff-time            (Duration/ofSeconds 5)
+                :poll-delay                    5
+                :system-error-poll-interval    [1 TimeUnit/MINUTES]
+                :system-error-callback-backoff (Duration/ofHours 1)
+                :max-execute-time              (Duration/ofSeconds 3)
+                :on-system-error               (fn [] (log/error "system in error state"))
+                :on-system-recovery            (fn [] (log/info "system recovered"))})
+        (yq/add-consumer! :slow (fn [_payload]
+                                  (log/info "start slow handler...")
+                                  ; sudo tc qdisc del dev wlp3s0 root netem
+                                  ; sudo tc qdisc add dev wlp3s0 root netem delay 10000ms
+                                  ; https://jvns.ca/blog/2017/04/01/slow-down-your-internet-with-tc/
+                                  (let [request (-> (HttpRequest/newBuilder)
+                                                    (.uri (java.net.URI. "https://postman-echo.com/get"))
+                                                    (.timeout (Duration/ofSeconds 10))
+                                                    (.GET)
+                                                    (.build))]
+                                    (log/info "body is:" (-> (.send (HttpClient/newHttpClient) request (HttpResponse$BodyHandlers/ofString))
+                                                             (.body))))
+                                  (log/info "slow handler is done!")))
+        (yq/start!)
+        @(d/transact conn [(put :slow {:work 123})])
+        #_(dotimes [x 1] @(d/transact conn [(put :q {:work x})]))
+        nil))))
\ No newline at end of file
diff --git a/test/com/github/ivarref/yoltq/log_init.clj b/test/com/github/ivarref/yoltq/log_init.clj
new file mode 100644
index 0000000..7eae557
--- /dev/null
+++ b/test/com/github/ivarref/yoltq/log_init.clj
@@ -0,0 +1,66 @@
+(ns com.github.ivarref.yoltq.log-init
+  (:require [clojure.term.colors :as colors]
+            [taoensso.timbre :as timbre]
+            [clojure.string :as str]))
+
+(set! *warn-on-reflection* true)
+
+(def level-colors
+  {;:warn colors/red
+   :error colors/red})
+
+(def ^:dynamic *override-color* nil)
+
+(defn min-length [n s]
+  (loop [s s]
+    (if (>= (count s) n)
+      s
+      (recur (str s " ")))))
+
+(defn local-console-format-fn
+  "A simpler log format, suitable for readable logs during development. colorized stacktraces"
+  [data]
+  (try
+    (let [{:keys [level ?err msg_ ?ns-str ?file hostname_
+                  timestamp_ ?line context]} data
+          ts (force timestamp_)]
+      (let [color-f (if (nil? *override-color*)
+                      (get level-colors level str)
+                      colors/green)
+            maybe-stacktrace (when ?err
+                               (str "\n" (timbre/stacktrace ?err)))]
+        (cond-> (str #_(str ?ns-str ":" ?line)
+                  #_(min-length (count "[yoltq:326] ")
+                                (str
+                                  "["
+                                  (some-> ?ns-str
+                                          (str/split #"\.")
+                                          (last))
+                                  ":" ?line))
+                  ts
+                  " "
+                  (color-f (min-length 5 (str/upper-case (name level))))
+                  " "
+
+                  (when-let [x-req-id (:x-request-id context)]
+                    (str "[" x-req-id "] "))
+                  #_(.getName ^Thread (Thread/currentThread))
+
+                  (color-f (force msg_))
+
+                  maybe-stacktrace))))
+
+
+    (catch Throwable t
+      (println "error in local-console-format-fn:" (ex-message t))
+      nil)))
+
+
+(defn init-logging! [min-levels]
+  (timbre/merge-config!
+    {:min-level min-levels
+     :timestamp-opts {:pattern "HH:mm:ss.SSS"
+                      :timezone :jvm-default}
+     :output-fn (fn [data] (local-console-format-fn data))
+     :appenders {:println (timbre/println-appender {:stream :std-out})}}))
+
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/readme_demo.clj b/test/com/github/ivarref/yoltq/readme_demo.clj
new file mode 100644
index 0000000..eae0a3e
--- /dev/null
+++ b/test/com/github/ivarref/yoltq/readme_demo.clj
@@ -0,0 +1,48 @@
+(ns com.github.ivarref.yoltq.readme-demo
+  (:require [clojure.tools.logging :as log]
+            [datomic.api :as d]
+            [com.github.ivarref.yoltq :as yq])
+  (:import (java.util UUID)))
+
+
+(defonce conn
+  (let [uri (str "datomic:mem://hello-world-" (UUID/randomUUID))]
+    (d/delete-database uri)
+    (d/create-database uri)
+    (d/connect uri)))
+
+
+(com.github.ivarref.yoltq.log-init/init-logging!
+  [[#{"datomic.*" "com.datomic.*" "org.apache.*"} :warn]
+   [#{"com.github.ivarref.yoltq.report-queue"} :debug]
+   [#{"com.github.ivarref.yoltq.poller"} :info]
+   [#{"com.github.ivarref.yoltq*"} :debug]
+   [#{"*"} :info]])
+
+
+(yq/stop!)
+
+(yq/init! {:conn conn})
+
+
+(yq/add-consumer! :q
+                  (let [cnt (atom 0)]
+                    (fn [payload]
+                      (when (= 1 (swap! cnt inc))
+                        (throw (ex-info "failed" {})))
+                      (log/info "got payload" payload))))
+
+(yq/start!)
+
+@(d/transact conn [(yq/put :q {:work 123})])
+
+(comment
+  (yq/add-consumer! :q (fn [_] (throw (ex-info "always fail" {})))))
+
+(comment
+  @(d/transact conn [(yq/put :q {:work 123})]))
+
+(comment
+  (do
+    (yq/add-consumer! :q (fn [_] :ok))
+    nil))
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..0c1b2f0
--- /dev/null
+++ b/test/com/github/ivarref/yoltq/test_utils.clj
@@ -0,0 +1,80 @@
+(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 yq]
+            [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]
+            [clojure.pprint :as pp])
+  (:import (java.util UUID)
+           (java.time Duration)))
+
+
+(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 @yq/*config*) [(i/put @yq/*config* id payload {})]))
+
+
+(defn advance! [tp]
+  (assert (some? ext/*now-ms-atom*) "Expected to be running in test-mode!")
+  (swap! ext/*now-ms-atom* + (if (number? tp)
+                               tp
+                               (.toMillis ^Duration 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 @yq/*config*))))
+
+
+(defn pp [x]
+  (pp/pprint x)
+  x)
+
+(defn get-init [& args]
+  (apply u/get-init @yq/*config* args))
+
+
+(defn get-error [& args]
+  (apply u/get-error @yq/*config* args))
+
+
+(defn get-hung [& args]
+  (apply u/get-hung @yq/*config* args))
+
+
+(defn take! [& args]
+  (apply i/take! @yq/*config* args))
+
+
+(defn execute! [& 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
new file mode 100644
index 0000000..a2ed269
--- /dev/null
+++ b/test/com/github/ivarref/yoltq/virtual_test.clj
@@ -0,0 +1,474 @@
+(ns com.github.ivarref.yoltq.virtual-test
+  (: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!)
+
+
+(deftest happy-case-1
+  (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)))))
+
+(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)]
+    (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)]
+    (yq/init! {:conn     conn
+               :handlers {:q {:f (fn [payload] payload)}}})
+    (u/put-transact! :q {:work 123})
+    (u/advance! (:init-backoff-time yq/default-opts))
+    (is (= {:work 123} (some->> (u/get-init :q)
+                                (u/take!)
+                                (u/execute!)
+                                :retval)))))
+
+
+(deftest happy-case-queue-fn-allow-cas-failure
+  (let [conn (u/empty-conn)
+        invoke-count (atom 0)]
+    (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"]]))
+                      {:allow-cas-failure? #{:e/val}})
+    @(d/transact conn #d/schema [[:e/id :one :string :id]
+                                 [:e/val :one :string]])
+    @(d/transact conn [{:e/id "demo"}
+                       (yq/put :q {:id "demo"})])
+    (u/advance! (:init-backoff-time yq/default-opts))
+    (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! yq/*config* dissoc :mark-status-fn!)
+
+    ; (mark-status! :done) failed but f succeeded.
+    (is (nil? (u/get-hung :q)))
+    (u/advance! (:hung-backoff-time @yq/*config*))
+    (is (some? (u/get-hung :q)))
+    (is (true? (some->> (u/get-hung :q)
+                        (u/take!)
+                        (u/execute!)
+                        :allow-cas-failure?)))
+    (is (= 2 @invoke-count))))
+
+
+(deftest backoff-test
+  (let [conn (u/empty-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})
+    (is (nil? (u/get-init :q)))
+
+    (u/advance! (dec (:init-backoff-time yq/default-opts)))
+    (is (nil? (u/get-init :q)))
+    (u/advance! 1)
+    (is (some? (u/get-init :q)))
+
+    (is (some? (some->> (u/get-init :q)
+                        (u/take!)
+                        (u/execute!)
+                        :exception)))
+
+    (u/advance! (dec (:error-backoff-time @yq/*config*)))
+    (is (nil? (u/get-error :q)))
+    (u/advance! 1)
+    (is (some? (u/get-error :q)))))
+
+
+(deftest get-hung-test
+  (let [conn (u/empty-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})
+    (u/advance! (:init-backoff-time yq/default-opts))
+    (is (some? (u/get-init :q)))
+
+    (is (= :processing (some->> (u/get-init :q)
+                                (u/take!)
+                                :com.github.ivarref.yoltq/status)))
+
+    (is (nil? (u/get-hung :q)))
+    (u/advance! (dec (:hung-backoff-time yq/default-opts)))
+    (is (nil? (u/get-hung :q)))
+    (u/advance! 1)
+    (is (some? (u/get-hung :q)))))
+
+
+(deftest basic-locking
+  (let [conn (u/empty-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" {})))}}})
+    (u/put-transact! :q {:work 123})
+    (u/advance! (:init-backoff-time yq/default-opts))
+    (is (some? (u/get-init :q)))
+
+    (let [job (u/get-init :q)]
+      (is (= :processing (some->> job (u/take!) :com.github.ivarref.yoltq/status)))
+      (u/take! job)
+      (is (= 1 @(:cas-failures @yq/*config*))))))
+
+
+(deftest retry-test
+  (let [conn (u/empty-conn)]
+    (yq/init! {:conn              conn
+               :init-backoff-time (:init-backoff-time yq/default-opts)
+               :handlers          {:q {:f
+                                       (let [c (atom 0)]
+                                         (fn [_]
+                                           (if (<= (swap! c inc) 2)
+                                             (throw (ex-info "janei" {}))
+                                             ::ok)))}}})
+    (u/put-transact! :q {:work 123})
+
+    (u/advance! (:init-backoff-time yq/default-opts))
+    (is (some? (some->> (u/get-init :q) (u/take!) (u/execute!) :exception)))
+
+    (u/advance! (:error-backoff-time @yq/*config*))
+    (is (some? (some->> (u/get-error :q) (u/take!) (u/execute!) :exception)))
+
+    (u/advance! (:error-backoff-time @yq/*config*))
+    (is (nil? (some->> (u/get-error :q) (u/take!) (u/execute!) :exception)))))
+
+
+(deftest max-retries-test
+  (let [conn (u/empty-conn)
+        call-count (atom 0)]
+    (yq/init! {:conn               conn
+               :error-backoff-time 0})
+    (yq/add-consumer! :q (fn [_]
+                           (swap! call-count inc)
+                           (throw (ex-info "janei" {})))
+                      {:max-retries 1})
+    (tq/put! :q {:work 123})
+    (is (some? (:exception (tq/run-one-report-queue!))))
+
+    (dotimes [_ 10]
+      (tq/run-queue-once! :q :error))
+    (is (= 2 @call-count))))
+
+
+(deftest max-retries-test-two
+  (let [conn (u/empty-conn)
+        call-count (atom 0)]
+    (yq/init! {:conn               conn
+               :error-backoff-time 0})
+    (yq/add-consumer! :q (fn [_]
+                           (swap! call-count inc)
+                           (throw (ex-info "janei" {})))
+                      {:max-retries 3})
+    (tq/put! :q {:work 123})
+    (is (some? (:exception (tq/run-one-report-queue!))))
+
+    (timbre/with-level :fatal
+                       (dotimes [_ 20]
+                         (tq/run-queue-once! :q :error)))
+    (is (= 4 @call-count))))
+
+
+(deftest hung-to-error
+  (let [conn (u/empty-conn)
+        call-count (atom 0)
+        missed-mark-status (atom nil)]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q
+                      (fn [_]
+                        (if (= 1 (swap! call-count inc))
+                          (throw (ex-info "error" {}))
+                          (log/info "return OK"))))
+    (tq/put! :q {:id "demo"})
+    (tq/run-one-report-queue!)                              ; now in status :error
+
+
+    (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*))
+    (tq/run-queue-once! :q :error)
+    (swap! yq/*config* dissoc :mark-status-fn!)
+    (is (= :done @missed-mark-status))
+
+    (is (nil? (uu/get-hung @yq/*config* :q)))
+    (u/advance! (:hung-backoff-time @yq/*config*))
+
+    (is (some? (uu/get-hung @yq/*config* :q)))
+
+    (is (= 2 @call-count))
+
+    (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 @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 #{})]
+    (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 [(yq/put :q 1)])
+    @(d/transact conn [(yq/put :q 2)])
+
+    (is (= 1 (tq/consume-expect! :q :done)))
+    (tq/consume-expect! :q :error)))
+
+
+(def ^:dynamic *some-binding* nil)
+
+
+(deftest binding-test
+  (let [conn (u/empty-conn)]
+    (yq/init! {:conn             conn
+               :capture-bindings [#'*some-binding* #'timbre/*context*]})
+    (yq/add-consumer! :q (fn [_] *some-binding*))
+    (binding [timbre/*context* {:x-request-id "wooho"}]
+      (binding [*some-binding* 1]
+        @(d/transact conn [(yq/put :q nil)]))
+      (binding [*some-binding* 2]
+        @(d/transact conn [(yq/put :q nil)]))
+      @(d/transact conn [(yq/put :q nil)]))
+
+    (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)]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q (fn [_] (:x-request-id timbre/*context*)))
+    (binding [timbre/*context* {:x-request-id "123"}]
+      @(d/transact conn [(yq/put :q nil)]))
+    (is (= "123" (tq/consume-expect! :q :done)))))
+
+
+(deftest force-retry-test
+  (let [conn (u/empty-conn)]
+    (yq/init! {:conn conn})
+    (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)))
+    (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 {:id "a1"} {:id "a1"})])
+    @(d/transact conn [(yq/put :b {:id "b1"} {:depends-on [:a "a1"]})])
+
+    ; can't consume :b yet:
+    (is (= {:depends-on [:a "a1"]} (tq/consume! :b)))
+
+    (is (= {:id "a1"} (tq/consume! :a)))
+    (is (= {:id "b1"} (tq/consume! :b)))))
+
+
+(deftest depends-on-queue-level
+  (let [conn (u/empty-conn)]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :a identity)
+    (yq/add-consumer! :b identity {:depends-on (fn [{:keys [id]}] [:a id])})
+    @(d/transact conn [(yq/put :a {:id "1"} {:id "1"})])
+    @(d/transact conn [(yq/put :b {:id "1"})])
+
+    ; can't consume :b yet:
+    (is (= {:depends-on [:a "1"]} (tq/consume! :b)))
+
+    (is (= {:id "1"} (tq/consume! :a)))
+    (is (= {:id "1"} (tq/consume! :b)))))
+
+
+(deftest verify-can-read-string
+  (let [conn (u/empty-conn)]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :a identity)
+    (timbre/with-level :fatal
+      (is (thrown? Exception @(d/transact conn [(yq/put :a {:broken #'=})]))))))
+
+
+(deftest payload-verifier
+  (let [conn (u/empty-conn)]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q identity
+                      {:valid-payload? (fn [{:keys [id]}]
+                                         (some? id))})
+    @(d/transact conn [(yq/put :q {:id "a"})])
+    (timbre/with-level :fatal
+                       (is (thrown? Exception @(d/transact conn [(yq/put :q {})]))))))
+
+
+(defn my-consumer
+  {:yoltq/queue-id :some-q}
+  [state payload]
+  (swap! state conj payload))
+
+(deftest queue-id-can-be-var
+  (let [conn (u/empty-conn)
+        received (atom #{})]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! #'my-consumer (partial my-consumer received))
+    @(d/transact conn [(yq/put #'my-consumer {:id "a"})])
+    (tq/consume! :some-q)
+    (is (= #{{:id "a"}} @received))
+    #_(timbre/with-level :fatal
+                         (is (thrown? Exception @(d/transact conn [(yq/put :q {})]))))))
+
+(deftest healthy-allowed-error-time-test
+  (let [conn (u/empty-conn)]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q (fn [_] (throw (ex-info "" {}))))
+    @(d/transact conn [(yq/put :q {:work 123})])
+    (tq/consume-expect! :q :error)
+    (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"))))
+
+(deftest job-group-test
+  (let [conn (u/empty-conn)]
+    (yq/init! {:conn conn})
+    (yq/add-consumer! :q1 identity)
+    (yq/add-consumer! :q2 identity)
+    @(d/transact conn [(yq/put :q1 {:work 123} {:job-group :group1})
+                       (yq/put :q1 {:work 456} {:job-group :group2})
+                       (yq/put :q2 {:work 789} {:job-group :group1})])
+    (is (= [{:qname :q1
+             :job-group :group1
+             :status :init
+             :count 1}]
+           (yq/job-group-progress :q1 :group1)))
+
+    (is (= {:work 123} (tq/consume! :q1)))
+
+    (is (= [{:qname :q1
+             :job-group :group1
+             :status :done
+             :count 1}]
+           (yq/job-group-progress :q1 :group1)))))