       (>!! outbox-ch entry)))
   state)
 
-(defn time-out-peers! [state peers]
+(defn time-out-peers! [state type peers]
   (if-not (empty? peers)
     (let [{:keys [onyx.core/log-prefix onyx.core/id onyx.core/task] :as event} (get-event state)
-          timeout-msg (format "%s Peers timed out %s" log-prefix (vec peers) id task)
+          timeout-msg (format "%s Peers timed out %s, peers: %s" log-prefix type (vec peers) id task)
           next-state (reduce evict-dead-peer! state peers)] 
       (-> timeout-msg (doto println) (doto info))
       ;; backoff for a while so we don't repeatedly write leave messages


(def DEBUG false)

(defn iteration [state replica-val n-iters]
  ;(when DEBUG (viz/update-monitoring! state-machine))
  (loop [state (exec (next-replica! state replica-val)) n n-iters]
    ;(print-state state)
    ; (when (zero? (rand-int 10000)) 
    ;   (print-state state))
    (if (and (advanced? state) 
             (or (not (new-iteration? state))
                 (pos? n)))
      (recur (exec state) (dec n))
      state)))

(defn run-task-lifecycle!
  "The main task run loop, read batch, ack messages, etc."

  (try
    (let [{:keys [onyx.core/replica-atom] :as event} (get-event state)]
      (loop [state state 
             prev-replica-val (get-replica state)
             replica-val @replica-atom]
        (debug (:onyx.core/log-prefix event) ", new task iteration")
        (if (and (= replica-val prev-replica-val)
                 (not (killed? state)))
          (recur (iteration state task-iterations) replica-val @replica-atom) 
          (let [next-state (next-replica! state replica-val)]
            (if (killed? next-state)
              (do
               (info (:onyx.core/log-prefix event) ", Fell out of task lifecycle loop")
               next-state)
              (recur next-state replica-val replica-val))))))
    (catch Throwable e
      (let [lifecycle (get-lifecycle state)
            action (if (:kill-job? (ex-data e))
