               [rotating-seq.core :as rsc]
               [onyx.log.commands.common :as common]
               [onyx.log.entry :as entry]
+              [onyx.monitoring.measurements :refer [measure-latency]]
               [onyx.static.planning :refer [find-task find-task-fast build-pred-fn]]
               [onyx.messaging.acking-daemon :as acker]
               [onyx.peer.pipeline-extensions :as p-ext]

                                            ;; input task produces a segment that goes nowhere.
                                            (or fused-vals 0)
                                            (System/currentTimeMillis))))
                      results-by-acker))] 
      (extensions/internal-ack-messages messenger event link acks)))
  event)

(defn flow-retry-messages [replica state messenger {:keys [onyx.core/results] :as event}]

                  (= ch completion-ch)
                  (let [{:keys [id peer-id]} v
                        peer-link (operation/peer-link @replica state event peer-id)]
                    (measure-latency
                     #(extensions/internal-complete-message messenger event id peer-link)
                     #(extensions/emit (:onyx.core/monitoring) {:event :peer/complete-message :latency %})))

                  (= ch retry-ch)
                  (->> (p-ext/retry-message pipeline event v)
