Treat TryRecvError::Closed in Inbound::poll_ready as a fatal error

Also:
- handle errors in service readiness the same as errors in requests

Closes #1655

zebra-network/src/peer/connection.rs

@@ -904,20 +904,25 @@ where
     /// of connected peers.
     async fn drive_peer_request(&mut self, req: Request) {
         trace!(?req);
+        use futures::TryFutureExt;
+        use tokio::sync::oneshot::error::TryRecvError;
         use tower::{load_shed::error::Overloaded, ServiceExt};
 
-        if self.inbound_service.ready_and().await.is_err() {
-            // Treat all service readiness errors as Overloaded
-            // TODO: treat `TryRecvError::Closed` in `Inbound::poll_ready` as a fatal error (#1655)
-            self.fail_with(PeerError::Overloaded);
-            return;
-        }
-
-        let rsp = match self.inbound_service.call(req.clone()).await {
+        let rsp = match self
+            .inbound_service
+            .ready_and()
+            .and_then(|service| service.call(req.clone()))
+            .await
+        {
             Err(e) => {
-                if e.is::<Overloaded>() {
+                if e.is::<TryRecvError>() {
+                    panic!(
+                        "unexpected initialization error in inbound service: {:?}",
+                        e
+                    );
+                } else if e.is::<Overloaded>() {
                     // TODO: rate-limit closing connections during inbound service overload (#2107)
-                    tracing::info!(%req, "inbound service is overloaded, dropping request");
+                    tracing::debug!(%req, "inbound service is overloaded, dropping request");
                     metrics::counter!("pool.overloaded", 1);
                 } else {
                     // We could send a reject to the remote peer, but that might cause