 
           (while @running?
             (try
-              (when-let [[mid mcontent attempt :as poll-reply]
-                         (wcar conn (dequeue qname opts))]
-                (if (= poll-reply "eoq-backoff")
-                  (when eoq-backoff-ms (Thread/sleep eoq-backoff-ms))
-                  (let [{:keys [status throwable backoff-ms]}
-                        (try (handler {:message mcontent :attempt attempt})
-                             (catch Throwable t {:status :error
-                                                 :throwable t}))]
-                    (case status
-                      :success (done mid)
-                      :retry   (retry mid backoff-ms)
-                      :error   (error mid poll-reply throwable)
-                      (do (done mid)
-                          (timbre/warn (str "Invalid handler status:" status)))))))
+              (let [eoq-backoff-ms* (if (fn? eoq-backoff-ms)
+                                      (eoq-backoff-ms (swap! dry-runs inc))
+                                      eoq-backoff-ms)
+                    opts* (assoc opts :eoq-backoff-ms eoq-backoff-ms*)]
+                (when-let [[mid mcontent attempt :as poll-reply]
+                           (wcar conn (dequeue qname opts*))]
+                  (if (= poll-reply "eoq-backoff")
+                    (when eoq-backoff-ms* (Thread/sleep eoq-backoff-ms*))
+                    (let [{:keys [status throwable backoff-ms]}
+                          (try (handler {:message mcontent :attempt attempt})
+                               (catch Throwable t {:status :error
+                                                   :throwable t}))]
+                      (reset! dry-runs 0)
+                      (case status
+                        :success (done mid)
+                        :retry   (retry mid backoff-ms)
+                        :error   (error mid poll-reply throwable)
+                        (do (done mid)
+                            (timbre/warn (str "Invalid handler status:" status))))))))
               (catch Throwable t
                 (timbre/error t "FATAL worker error!")
                 (throw t)))


(defprotocol IWorker (start [this]) (stop [this]))

(defrecord Worker [conn qname running? opts]
  IWorker
  (stop  [_] (let [stopped? @running?] (reset! running? false) stopped?))
  (start [_]
    (when-not @running?
      (reset! running? true)
      (future
        (let [{:keys [handler throttle-ms eoq-backoff-ms]} opts
              qk    (partial qkey qname)

                                   {:status :success})
                         lock-ms        (* 60 60 1000)
                         throttle-ms    200
                         eoq-backoff-ms (fn [dry-runs] (exp-backoff dry-runs
                                                        {:max 10000}))
                         auto-start?    true}}]]
  (let [w (Worker. conn qname (atom false) (atom 0)
                   {:handler        handler
                    :lock-ms        lock-ms
                    :throttle-ms    throttle-ms
                    :eoq-backoff-ms eoq-backoff-ms})]
    (when auto-start? (start w)) w))

(comment
  (def w1 (worker {} "myq"))
  (wcar {} (enqueue "myq" "msg"))
  (stop w1)
  (queue-status {} "myq"))

;;;; Renamed/deprecated

