Re-Apply "Refactor connection.rs to make fail_with errors impossible" (double revert)

zebra-network/src/peer.rs

@@ -15,7 +15,6 @@ use client::ClientRequest;
 use client::ClientRequestReceiver;
 use client::InProgressClientRequest;
 use client::MustUseOneshotSender;
-use error::ErrorSlot;
 
 pub use client::Client;
 pub use connection::Connection;

zebra-network/src/peer/client.rs

@@ -6,22 +6,21 @@ use std::{
 
 use futures::{
     channel::{mpsc, oneshot},
-    future, ready,
+    ready,
     stream::{Stream, StreamExt},
 };
 use tower::Service;
 
 use crate::protocol::internal::{Request, Response};
 
-use super::{ErrorSlot, PeerError, SharedPeerError};
+use super::{PeerError, SharedPeerError};
 
 /// The "client" duplex half of a peer connection.
 pub struct Client {
     // Used to shut down the corresponding heartbeat.
     // This is always Some except when we take it on drop.
     pub(super) shutdown_tx: Option<oneshot::Sender<()>>,
     pub(super) server_tx: mpsc::Sender<ClientRequest>,
-    pub(super) error_slot: ErrorSlot,
 }
 
 /// A message from the `peer::Client` to the `peer::Server`.
@@ -47,8 +46,6 @@ pub(super) struct ClientRequestReceiver {
 
 /// A message from the `peer::Client` to the `peer::Server`,
 /// after it has been received by the `peer::Server`.
-///
-///
 #[derive(Debug)]
 #[must_use = "tx.send() must be called before drop"]
 pub(super) struct InProgressClientRequest {
@@ -98,13 +95,6 @@ impl From<ClientRequest> for InProgressClientRequest {
     }
 }
 
-impl ClientRequestReceiver {
-    /// Forwards to `inner.close()`
-    pub fn close(&mut self) {
-        self.inner.close()
-    }
-}
-
 impl Stream for ClientRequestReceiver {
     type Item = InProgressClientRequest;
 
@@ -199,10 +189,7 @@ impl Service<Request> for Client {
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         if ready!(self.server_tx.poll_ready(cx)).is_err() {
-            Poll::Ready(Err(self
-                .error_slot
-                .try_get_error()
-                .expect("failed servers must set their error slot")))
+            Poll::Ready(Err(PeerError::ConnectionClosed.into()))
         } else {
             Poll::Ready(Ok(()))
         }
@@ -221,13 +208,7 @@ impl Service<Request> for Client {
         match self.server_tx.try_send(ClientRequest { request, span, tx }) {
             Err(e) => {
                 if e.is_disconnected() {
-                    let ClientRequest { tx, .. } = e.into_inner();
-                    let _ = tx.send(Err(PeerError::ConnectionClosed.into()));
-                    future::ready(Err(self
-                        .error_slot
-                        .try_get_error()
-                        .expect("failed servers must set their error slot")))
-                    .boxed()
+                    async { Err(PeerError::ConnectionClosed.into()) }.boxed()
                 } else {
                     // sending fails when there's not enough
                     // channel space, but we called poll_ready