                   (= ch completion-ch)
                   (let [{:keys [id peer-id]} v
                         peer-link (operation/peer-link @replica state event peer-id)]
-                    (extensions/internal-complete-message messenger event id peer-link))
+                    (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)

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