Test: Revert "Refactor connection.rs to make fail_with errors impossible"

zebra-network/src/peer.rs

@@ -15,6 +15,7 @@ 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,21 +6,22 @@ use std::{
 
 use futures::{
     channel::{mpsc, oneshot},
-    ready,
+    future, ready,
     stream::{Stream, StreamExt},
 };
 use tower::Service;
 
 use crate::protocol::internal::{Request, Response};
 
-use super::{PeerError, SharedPeerError};
+use super::{ErrorSlot, 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`.
@@ -46,6 +47,8 @@ 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 {
@@ -95,6 +98,13 @@ 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;
 
@@ -189,7 +199,10 @@ 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(PeerError::ConnectionClosed.into()))
+            Poll::Ready(Err(self
+                .error_slot
+                .try_get_error()
+                .expect("failed servers must set their error slot")))
         } else {
             Poll::Ready(Ok(()))
         }
@@ -208,7 +221,13 @@ impl Service<Request> for Client {
         match self.server_tx.try_send(ClientRequest { request, span, tx }) {
             Err(e) => {
                 if e.is_disconnected() {
-                    async { Err(PeerError::ConnectionClosed.into()) }.boxed()
+                    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()
                 } else {
                     // sending fails when there's not enough
                     // channel space, but we called poll_ready