diff --git a/src/records/memory/network_mapping/add.rs b/src/records/memory/network_mapping/add.rs index 8f81838..057a045 100644 --- a/src/records/memory/network_mapping/add.rs +++ b/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(); diff --git a/src/rpc/read_bytes_from_stream.rs b/src/rpc/read_bytes_from_stream.rs index 06ad6f0..d9aba30 100644 --- a/src/rpc/read_bytes_from_stream.rs +++ b/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>, 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}"); } diff --git a/src/rpc/responses.rs b/src/rpc/responses.rs index 6c0f665..6721864 100644 --- a/src/rpc/responses.rs +++ b/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>, 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; diff --git a/src/rpc/server/command_loop_state.rs b/src/rpc/server/command_loop_state.rs index 0d3d9e9..ae50da3 100644 --- a/src/rpc/server/command_loop_state.rs +++ b/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); diff --git a/src/rpc/server/handshake_processing.rs b/src/rpc/server/handshake_processing.rs index 850af98..ce0c5fb 100644 --- a/src/rpc/server/handshake_processing.rs +++ b/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(()) } diff --git a/src/rpc/server/rpc_command_loop.rs b/src/rpc/server/rpc_command_loop.rs index 00ac447..9598c8f 100644 --- a/src/rpc/server/rpc_command_loop.rs +++ b/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;