[Aptos Data Client] Only update metrics on successful responses.

Author: JoshLind

state-sync/aptos-data-client/src/client.rs

@@ -395,6 +395,7 @@ impl AptosDataClient {
         T: TryFrom<StorageServiceResponse, Error = E>,
         E: Into<Error>,
     {
+        // Select a peer to service the request
         let peer = self.choose_peer_for_request(&request).map_err(|error| {
             debug!(
                 (LogSchema::new(LogEntry::StorageServiceRequest)
@@ -404,7 +405,8 @@ impl AptosDataClient {
             );
             error
         })?;
-        let _timer = start_request_timer(&metrics::REQUEST_LATENCIES, &request.get_label(), peer);
+
+        // Send the request to the peer and transform the response
         self.send_request_to_peer_and_decode(peer, request, request_timeout_ms)
             .await
     }
@@ -420,13 +422,29 @@ impl AptosDataClient {
         T: TryFrom<StorageServiceResponse, Error = E>,
         E: Into<Error>,
     {
+        // Start the timer for the request
+        let timer = start_request_timer(&metrics::REQUEST_LATENCIES, &request.get_label(), peer);
+
+        // Get the response from the peer
         let response = self
             .send_request_to_peer(peer, request.clone(), request_timeout_ms)
-            .await?;
+            .await;
 
-        let (context, storage_response) = response.into_parts();
+        // If an error occurred, stop the timer (without updating the metrics)
+        // and return the error. Otherwise, stop the timer and update the metrics.
+        let storage_response = match response {
+            Ok(storage_response) => {
+                timer.stop_and_record(); // Update the latency metrics
+                storage_response
+            },
+            Err(error) => {
+                timer.stop_and_discard(); // Discard the timer without updating the metrics
+                return Err(error);
+            },
+        };
 
         // Ensure the response obeys the compression requirements
+        let (context, storage_response) = storage_response.into_parts();
         if request.use_compression && !storage_response.is_compressed() {
             return Err(Error::InvalidResponse(format!(
                 "Requested compressed data, but the response was uncompressed! Response: {:?}",
@@ -439,10 +457,10 @@ impl AptosDataClient {
             )));
         }
 
-        // try to convert the storage service enum into the exact variant we're expecting.
+        // Try to convert the storage service enum into the exact variant we're expecting
         match T::try_from(storage_response) {
             Ok(new_payload) => Ok(Response::new(context, new_payload)),
-            // if the variant doesn't match what we're expecting, report the issue.
+            // If the variant doesn't match what we're expecting, report the issue
             Err(err) => {
                 context
                     .response_callback

state-sync/aptos-data-client/src/poller.rs

@@ -9,7 +9,7 @@ use crate::{
     latency_monitor::LatencyMonitor,
     logging::{LogEntry, LogEvent, LogSchema},
     metrics,
-    metrics::{set_gauge, start_request_timer, DataType, PRIORITIZED_PEER, REGULAR_PEER},
+    metrics::{set_gauge, DataType, PRIORITIZED_PEER, REGULAR_PEER},
     utils,
     utils::choose_peers_by_latency,
 };
@@ -408,21 +408,13 @@ pub(crate) fn poll_peer(
         let use_compression = data_summary_poller.data_client_config.use_compression;
         let storage_request = StorageServiceRequest::new(data_request, use_compression);
 
-        // Start the peer polling timer
-        let timer = start_request_timer(
-            &metrics::REQUEST_LATENCIES,
-            &storage_request.get_label(),
-            peer,
-        );
-
         // Fetch the storage summary for the peer and stop the timer
         let request_timeout = data_summary_poller.data_client_config.response_timeout_ms;
         let result: crate::error::Result<StorageServerSummary> = data_summary_poller
             .data_client
             .send_request_to_peer_and_decode(peer, storage_request, request_timeout)
             .await
             .map(Response::into_payload);
-        drop(timer);
 
         // Mark the in-flight poll as now complete
         data_summary_poller.in_flight_request_complete(&peer);