 (defn store-command' [q old-command]
   (apply tqueue/store-command q (unroll-old-command old-command)))
 
+(def default-timeout-in-ms
+  (* 1000 60 5))
+
+(defn take-with-timeout!!
+  "Takes from `port` via <!!, but will throw an exception if
+  `timeout-in-ms` expires"
+  [port timeout-in-ms]
+  (async/alt!!
+    (async/timeout timeout-in-ms)
+    (throw (Exception. (format "Take from channel failed after timeout of '%s' ms was exceeded" timeout-in-ms)))
+    port
+    ([v] v)))
+
 (deftest command-processor-integration
   (let [command {:command "replace catalog" :version 5
                  :payload (get-in wire-catalogs [5 :empty])}]

            (reset! second-message? true)

            (is (empty? (fs/list-dir (:path dlo))))
            (let [failed-cmdref (async/<!! command-chan)]
              (is (= 1 (count (:attempts failed-cmdref))))
              (is (re-matches
                   #"(?sm).*ERROR: could not serialize access due to concurrent update.*"

            (reset! second-message? true)
            (is (empty? (fs/list-dir (:path dlo))))

            (let [failed-cmdref (take-with-timeout!! command-chan default-timeout-in-ms)]
              (is (= 1 (count (:attempts failed-cmdref))))
              (is (re-matches
                   #"(?sm).*ERROR: could not serialize access due to concurrent update.*"
