 
 (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)))

      ;; ensure we don't try to recover output again before synced
      (set-context! state (assoc context :recovered? true)))))

(defn poll-recover-input-function [state]
  (let [messenger (get-messenger state)
        subscriber (m/subscriber messenger)
        _ (sub/poll! subscriber)]
    (if (and (sub/blocked? subscriber)
             (sub/recovered? subscriber))
      (-> state

      state)))

(defn poll-recover-output [state]
  (let [subscriber (m/subscriber (get-messenger state))]
    (sub/poll! subscriber)
    (poll-recover-check-upstream state)
    (if (and (sub/blocked? subscriber)
             (sub/recovered? subscriber))
      (-> state
