Prevent server from exiting on ECONNABORTED

On FreeBSD, accept can fail with ECONNABORTED, which means that
"A connection arrived, but it was on the listen queue"
(see `man 2 accept`).

Without this patch, a server without the tls feature  exits returing 0
in this case, which makes it vulnerable not only to intentional denial
of service, but also to unintentional crashes, e.g., by haproxy TCP
health checks.

The problem can be reproduced on any FreeBSD system by running the
tonic "helloworld" example (without feature TLS) and then sending a
packet using nmap:

    cd examples
    cargo run --bin helloworld-server --no-default-features &
    nmap -sT -p 50051 -6 ::1
    # server exits

When running the example with the feature tls enabled, it won't exit
(as the tls event loop in tonic/src/transport/server/incoming.rs
handles errors gracefully):

    cd examples
    cargo run --bin helloworld-server --no-default-features \
      features=tls &
    nmap -sT -p 50051 -6 ::1
    # server keeps running

This patch is not optimal - it removes some generic error parameters
to gain access to `std::io::Error::kind()`. The logic itself should be
sound.

See also:

- https://man.freebsd.org/cgi/man.cgi?accept(2)
  Accept man page
- giampaolo/pyftpdlib#105
  giampaolo/pyftpdlib@0f82232
  Basically the same issue (and its fix) in another project

tonic/src/transport/server/incoming.rs

@@ -16,17 +16,22 @@ use tokio_stream::{Stream, StreamExt};
 use tracing::warn;
 
 #[cfg(not(feature = "tls"))]
-pub(crate) fn tcp_incoming<IO, IE>(
-    incoming: impl Stream<Item = Result<IO, IE>>,
+pub(crate) fn tcp_incoming<IO>(
+    incoming: impl Stream<Item = Result<IO, std::io::Error>>,
 ) -> impl Stream<Item = Result<ServerIo<IO>, crate::Error>>
 where
     IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
-    IE: Into<crate::Error>,
 {
     async_stream::try_stream! {
         let mut incoming = pin!(incoming);
 
         while let Some(item) = incoming.next().await {
+            if let Some(e) = item.as_ref().err() {
+                   if e.kind() == std::io::ErrorKind::ConnectionAborted {
+                       tracing::debug!(message = e.to_string(), error = %e);
+                       continue;
+                   }
+            }
             yield item.map(ServerIo::new_io)?
         }
     }

tonic/src/transport/server/mod.rs

@@ -487,7 +487,7 @@ impl<L> Server<L> {
         }
     }
 
-    pub(crate) async fn serve_with_shutdown<S, I, F, IO, IE, ResBody>(
+    pub(crate) async fn serve_with_shutdown<S, I, F, IO, ResBody>(
         self,
         svc: S,
         incoming: I,
@@ -500,10 +500,9 @@ impl<L> Server<L> {
         <<L as Layer<S>>::Service as Service<Request<BoxBody>>>::Future: Send + 'static,
         <<L as Layer<S>>::Service as Service<Request<BoxBody>>>::Error:
             Into<crate::Error> + Send + 'static,
-        I: Stream<Item = Result<IO, IE>>,
+        I: Stream<Item = Result<IO, std::io::Error>>,
         IO: AsyncRead + AsyncWrite + Connected + Unpin + Send + 'static,
         IO::ConnectInfo: Clone + Send + Sync + 'static,
-        IE: Into<crate::Error>,
         F: Future<Output = ()>,
         ResBody: http_body::Body<Data = Bytes> + Send + 'static,
         ResBody::Error: Into<crate::Error>,
@@ -733,7 +732,7 @@ impl<L> Router<L> {
         let incoming = TcpIncoming::new(addr, self.server.tcp_nodelay, self.server.tcp_keepalive)
             .map_err(super::Error::from_source)?;
         self.server
-            .serve_with_shutdown::<_, _, future::Ready<()>, _, _, ResBody>(
+            .serve_with_shutdown::<_, _, future::Ready<()>, _, ResBody>(
                 self.routes.prepare(),
                 incoming,
                 None,
@@ -775,15 +774,11 @@ impl<L> Router<L> {
     /// This method discards any provided [`Server`] TCP configuration.
     ///
     /// [`Server`]: struct.Server.html
-    pub async fn serve_with_incoming<I, IO, IE, ResBody>(
-        self,
-        incoming: I,
-    ) -> Result<(), super::Error>
+    pub async fn serve_with_incoming<I, IO, ResBody>(self, incoming: I) -> Result<(), super::Error>
     where
-        I: Stream<Item = Result<IO, IE>>,
+        I: Stream<Item = Result<IO, std::io::Error>>,
         IO: AsyncRead + AsyncWrite + Connected + Unpin + Send + 'static,
         IO::ConnectInfo: Clone + Send + Sync + 'static,
-        IE: Into<crate::Error>,
         L: Layer<Routes>,
         L::Service:
             Service<Request<BoxBody>, Response = Response<ResBody>> + Clone + Send + 'static,
@@ -794,7 +789,7 @@ impl<L> Router<L> {
         ResBody::Error: Into<crate::Error>,
     {
         self.server
-            .serve_with_shutdown::<_, _, future::Ready<()>, _, _, ResBody>(
+            .serve_with_shutdown::<_, _, future::Ready<()>, _, ResBody>(
                 self.routes.prepare(),
                 incoming,
                 None,
@@ -810,16 +805,15 @@ impl<L> Router<L> {
     /// This method discards any provided [`Server`] TCP configuration.
     ///
     /// [`Server`]: struct.Server.html
-    pub async fn serve_with_incoming_shutdown<I, IO, IE, F, ResBody>(
+    pub async fn serve_with_incoming_shutdown<I, IO, F, ResBody>(
         self,
         incoming: I,
         signal: F,
     ) -> Result<(), super::Error>
     where
-        I: Stream<Item = Result<IO, IE>>,
+        I: Stream<Item = Result<IO, std::io::Error>>,
         IO: AsyncRead + AsyncWrite + Connected + Unpin + Send + 'static,
         IO::ConnectInfo: Clone + Send + Sync + 'static,
-        IE: Into<crate::Error>,
         F: Future<Output = ()>,
         L: Layer<Routes>,
         L::Service: