[rust/rqd] Optimize retry mechanism and add logging info (#1881)

* When rqd fails to connect, don't swallow the failure message.
* When on Retrying state, avoid cycling back to Initializing but start a
new request immediatly and
transition to Cloning.

Author: DiegoTavares

rust/config/rqd.yaml

@@ -56,6 +56,10 @@ grpc:
   # Default: 10.0
   # backoff_jitter_percentage: 10.0
 
+  # Number of attempts to retry on Transport Error
+  # Default: 20
+  # backoff_retry_attempts: 10
+
 # =============================================================================
 # MACHINE CONFIGURATION
 # =============================================================================

rust/crates/rqd/src/config/mod.rs

@@ -44,6 +44,7 @@ pub struct GrpcConfig {
     #[serde(with = "humantime_serde")]
     pub backoff_delay_max: Duration,
     pub backoff_jitter_percentage: f64,
+    pub backoff_retry_attempts: usize,
 }
 
 impl Default for GrpcConfig {
@@ -56,6 +57,7 @@ impl Default for GrpcConfig {
             backoff_delay_min: Duration::from_millis(10),
             backoff_delay_max: Duration::from_secs(60),
             backoff_jitter_percentage: 10.0,
+            backoff_retry_attempts: 20,
         }
     }
 }

rust/crates/rqd/src/report/report_client.rs

@@ -89,7 +89,7 @@ impl ReportClient {
 
         // Requests will retry indefinitely
         let retry_policy = BackoffPolicy {
-            attempts: None,
+            attempts: Some(config.backoff_retry_attempts),
             backoff,
         };
         let retry_layer = RetryLayer::new(retry_policy);

rust/crates/rqd/src/report/retry/backoff_policy.rs

@@ -45,16 +45,16 @@ impl BackoffPolicy {
     pub fn has_attempts_left(&mut self) -> bool {
         match self.attempts {
             Some(0) => false,
-            Some(ref mut attemps_left) => {
-                *attemps_left -= 1;
+            Some(ref mut attempts_left) => {
+                *attempts_left -= 1;
                 true
             }
             None => true,
         }
     }
 }
 
-impl<E> Policy<Req, Res, E> for BackoffPolicy {
+impl<E: std::fmt::Display> Policy<Req, Res, E> for BackoffPolicy {
     type Future = tokio::time::Sleep;
     type ClonedOutput = ClonedReq<Req>;
     type ClonedFuture = Pin<Box<dyn std::future::Future<Output = Self::ClonedOutput> + Send>>;
@@ -78,9 +78,9 @@ impl<E> Policy<Req, Res, E> for BackoffPolicy {
                     Outcome::Return(result)
                 }
             }
-            Err(_err) => {
+            Err(err) => {
                 if self.has_attempts_left() {
-                    warn!("Retrying for Transport error.");
+                    warn!("Retrying for Transport error. {}", err);
                     // Retry all transport errors
                     Outcome::Retry(self.backoff.next_backoff())
                 } else {
@@ -110,7 +110,6 @@ pub struct ClonedReq<Req> {
     pub original_req: Req,
     pub cloned_req: Req,
 }
-
 impl ClonedRequest<Req> for ClonedReq<Req> {
     fn inner(self) -> (Req, Req) {
         (self.original_req, self.cloned_req)

rust/crates/rqd/src/report/retry/future.rs

@@ -136,8 +136,17 @@ where
 
                 StateProj::Retrying => {
                     ready!(this.retry.as_mut().project().service.poll_ready(cx))?;
-                    trace!("Retrying");
-                    this.state.set(State::Initialized);
+                    trace!("Retrying - making fresh call");
+
+                    // Take the request and clone it for retry
+                    let req = this
+                        .request
+                        .take()
+                        .expect("request should be available for retry");
+                    let clone_future = this.retry.policy.clone_request(req);
+                    this.state.set(State::Cloning {
+                        future: clone_future,
+                    });
                 }
             }
         }