                         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 %})))
+                     #(extensions/emit (:onyx.core/monitoring event) {:event :peer/complete-message :latency %})))
 
                   (= ch retry-ch)
                   (->> (p-ext/retry-message pipeline event v)

            link (operation/peer-link @replica state event (:completion-id root))]
        (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]

                      results-by-acker))]
      (measure-latency
       #(extensions/internal-ack-messages messenger event link acks)
       #(extensions/emit (:onyx.core/monitoring event) {:event :peer/ack-messages :latency %}))))
  event)

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