Merge branch 'develop' into deps/tokio

abci/src/server/codec.rs

@@ -97,6 +97,18 @@ impl<'a> Codec {
 
         Self::process_worker_queues(codec, request_tx, response_rx, cancel).await;
     }
+
+    /// Worker that moves messages between codec and requests and responses
+    /// queues.
+    ///
+    /// Reads messages from ABCI Client from `codec`, sends them to Tenderdash
+    /// via `request_tx`, receives Tenderdash responses from `response_rx`
+    /// and forwards to
+    ///
+    /// ## Error handling
+    ///
+    /// On error, it cancels the `cancel` [CancellationToken] and exits.
+    /// It causes `response_rx` to be closed.
     async fn process_worker_queues<L: AsyncRead + AsyncWrite + Unpin>(
         mut codec: Framed<L, Coder>,
         request_tx: Sender<proto::abci::Request>,
@@ -149,9 +161,10 @@ impl<'a> Codec {
     }
 
     pub fn send(&self, value: Response) -> Result<(), Error> {
-        self.response_tx
-            .blocking_send(value)
-            .map_err(|e| Error::Async(e.to_string()))
+        self.response_tx.blocking_send(value).map_err(|_| {
+            // channel closed, `process_worker_queues` either errored or cancelled
+            Error::Cancelled()
+        })
     }
 }
 

abci/tests/kvstore.rs

@@ -63,17 +63,21 @@ fn test_kvstore() {
     fs::set_permissions(SOCKET, perms).expect("set perms");
 
     let socket_uri = bind_address.to_string();
-    let _td = common::docker::TenderdashDocker::new("tenderdash", None, &socket_uri);
+    let td = common::docker::TenderdashDocker::new("tenderdash", None, &socket_uri);
 
+    let next_client = server.next_client();
+    tracing::debug!(?next_client, "next client");
     assert!(matches!(
-        server.next_client(),
+        next_client,
         Err(tenderdash_abci::Error::Cancelled())
     ));
     drop(server);
 
     let kvstore_app = kvstore.into_inner().expect("kvstore lock is poisoned");
     assert_eq!(kvstore_app.persisted_state, state_reference);
     assert_eq!(kvstore_app.last_block_height, 1);
+
+    drop(td);
 }
 
 /// An example storage.