WIP

Cargo.toml

@@ -19,6 +19,7 @@ futures = "0.3"
 tracing = { git = "https://github.com/tokio-rs/tracing/", tag = "tracing-0.1.30" }
 prost = "0.11.0"
 tracing-tower = { git = "https://github.com/tokio-rs/tracing/", tag = "tracing-0.1.30" }
+backoff = { version = "0.4.0", features = ["tokio"] }
 
 [dev-dependencies]
 structopt = "0.3"

src/lib.rs

@@ -26,6 +26,7 @@ mod codec;
 pub type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;
 
 /// A category of ABCI method.
+#[derive(Debug)]
 pub enum MethodKind {
     Consensus,
     Mempool,

src/server.rs

@@ -1,5 +1,7 @@
 use std::convert::{TryFrom, TryInto};
+use std::sync::Arc;
 
+use backoff::ExponentialBackoff;
 use futures::future::{FutureExt, TryFutureExt};
 use futures::sink::SinkExt;
 use futures::stream::{FuturesOrdered, StreamExt};
@@ -19,6 +21,7 @@ use crate::{
 
 /// An ABCI server which listens for connections and forwards requests to four
 /// component ABCI [`Service`]s.
+#[derive(Clone)]
 pub struct Server<C, M, I, S> {
     consensus: C,
     mempool: M,
@@ -122,18 +125,25 @@ impl<C, M, I, S> Server<C, M, I, S>
 where
     C: Service<ConsensusRequest, Response = ConsensusResponse, Error = BoxError>
         + Send
+        + Sync
         + Clone
         + 'static,
     C::Future: Send + 'static,
     M: Service<MempoolRequest, Response = MempoolResponse, Error = BoxError>
         + Send
+        + Sync
         + Clone
         + 'static,
     M::Future: Send + 'static,
-    I: Service<InfoRequest, Response = InfoResponse, Error = BoxError> + Send + Clone + 'static,
+    I: Service<InfoRequest, Response = InfoResponse, Error = BoxError>
+        + Send
+        + Sync
+        + Clone
+        + 'static,
     I::Future: Send + 'static,
     S: Service<SnapshotRequest, Response = SnapshotResponse, Error = BoxError>
         + Send
+        + Sync
         + Clone
         + 'static,
     S::Future: Send + 'static,
@@ -142,34 +152,65 @@ where
         ServerBuilder::default()
     }
 
-    pub async fn listen<A: ToSocketAddrs + std::fmt::Debug>(self, addr: A) -> Result<(), BoxError> {
+    pub async fn listen<A: ToSocketAddrs + std::fmt::Debug + Copy + Send + Sync + 'static>(
+        self,
+        addr: A,
+    ) -> Result<(), BoxError> {
         tracing::info!(?addr, "starting ABCI server");
-        let listener = TcpListener::bind(addr).await?;
+        let listener = Arc::new(TcpListener::bind(addr).await?);
         let local_addr = listener.local_addr()?;
         tracing::info!(?local_addr, "bound tcp listener");
 
         loop {
-            match listener.accept().await {
-                Ok((socket, addr)) => {
-                    // set parent: None for the connection span, as it should
-                    // exist independently of the listener's spans.
-                    let span = tracing::span!(parent: None, Level::ERROR, "abci", ?addr);
-                    let conn = Connection {
-                        consensus: self.consensus.clone(),
-                        mempool: self.mempool.clone(),
-                        info: self.info.clone(),
-                        snapshot: self.snapshot.clone(),
-                    };
-                    tokio::spawn(async move { conn.run(socket).await.unwrap() }.instrument(span));
-                }
-                Err(e) => {
-                    tracing::warn!({ %e }, "error accepting new tcp connection");
-                }
-            }
+            // set parent: None for the connection span, as it should
+            // exist independently of the listener's spans.
+            // let span = tracing::span!(parent: None, Level::ERROR, "abci", ?addr);
+
+            let server = self.clone();
+            let listener_clone = listener.clone();
+            tokio::spawn(async move {
+                let s = server.clone();
+                backoff::future::retry::<_, BoxError, _, _, _>(
+                    ExponentialBackoff::default(),
+                    || async {
+                        match listener_clone.accept().await {
+                            Ok((socket, _addr)) => {
+                                let conn = Connection {
+                                    consensus: s.consensus.clone(),
+                                    mempool: s.mempool.clone(),
+                                    info: s.info.clone(),
+                                    snapshot: s.snapshot.clone(),
+                                };
+
+                                if let Err(e) = conn.run(socket).await {
+                                    match e.downcast::<tower::load_shed::error::Overloaded>() {
+                                        Err(e) => {
+                                            tracing::error!({ %e }, "error in a connection handler");
+                                            return Err(backoff::Error::Permanent(e));
+                                        }
+                                        Ok(e) => {
+                                            tracing::warn!("Service overloaded - backing off");
+                                            return Err(backoff::Error::transient(e));
+                                        }
+                                    }
+                                }
+                                Ok(())
+                            }
+                            Err(e) => {
+                                tracing::error!({ %e }, "error accepting new tcp connection");
+                                Ok(())
+                            }
+                        }
+                    },
+                )
+                .await
+            });
+            // .instrument(span);
         }
     }
 }
 
+#[derive(Clone)]
 struct Connection<C, M, I, S> {
     consensus: C,
     mempool: M,