startup connection handling fixes

src/rpc/client/block_hash_vote.rs

@@ -1,5 +1,7 @@
 use crate::encode;
-use crate::records::memory::response_channels::{reserve_entry_with_context, Command};
+use crate::records::memory::response_channels::{
+    delete_entry, reserve_entry_with_context, Command,
+};
 use crate::rpc::command_maps::RPC_BLOCK_HASH_AT_HEIGHT;
 use crate::rpc::responses::RpcResponse;
 use crate::{timeout, Arc, Duration, Mutex, TcpStream};
@@ -11,7 +13,7 @@ pub async fn request_block_hash_at_height(
     height: u32,
 ) -> Result<String, String> {
     let (hashmap_key, _tx, rx) = reserve_entry_with_context(
-        map,
+        map.clone(),
         Some(RPC_BLOCK_HASH_AT_HEIGHT),
         Some(connections_key.clone()),
     )
@@ -21,7 +23,12 @@ pub async fn request_block_hash_at_height(
     message.extend_from_slice(&hashmap_key);
     message.extend_from_slice(&height.to_le_bytes());
 
-    RpcResponse::send_raw(&stream, Some(&connections_key), &message).await;
+    if !RpcResponse::send_raw(&stream, Some(&connections_key), &message).await {
+        delete_entry(map, hashmap_key).await;
+        return Err(format!(
+            "Failed to send block hash request at height {height}"
+        ));
+    }
 
     let mut rx = rx.lock().await;
     let buffer = timeout(Duration::from_secs(5), rx.recv())

src/rpc/server/handshake.rs

@@ -58,6 +58,16 @@ async fn get_connection_counts() -> (u8, u8) {
     (incoming, outgoing)
 }
 
+async fn incoming_setup_stream_is_current(
+    connections_key: &str,
+    stream: &Arc<Mutex<TcpStream>>,
+) -> bool {
+    Connection::get_stream_from_memory(connections_key)
+        .await
+        .map(|current| Arc::ptr_eq(&current, stream))
+        .unwrap_or(false)
+}
+
 async fn sync_incoming_peer_before_operational(
     stream: Arc<Mutex<TcpStream>>,
     db: &Db,
@@ -225,6 +235,12 @@ async fn complete_incoming_miner_setup(
         remove_stream_from_memory(&stream).await;
         return;
     }
+    if !incoming_setup_stream_is_current(connections_key, &stream).await {
+        warn!(
+            "[startup] incoming peer setup stopped because its stream was removed or replaced: peer={connections_key}"
+        );
+        return;
+    }
     mark_peer_wallet_registry_synced(connections_key).await;
 
     if let Err(err) = register_connected_wallet(
@@ -240,6 +256,12 @@ async fn complete_incoming_miner_setup(
         remove_stream_from_memory(&stream).await;
         return;
     }
+    if !incoming_setup_stream_is_current(connections_key, &stream).await {
+        warn!(
+            "[startup] incoming peer setup stopped because its stream was removed or replaced: peer={connections_key}"
+        );
+        return;
+    }
 
     let remote_height =
         match request_remote_height(stream.clone(), map.clone(), connections_key.to_string()).await
@@ -253,6 +275,12 @@ async fn complete_incoming_miner_setup(
                 return;
             }
         };
+    if !incoming_setup_stream_is_current(connections_key, &stream).await {
+        warn!(
+            "[startup] incoming peer setup stopped because its stream was removed or replaced: peer={connections_key}"
+        );
+        return;
+    }
     if let Err(err) = ensure_compatible_genesis(
         stream.clone(),
         map.clone(),