disconnection fix

src/records/memory/network_mapping/add.rs

@@ -1060,6 +1060,23 @@ impl NodeInfo {
             if !candidate_lost_collision {
                 if let Some(existing_node) = address_map.get_mut(&edit.address) {
                     if existing_node.deleted_timestamp > 0 {
+                        // A revived membership remains pending with its own add
+                        // timestamp as the deletion marker until monitor activation.
+                        // Redundant delivery of that exact signed record is already
+                        // represented locally; acknowledge it without reviving it or
+                        // changing any mapping state. Actual later deletions and
+                        // different records must still pass the ordering check.
+                        if existing_node.deleted_timestamp == existing_node.added_timestamp
+                            && existing_node.deleted_block == 0
+                            && existing_node.monitoring.is_empty()
+                            && existing_node.ip == edit.ip
+                            && existing_node.port == edit.port
+                            && existing_node.added_by == edit.modified_by
+                            && existing_node.added_timestamp == edit.modified_timestamp
+                            && existing_node.added_signature == edit.modified_signature
+                        {
+                            return RpcResponse::Binary(b"Success".to_vec());
+                        }
                         if edit.modified_timestamp <= existing_node.deleted_timestamp {
                             return RpcResponse::Binary(
                                 b"Error: Reconnection record did not follow deletion".to_vec(),
@@ -1140,6 +1157,70 @@ impl NodeInfo {
 mod tests {
     use super::*;
 
+    #[tokio::test]
+    async fn repeated_signed_rejoin_preserves_pending_membership() {
+        use crate::wallets::structures::SavedWallet;
+        let db = sled::Config::new().temporary(true).open().unwrap();
+        let (public, private) = Wallet::generate_keypair();
+        let address_bytes = Wallet::public_key_bytes_to_short_address_bytes(&public).unwrap();
+        let address = Wallet::bytes_to_short_address(&address_bytes).unwrap();
+        crate::records::wallet_registry::register_short_address(&db, &address_bytes, &public).unwrap();
+        let wallet = Arc::new(Wallet {
+            saved: SavedWallet {
+                short_address: address.clone(), vanity_address: None,
+                public_key: crate::encode(public), private_key: private,
+            },
+            encryption_key: String::new(),
+        });
+        let mut old = NodeInfo::new("8.8.8.8".into(), 50050, 0, address.clone(), 50, String::new());
+        old.deleted_timestamp = 100;
+        old.deleted_block = 10;
+        let previous = ADDRESS_MAP.lock().await.insert(address.clone(), old);
+        let edit = SignedNodeEdit {
+            address: address.clone(), ip: "8.8.8.8".into(), port: 50050,
+            modified_by: address.clone(), modified_timestamp: 200,
+            modified_signature: NodeInfo::added_signature(&address, "8.8.8.8", 50050, 200, &wallet).await,
+        };
+        let params = || AddAddressParams {
+            map: Arc::new(Mutex::new(Command::new())), edit: edit.clone(),
+            monitors: Vec::new(), blocks_mined: 0, remote_ip: String::new(),
+            db: db.clone(), wallet: wallet.clone(), connections_key: "8.8.4.4:50050".into(),
+        };
+        let RpcResponse::Binary(first) = NodeInfo::add_address_now(params()).await;
+        let RpcResponse::Binary(repeated) = NodeInfo::add_address_now(params()).await;
+        let mut different = params();
+        different.edit.port = 50051;
+        different.edit.modified_signature = NodeInfo::added_signature(
+            &address, "8.8.8.8", 50051, 200, &wallet,
+        ).await;
+        let RpcResponse::Binary(conflicting) = NodeInfo::add_address_now(different).await;
+        let mut mapping = ADDRESS_MAP.lock().await;
+        let actual = mapping.remove(&address).unwrap();
+        let mut deleted = NodeInfo::new(
+            edit.ip.clone(), edit.port, 0, edit.modified_by.clone(),
+            edit.modified_timestamp, edit.modified_signature.clone(),
+        );
+        deleted.deleted_timestamp = 250;
+        deleted.deleted_block = 20;
+        mapping.insert(address.clone(), deleted);
+        drop(mapping);
+        let RpcResponse::Binary(stale) = NodeInfo::add_address_now(params()).await;
+        let mut mapping = ADDRESS_MAP.lock().await;
+        let after_stale = mapping.remove(&address).unwrap();
+        if let Some(previous) = previous { mapping.insert(address, previous); }
+        drop(mapping);
+        assert_eq!(String::from_utf8(first).unwrap(), "Success");
+        assert_eq!(String::from_utf8(repeated).unwrap(), "Success");
+        assert_eq!(String::from_utf8(conflicting).unwrap(), "Error: Reconnection record did not follow deletion");
+        assert_eq!(String::from_utf8(stale).unwrap(), "Error: Reconnection record did not follow deletion");
+        assert_eq!(after_stale.deleted_timestamp, 250);
+        assert_eq!(after_stale.deleted_block, 20);
+        assert_eq!(actual.added_timestamp, 200);
+        assert_eq!(actual.deleted_timestamp, 200);
+        assert_eq!(actual.deleted_block, 0);
+        assert!(actual.monitoring.is_empty());
+    }
+
     #[test]
     fn full_admission_window_allows_known_wallet_but_not_a_new_wallet_on_same_ip() {
         let mut map = HashMap::new();

src/rpc/read_bytes_from_stream.rs

@@ -11,7 +11,7 @@ use crate::{timeout, AsyncReadExt, AsyncWriteExt, Duration};
 const STREAM_READ_TIMEOUT_SECONDS: u64 = 30;
 
 async fn read_exact_from_stream(
-    _key: &str,
+    key: &str,
     stream_locked: &Arc<Mutex<TcpStream>>,
     stream: &mut TcpStream,
     buffer: &mut [u8],
@@ -26,6 +26,7 @@ async fn read_exact_from_stream(
     {
         Ok(Ok(_)) => Ok(()),
         Ok(Err(err)) => {
+            warn!("[rpc] removing connection after payload read failure: peer={key} requested_bytes={} error={err}", buffer.len());
             if let Err(e) = stream.shutdown().await {
                 warn!("Error shutting down stream: {e}");
             }
@@ -33,6 +34,7 @@ async fn read_exact_from_stream(
             Err(err.to_string())
         }
         Err(_) => {
+            warn!("[rpc] removing connection after payload read timeout: peer={key} requested_bytes={} timeout_seconds={STREAM_READ_TIMEOUT_SECONDS}", buffer.len());
             if let Err(e) = stream.shutdown().await {
                 warn!("Error shutting down stream: {e}");
             }

src/rpc/responses.rs

@@ -10,7 +10,6 @@ use crate::Duration;
 use crate::Mutex;
 use crate::Serialize;
 use crate::TcpStream;
-use std::io::ErrorKind;
 use tokio::sync::Semaphore;
 use tokio::task::JoinSet;
 
@@ -23,17 +22,6 @@ pub enum RpcResponse {
 }
 
 impl RpcResponse {
-    fn is_expected_disconnect(err: &std::io::Error) -> bool {
-        matches!(
-            err.kind(),
-            ErrorKind::BrokenPipe
-                | ErrorKind::ConnectionReset
-                | ErrorKind::ConnectionAborted
-                | ErrorKind::NotConnected
-                | ErrorKind::UnexpectedEof
-        )
-    }
-
     async fn write_bytes(
         stream: &Arc<Mutex<TcpStream>>,
         connections_key: Option<&str>,
@@ -62,9 +50,7 @@ impl RpcResponse {
                 true
             }
             Ok(Err(err)) => {
-                if !Self::is_expected_disconnect(&err) {
-                    warn!("Error sending binary response to {peer}: {err:?}");
-                }
+                warn!("[rpc] removing connection after socket write failure: peer={peer} socket_peer={:?} error={err:?}", stream_guard.peer_addr());
                 let _ = stream_guard.shutdown().await;
                 drop(stream_guard);
                 set_stream_health(stream, ConnectionHealth::Closed).await;

src/rpc/server/command_loop_state.rs

@@ -35,6 +35,7 @@ async fn read_next_command_byte(
         match stream.try_read(&mut buffer) {
             Ok(1) => return Ok(Some(buffer[0])),
             Ok(0) => {
+                warn!("[rpc] command stream reached EOF: socket_peer={:?}", stream.peer_addr());
                 drop(stream);
                 remove_stream_from_memory(stream_locked).await;
                 return Ok(None);
@@ -45,7 +46,7 @@ async fn read_next_command_byte(
                 sleep(timeout_duration).await;
             }
             Err(err) => {
-                warn!("Dropped stream: {err:?}");
+                warn!("[rpc] command read failed: socket_peer={:?} error={err:?}", stream.peer_addr());
                 drop(stream);
                 remove_stream_from_memory(stream_locked).await;
                 return Ok(None);

src/rpc/server/handshake_processing.rs

@@ -182,13 +182,13 @@ pub async fn combine_and_send_data(params: CombineAndSendDataParams) -> Result<(
     }
 
     // start the rpc loop
-    tokio::spawn(start_loop(
-        stream.clone(),
-        db,
-        connections_key.to_string(),
-        wallet,
-        map,
-    ));
+    let loop_stream = stream.clone();
+    let loop_key = connections_key.to_string();
+    tokio::spawn(async move {
+        if let Err(err) = start_loop(loop_stream, db, loop_key.clone(), wallet, map).await {
+            error!("[rpc] command loop exited with error: peer={loop_key} error={err}");
+        }
+    });
 
     Ok(())
 }

src/rpc/server/rpc_command_loop.rs

@@ -555,6 +555,10 @@ pub async fn start_loop(
                 .await?;
                 let should_drop_rejected_miner = client_type == ClientType::Miner
                     && matches!(&result, responses::RpcResponse::Binary(bytes) if String::from_utf8_lossy(bytes).starts_with("Error:"));
+                if should_drop_rejected_miner {
+                    let responses::RpcResponse::Binary(bytes) = &result;
+                    crate::log::warn!("[rpc] removing miner after membership announcement rejection: peer={} reason={}", connections_key, String::from_utf8_lossy(bytes));
+                }
                 result
                     .send(&stream_locked, Some(&connections_key), uid)
                     .await;