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

          ;; check if downstream peers are still up
          (->> pubs
               (mapcat pub/timed-out-subscribers)
               (time-out-peers! this)))
        this)))
  (print-state [this]
    (let [task-map (:onyx.core/task-map event)] 

  (next-replica! [this new-replica]
    (if (= replica new-replica)
      this
      (let [{:keys [onyx.core/job-id onyx.core/task-id onyx.core/job-id onyx.core/id onyx.core/task-kill-flag]} event
            old-version (get-in replica [:allocation-version job-id])
            new-version (get-in new-replica [:allocation-version job-id])]
        (cond (= old-version new-version)
              (-> this 
                  (set-coordinator! (coordinator/next-state coordinator replica new-replica))
                  (set-replica! new-replica))

              (let [allocated (common/peer->allocated-job (:allocations new-replica) id)] 
                (or (killed? this)
                    (not= task-id (:task allocated))
                    (not= job-id (:job allocated))))
              ;; Manually hit the kill switch early since we've been reallocated
              ;; and we want to escape ASAP
              (do
               (reset! task-kill-flag true)
               this)

              :else
              (let [next-messenger (ms/next-messenger-state! messenger event replica new-replica)]
                (assert (m/subscriber next-messenger) [:old-replica replica
                                                       :new-replica new-replica])
                (checkpoint/cancel! (:onyx.core/storage event))
                (-> this
                    (set-sealed! false)
                    (set-messenger! next-messenger)
                    (set-coordinator! (coordinator/next-state coordinator replica new-replica))
                    (set-replica! new-replica)
                    (reset-event!)
                    (goto-recover!)))))))
  (set-windows-state! [this new-windows-state]
    (set! windows-state new-windows-state)
    this)
