   (->> (get-messenger state)
        (m/subscriber)
        (sub/timed-out-publishers)
-       (time-out-peers! state)
+       (time-out-peers! state :upstream)
        (advance)))
 
 (defn offer-heartbeats [state]

      (>!! outbox-ch entry)))
  state)

(defn time-out-peers! [state 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)
          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 n-iters]
  ;(when DEBUG (viz/update-monitoring! state-machine))
  (loop [state (exec state) n n-iters]
    ;(print-state state)
    ; (when (zero? (rand-int 10000)) 
    ;   (print-state state))
    (if (and (advanced? state) (pos? n))
      (recur (exec state) ;; we could unroll exec loop a bit
             (if (new-iteration? state) 
               (dec n)
               n))
      state)))

(def task-iterations 1)

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