 
 (defn check-upstream-heartbeats [state liveness-timeout-ns]
   (let [curr-time (System/nanoTime)]
-    (->> (sub/status-pubs (m/subscriber (get-messenger state)))
-         (filter (fn [[peer-id spub]] 
-                   (< (+ (status-pub/get-heartbeat spub)
-                         liveness-timeout-ns)
-                      curr-time)))
-         (map key)
-         (reduce evict-peer! state)
-         (advance))))
+    (->> state
+         (get-messenger)
+         (m/subscriber)
+         (sub/status-pubs)
+         (sequence (comp (filter (fn [[peer-id spub]] 
+                                   (< (+ (status-pub/get-heartbeat spub)
+                                         liveness-timeout-ns)
+                                      curr-time)))
+                         (map key)))
+         (reduce evict-peer! state))))
 
 (defn offer-heartbeats [state]
   (advance (heartbeat! state)))

      (let [event (get-event state)
            ;; output recovery is only supported with onyx/n-peers set
            ;; as we can't currently scale slot recovery up and down
            stored (res/recover-output event recover-coordinates)
            _ (info (:onyx.core/log-prefix event) "Recover output pipeline checkpoint:" stored)]
        (p/recover! pipeline (t/replica-version state) stored)))
    (if (p/synced? pipeline (t/epoch state))
      (-> state

               transform/apply-fn-single)]
    (fn [state]
      (transform/apply-fn a-fn f state))))

(def state-fn-builders
  {:recover [{:lifecycle :lifecycle/poll-recover
