feat: Make total backoff duration configurable

This is how the `backoff` crate supports limiting how long we back off
for. The default is 15 mins (!) which is crazy long for our purposes.

The configuration is exposed via the `Config` struct under
`retry_duration_msec`. The default is 30s, which is completely arbitrary
but reasonable-sounding. Retry logic was refactored into a method to
make it easier to use the configuration.

src/config.rs

@@ -80,6 +80,11 @@ pub struct Config {
     /// Duration of a UPnP port mapping.
     #[structopt(long)]
     pub upnp_lease_duration: Option<u32>,
+    /// How long to retry establishing connections and sending messages.
+    ///
+    /// The duration is in milliseconds. Setting this to 0 will effectively disable retries.
+    #[structopt(long, default_value = "30000")]
+    pub retry_duration_msec: u64,
 }
 
 /// To be used to read and write our certificate and private key to disk esp. as a part of our

src/endpoint.rs

@@ -300,7 +300,7 @@ impl Endpoint {
         Ok(())
     }
 
-    /// Connects to another peer, retries X times if the connection fails for any reason
+    /// Connects to another peer, retries for `config.retry_duration_msec` if the connection fails.
     ///
     /// Returns `Connection` which is a handle for sending messages to the peer and
     /// `IncomingMessages` which is a stream of messages received from the peer.
@@ -364,7 +364,7 @@ impl Endpoint {
 
     /// Attempt a connection to a node_addr using exponential backoff
     async fn attempt_connection(&self, node_addr: &SocketAddr) -> Result<quinn::NewConnection> {
-        retry(ExponentialBackoff::default(), || async {
+        self.retry(|| async {
             trace!("Attempting to connect to {:?}", node_addr);
             let connecting = match self.quic_endpoint.connect_with(
                 self.client_cfg.clone(),
@@ -510,10 +510,8 @@ impl Endpoint {
         }
         self.connect_to(dest).await?;
 
-        retry(ExponentialBackoff::default(), || async {
-            Ok(self.try_send_message(msg.clone(), dest).await?)
-        })
-        .await
+        self.retry(|| async { Ok(self.try_send_message(msg.clone(), dest).await?) })
+            .await
     }
 
     /// Close all the connections of this endpoint immediately and stop accepting new connections.
@@ -564,4 +562,16 @@ impl Endpoint {
     pub fn bootstrap_nodes(&self) -> &[SocketAddr] {
         &self.bootstrap_nodes
     }
+
+    fn retry<I, E, Fn, Fut>(&self, op: Fn) -> impl futures::Future<Output = Result<I, E>>
+    where
+        Fn: FnMut() -> Fut,
+        Fut: futures::Future<Output = Result<I, backoff::Error<E>>>,
+    {
+        let backoff = ExponentialBackoff {
+            max_elapsed_time: Some(Duration::from_millis(self.qp2p_config.retry_duration_msec)),
+            ..Default::default()
+        };
+        retry(backoff, op)
+    }
 }