     (when (seq (filter (fn [leaf] (= :retry (:action (:routes leaf)))) (:leaves result)))
       (let [root (:root result)
             link (operation/peer-link @replica state event (:completion-id root))]
-        (extensions/internal-retry-message
-          messenger
-          event
-          (:id root)
-          link))))
+        (measure-latency
+         #(extensions/internal-retry-message messenger event (:id root) link)
+         #(extensions/emit (:onyx.core/monitoring) {:event :peer/retry-message :latency %})))))
   event)
 
 (defn inject-batch-resources [compiled-before-batch-fn pipeline event]

              [rotating-seq.core :as rsc]
              [onyx.log.commands.common :as common]
              [onyx.log.entry :as entry]
              [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]

            (swap! state dissoc k)
            (extensions/close-peer-connection (:onyx.core/messenger event) 
                                              event 
                                              (:link (get (:links snapshot) k)))
            (extensions/emit (:onyx.core/monitoring event) {:event :peer/gc-peer-link})))
        (catch InterruptedException e
          (throw e))
        (catch Throwable e
