async `protocol.read().await` does not return when underlying connection is closed

[0xB10C]

As demonstrated by 0xB10C@d89df30, protocol.read().await does not err nor return when the underlying connection is closed.

diff --git a/protocol/tests/round_trips.rs b/protocol/tests/round_trips.rs
index 8b07daa..06c144e 100644
--- a/protocol/tests/round_trips.rs
+++ b/protocol/tests/round_trips.rs
@@ -384,7 +384,7 @@ async fn regtest_handshake_async() {
     };
     use tokio::net::TcpStream;

-    let bitcoind = regtest_process(TransportVersion::V2);
+    let mut bitcoind = regtest_process(TransportVersion::V2);

     let stream = TcpStream::connect(bitcoind.params.p2p_socket.unwrap())
         .await
@@ -437,6 +437,25 @@ async fn regtest_handshake_async() {
     assert_eq!(response_message.cmd(), "version");

     println!("Successfully exchanged version messages using async Protocol API!");
+
+    let message = consensus::serialize(&V2NetworkMessage::new(NetworkMessage::Verack));
+    println!("Sending verack message using Protocol::write()");
+    protocol.write(&Payload::genuine(message)).await.unwrap();
+
+    let mut node_up = true;
+    loop {
+        let payload = protocol.read().await.unwrap();
+        let response_message = consensus::deserialize::<V2NetworkMessage>(payload.contents()).unwrap();
+        println!("read {:?}", response_message);
+
+        // Stopping the node here. The next few read()'s should fail?
+        if node_up {
+            node_up = false;
+            bitcoind.stop().unwrap();
+            println!("node is now shutdown!");
+        }
+    }
+
 }

 #[test]

Running this, I always get 100% CPU usage on one core.

Tested against a v30.0 bitcoind.

$ cargo test --features tokio regtest_handshake_async -- --nocapture

running 1 test
Using bitcoind at /nix/store/wbdd2i4vmw1mn25vkzbx0gssa04si9r9-bitcoind-30.0/bin/bitcoind
Starting async BIP-324 handshake
Async handshake completed successfully!
Sending version message using async Protocol::write()
Reading version response using async Protocol::read()
Successfully exchanged version messages using async Protocol API!
Sending verack message using Protocol::write()
read V2NetworkMessage { payload: Verack }
node is now shutdown!
read V2NetworkMessage { payload: SendCmpct(SendCmpct { send_compact: false, version: 2 }) }
read V2NetworkMessage { payload: Ping(17509515947851498945) }
read V2NetworkMessage { payload: FeeFilter(FeeRate(9936506000)) }
test regtest_handshake_async has been running for over 60 seconds
^C

I haven't actually spent time investigating deeper.

[0xB10C]

I think the culprit is not checking if the read()s in ProtocolReader::read() return zero bytes. If they do, this indicates EOF. So a simple fix would be something similar to:

diff --git a/protocol/src/futures.rs b/protocol/src/futures.rs
index b8d6526..e22229b 100644
--- a/protocol/src/futures.rs
+++ b/protocol/src/futures.rs
@@ -406,7 +406,11 @@ where
                     bytes_read,
                 } => {
                     while *bytes_read < NUM_LENGTH_BYTES {
-                        *bytes_read += self.reader.read(&mut length_bytes[*bytes_read..]).await?;
+                        let len = self.reader.read(&mut length_bytes[*bytes_read..]).await?;
+                        if len == 0 {
+                            return Err(std::io::Error::new(std::io::ErrorKind::ConnectionAborted, "read zero bytes").into())
+                        }
+                        *bytes_read += len;
                     }
 
                     let packet_bytes_len = self.inbound_cipher.decrypt_packet_len(*length_bytes);
@@ -417,7 +421,11 @@ where
                     bytes_read,
                 } => {
                     while *bytes_read < packet_bytes.len() {
-                        *bytes_read += self.reader.read(&mut packet_bytes[*bytes_read..]).await?;
+                        let len = self.reader.read(&mut packet_bytes[*bytes_read..]).await?;
+                        if len == 0 {
+                            return Err(std::io::Error::new(std::io::ErrorKind::ConnectionAborted, "read zero bytes").into())
+                        }
+                        *bytes_read += len;
                     }
 
                     let plaintext_len = InboundCipher::decryption_buffer_len(packet_bytes.len());

[rustaceanrob]

Appears to be correct. Would you like to submit that patch as a PR?

[0xB10C]

Opened #160