diff --git a/src/bin/lookup_online_node_count.rs b/src/bin/lookup_online_node_count.rs index c760104..7d2c828 100644 --- a/src/bin/lookup_online_node_count.rs +++ b/src/bin/lookup_online_node_count.rs @@ -56,8 +56,7 @@ fn online_node_count(response: &[u8]) -> Result { .map_err(|_| "node deletion timestamp was invalid")?, ); let monitor_count = u16::from_le_bytes( - response[offset + NODE_MONITOR_COUNT_OFFSET - ..offset + NODE_MONITOR_COUNT_OFFSET + 2] + response[offset + NODE_MONITOR_COUNT_OFFSET..offset + NODE_MONITOR_COUNT_OFFSET + 2] .try_into() .map_err(|_| "node monitor count was invalid")?, ) as usize; diff --git a/src/records/memory/connections.rs b/src/records/memory/connections.rs index b97c0a5..7e822a5 100644 --- a/src/records/memory/connections.rs +++ b/src/records/memory/connections.rs @@ -6,7 +6,7 @@ use crate::records::memory::network_mapping::monitor::{MONITOR_ACTION_ADD, MONIT use crate::records::memory::network_mapping::structs::{MonitorAddressParams, SignedMonitorEdit}; use crate::records::memory::network_mapping::NodeInfo; use crate::records::memory::response_channels::{ - delete_entry, reserve_entry_with_context, retire_peer_entries, Command, + delete_entry, reserve_entry_with_context, Command, }; use crate::records::memory::structs::{Connection, ConnectionHealth, StoreConnectionParams}; use crate::rpc::client::handshake::connect_and_handshake; @@ -131,13 +131,6 @@ pub(crate) fn spawn_outage_recovery_from_runtime() { spawn_outage_recovery(context.db, context.wallet, context.map); } -pub(crate) async fn retire_peer_requests_from_runtime(peer: &str) { - let Some(context) = node_runtime_context() else { - return; - }; - retire_peer_entries(context.map, peer).await; -} - async fn reconnect_replacement_inner(excluded_ip: &str) { // When an outgoing peer disappears, try to replace it with another // active node that is not already connected and is not the failed IP. @@ -206,7 +199,7 @@ async fn retry_dropped_outgoing(ip: String, port: u16) { // Losing every miner peer is process-level outage recovery, not an // independent retry of each socket that happened to time out. - if peer_connection_count().await == 0 { + if miner_connection_count().await == 0 { spawn_outage_recovery(context.db, context.wallet, context.map); return; } @@ -532,7 +525,6 @@ impl Connection { } } drop(guard); - retire_peer_entries(command_map.clone(), &format!("{ip}:{port}")).await; if connection_type == ConnectionType::Outgoing { spawn_retry_dropped_outgoing(ip.clone(), port); } @@ -576,13 +568,20 @@ impl Connection { .count() } - pub fn mark_wallet_registry_synced(&mut self, key: &str) -> bool { + pub fn mark_wallet_registry_synced( + &mut self, + key: &str, + expected_stream: &Arc>, + ) -> bool { let Some((ip, port)) = split_ip_port_key(key) else { return false; }; let ip_bytes = ip_to_binary(&ip); for (connection_key, info) in self.connection_map.iter_mut() { - if connection_key.ip == ip_bytes && connection_key.port == port { + if connection_key.ip == ip_bytes + && connection_key.port == port + && Arc::ptr_eq(&info.stream, expected_stream) + { info.wallet_registry_synced = true; return true; } @@ -590,13 +589,20 @@ impl Connection { false } - pub fn mark_network_map_synced(&mut self, key: &str) -> bool { + pub fn mark_network_map_synced( + &mut self, + key: &str, + expected_stream: &Arc>, + ) -> bool { let Some((ip, port)) = split_ip_port_key(key) else { return false; }; let ip_bytes = ip_to_binary(&ip); for (connection_key, info) in self.connection_map.iter_mut() { - if connection_key.ip == ip_bytes && connection_key.port == port { + if connection_key.ip == ip_bytes + && connection_key.port == port + && Arc::ptr_eq(&info.stream, expected_stream) + { info.network_map_synced = true; return true; } @@ -638,6 +644,7 @@ impl Connection { pub fn mark_local_setup_complete( &mut self, key: &str, + expected_stream: &Arc>, command_map: Arc>, ) -> Option<(Arc>, bool)> { let Some((ip, port)) = split_ip_port_key(key) else { @@ -645,7 +652,10 @@ impl Connection { }; let ip_bytes = ip_to_binary(&ip); for (connection_key, info) in self.connection_map.iter_mut() { - if connection_key.ip == ip_bytes && connection_key.port == port { + if connection_key.ip == ip_bytes + && connection_key.port == port + && Arc::ptr_eq(&info.stream, expected_stream) + { if ClientType::from_bytes(&info.client_type) != Some(ClientType::Miner) || !info.wallet_registry_synced || !info.network_map_synced @@ -660,9 +670,27 @@ impl Connection { None } + pub fn local_setup_acknowledged( + &self, + key: &str, + expected_stream: &Arc>, + ) -> bool { + let Some((ip, port)) = split_ip_port_key(key) else { + return false; + }; + let ip_bytes = ip_to_binary(&ip); + self.connection_map.iter().any(|(connection_key, info)| { + connection_key.ip == ip_bytes + && connection_key.port == port + && Arc::ptr_eq(&info.stream, expected_stream) + && info.local_setup_acknowledged + }) + } + pub fn mark_remote_setup_complete( &mut self, key: &str, + expected_stream: &Arc>, command_map: Arc>, ) -> bool { let Some((ip, port)) = split_ip_port_key(key) else { @@ -670,7 +698,10 @@ impl Connection { }; let ip_bytes = ip_to_binary(&ip); for (connection_key, info) in self.connection_map.iter_mut() { - if connection_key.ip == ip_bytes && connection_key.port == port { + if connection_key.ip == ip_bytes + && connection_key.port == port + && Arc::ptr_eq(&info.stream, expected_stream) + { info.remote_setup_complete = true; return Self::finalize_setup_if_complete(info, ip, port, command_map); } @@ -681,6 +712,7 @@ impl Connection { pub fn mark_local_setup_acknowledged( &mut self, key: &str, + expected_stream: &Arc>, command_map: Arc>, ) -> bool { let Some((ip, port)) = split_ip_port_key(key) else { @@ -688,7 +720,10 @@ impl Connection { }; let ip_bytes = ip_to_binary(&ip); for (connection_key, info) in self.connection_map.iter_mut() { - if connection_key.ip == ip_bytes && connection_key.port == port { + if connection_key.ip == ip_bytes + && connection_key.port == port + && Arc::ptr_eq(&info.stream, expected_stream) + { info.local_setup_acknowledged = true; return Self::finalize_setup_if_complete(info, ip, port, command_map); } @@ -1233,21 +1268,21 @@ pub async fn miner_connection_count() -> usize { .unwrap_or(0) } -pub async fn mark_peer_wallet_registry_synced(key: &str) -> bool { +pub async fn mark_peer_wallet_registry_synced(key: &str, stream: &Arc>) -> bool { CONNECTIONS .write() .await .as_mut() - .map(|connection| connection.mark_wallet_registry_synced(key)) + .map(|connection| connection.mark_wallet_registry_synced(key, stream)) .unwrap_or(false) } -pub async fn mark_peer_network_map_synced(key: &str) -> bool { +pub async fn mark_peer_network_map_synced(key: &str, stream: &Arc>) -> bool { CONNECTIONS .write() .await .as_mut() - .map(|connection| connection.mark_network_map_synced(key)) + .map(|connection| connection.mark_network_map_synced(key, stream)) .unwrap_or(false) } @@ -1300,7 +1335,11 @@ pub async fn peer_catch_up_target(key: &str) -> Option { }) } -pub async fn mark_peer_operational(key: &str, map: Arc>) -> bool { +pub async fn mark_peer_operational( + key: &str, + stream: &Arc>, + map: Arc>, +) -> bool { if peer_is_operational(key).await { return true; } @@ -1309,11 +1348,18 @@ pub async fn mark_peer_operational(key: &str, map: Arc>) -> bool .write() .await .as_mut() - .and_then(|connection| connection.mark_local_setup_complete(key, map.clone())); - let Some((stream, _already_acknowledged)) = setup else { + .and_then(|connection| connection.mark_local_setup_complete(key, stream, map.clone())); + let Some((stored_stream, already_acknowledged)) = setup else { return false; }; + // The peer already acknowledged our setup-complete message. It may still + // be performing a long chain sync before sending its own setup-complete + // message, so do not repeatedly send setup or treat that wait as failure. + if already_acknowledged { + return peer_is_operational(key).await; + } + // Setup completion is idempotent. Keep sending it while the peer exists // until both sides have received and acknowledged the other's setup state. // A previous local acknowledgement alone does not make the peer operational. @@ -1328,7 +1374,7 @@ pub async fn mark_peer_operational(key: &str, map: Arc>) -> bool message.push(RPC_SETUP_COMPLETE); message.extend_from_slice(&uid); - if !RpcResponse::send_raw(&stream, Some(key), &message).await { + if !RpcResponse::send_raw(&stored_stream, Some(key), &message).await { delete_entry(map.clone(), uid).await; continue; } @@ -1343,7 +1389,9 @@ pub async fn mark_peer_operational(key: &str, map: Arc>) -> bool .write() .await .as_mut() - .map(|connection| connection.mark_local_setup_acknowledged(key, map.clone())) + .map(|connection| { + connection.mark_local_setup_acknowledged(key, stream, map.clone()) + }) .unwrap_or(false); if operational { return true; @@ -1358,31 +1406,69 @@ pub async fn mark_peer_operational(key: &str, map: Arc>) -> bool false } -pub async fn mark_peer_remote_setup_complete(key: &str, map: Arc>) -> bool { +pub async fn mark_peer_remote_setup_complete( + key: &str, + stream: &Arc>, + map: Arc>, +) -> bool { CONNECTIONS .write() .await .as_mut() - .map(|connection| connection.mark_remote_setup_complete(key, map)) + .map(|connection| connection.mark_remote_setup_complete(key, stream, map)) .unwrap_or(false) } -pub fn spawn_peer_setup_retry(key: String, map: Arc>) { +pub fn spawn_peer_setup_retry( + key: String, + stream: Arc>, + map: Arc>, +) { tokio::spawn(async move { - let Some(_retry_guard) = try_start_setup_retry(key.clone()) else { + let retry_key = format!("{key}:{:p}", Arc::as_ptr(&stream)); + let Some(_retry_guard) = try_start_setup_retry(retry_key) else { return; }; - loop { + for attempt in 1..=3 { sleep(Duration::from_secs(10)).await; - if Connection::get_stream_from_memory(&key).await.is_none() + let current_stream = Connection::get_stream_from_memory(&key).await; + if current_stream + .as_ref() + .map(|current| !Arc::ptr_eq(current, &stream)) + .unwrap_or(true) || peer_is_operational(&key).await { - break; + return; } - if mark_peer_operational(&key, map.clone()).await { - break; + if mark_peer_operational(&key, &stream, map.clone()).await { + return; } + let local_setup_acknowledged = CONNECTIONS + .read() + .await + .as_ref() + .map(|connection| connection.local_setup_acknowledged(&key, &stream)) + .unwrap_or(false); + if local_setup_acknowledged { + // The connection remains valid while the remote peer finishes + // synchronization. Its eventual setup-complete message will + // finish promotion without another local retry. + return; + } + warn!("[handshake_state] peer setup remains incomplete: peer={key} retry={attempt}/3"); + } + + if !peer_is_operational(&key).await + && Connection::get_stream_from_memory(&key) + .await + .map(|current| Arc::ptr_eq(¤t, &stream)) + .unwrap_or(false) + { + warn!( + "[handshake_state] removing incomplete peer after bounded setup retries: peer={key}" + ); + remove_stream_from_memory(&stream).await; } }); } @@ -1599,4 +1685,32 @@ mod tests { assert!(try_start_setup_retry("test-setup:first".to_string()).is_some()); drop(second); } + + #[tokio::test] + async fn setup_progress_can_only_be_recorded_by_the_stored_stream() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = listener.local_addr().unwrap(); + let client = tokio::net::TcpStream::connect(endpoint).await.unwrap(); + let (server, _) = listener.accept().await.unwrap(); + let stored_stream = Arc::new(Mutex::new(server)); + let stale_stream = Arc::new(Mutex::new(client)); + let mut connections = Connection::new(); + let command_map = Arc::new(Mutex::new(Command::new())); + let key = format!("127.0.0.1:{}", endpoint.port()); + + assert!(connections.store_connection(StoreConnectionParams { + connection_type: ConnectionType::Incoming, + ip: "127.0.0.1".to_string(), + port: endpoint.port(), + stream: stored_stream.clone(), + client_type: ClientType::Miner, + wallet_short_address: "ab13318c26250b048db92920a80a86127c933b0c.cltc".to_string(), + command_map: command_map.clone(), + })); + + assert!(!connections.mark_wallet_registry_synced(&key, &stale_stream)); + assert!(!connections.mark_network_map_synced(&key, &stale_stream)); + assert!(connections.mark_wallet_registry_synced(&key, &stored_stream)); + assert!(connections.mark_network_map_synced(&key, &stored_stream)); + } } diff --git a/src/records/memory/network_mapping/mined_counts.rs b/src/records/memory/network_mapping/mined_counts.rs index fbb3cc2..2c06fcc 100644 --- a/src/records/memory/network_mapping/mined_counts.rs +++ b/src/records/memory/network_mapping/mined_counts.rs @@ -1,5 +1,6 @@ use super::*; use crate::common::check_genesis::genesis_checkup; +use crate::miner::flag::is_normal_mode; use crate::records::unpack_block::unpack_header::load_block_header; use crate::sled::Batch; @@ -65,6 +66,13 @@ impl NodeInfo { }; if let Some(count) = count { + // Startup sync and orphan replay can accept thousands of blocks in + // seconds. Keep those counts in memory and persist the complete + // cache once the operation finishes instead of forcing a Sled + // write for every historical block. + if !is_normal_mode() { + return Ok(()); + } let mut batch = Batch::default(); batch.insert(address.as_bytes(), &[count]); batch.insert(MINED_COUNTS_HEIGHT_KEY, &get_height(db).to_le_bytes()); diff --git a/src/records/memory/network_mapping/monitor.rs b/src/records/memory/network_mapping/monitor.rs index 59e4638..4e979a9 100644 --- a/src/records/memory/network_mapping/monitor.rs +++ b/src/records/memory/network_mapping/monitor.rs @@ -65,6 +65,35 @@ mod tests { } } + #[test] + fn reconciliation_excludes_events_involving_deleted_nodes() { + let mut map = HashMap::new(); + map.insert("active-a".to_string(), node(&[])); + map.insert("active-b".to_string(), node(&[])); + let mut deleted = node(&[]); + deleted.deleted_timestamp = 100; + map.insert("deleted".to_string(), deleted); + + let event = |monitored: &str, monitoring: &str| SignedMonitorEdit { + action: MONITOR_ACTION_ADD, + monitored_address: monitored.to_string(), + monitoring_address: monitoring.to_string(), + target_ip: "1.2.3.4".to_string(), + modified_timestamp: 50, + modified_block: 10, + modified_signature: "signature".to_string(), + }; + let events = HashMap::from([ + ("active".to_string(), event("active-a", "active-b")), + ("deleted-target".to_string(), event("deleted", "active-b")), + ("deleted-signer".to_string(), event("active-a", "deleted")), + ]); + + let reconciled = NodeInfo::reconcilable_monitor_events_from(&map, &events); + assert_eq!(reconciled.len(), 1); + assert!(reconciled.contains_key("active")); + } + #[test] fn remove_wins_an_equal_timestamp_monitor_race() { let add = SignedMonitorEdit { @@ -186,6 +215,24 @@ mod tests { } impl NodeInfo { + pub(super) fn reconcilable_monitor_events_from( + address_map: &HashMap, + event_state: &HashMap, + ) -> HashMap { + event_state + .iter() + .filter(|(_, event)| { + address_map + .get(&event.monitored_address) + .is_some_and(|node| node.deleted_timestamp == 0) + && address_map + .get(&event.monitoring_address) + .is_some_and(|node| node.deleted_timestamp == 0) + }) + .map(|(key, event)| (key.clone(), event.clone())) + .collect() + } + pub async fn monitor_signature( action: u8, monitored_address: &str, @@ -579,7 +626,9 @@ impl NodeInfo { } pub(crate) async fn signed_monitor_state_bytes() -> Vec { + let map = ADDRESS_MAP.lock().await; let state = MONITOR_EVENT_STATE.lock().await; + let state = Self::reconcilable_monitor_events_from(&map, &state); Self::signed_monitor_state_bytes_from(&state) } diff --git a/src/records/memory/network_mapping/queries.rs b/src/records/memory/network_mapping/queries.rs index 48de7d6..f21da3f 100644 --- a/src/records/memory/network_mapping/queries.rs +++ b/src/records/memory/network_mapping/queries.rs @@ -95,6 +95,7 @@ impl NodeInfo { pub(crate) async fn signed_mapping_state() -> (Vec, Vec) { let map = ADDRESS_MAP.lock().await; let monitor_events = MONITOR_EVENT_STATE.lock().await; + let monitor_events = Self::reconcilable_monitor_events_from(&map, &monitor_events); let mut memberships: Vec = map .iter() @@ -221,6 +222,7 @@ impl NodeInfo { pub async fn canonical_mapping_digest() -> Result, String> { let map = ADDRESS_MAP.lock().await; let monitor_events = MONITOR_EVENT_STATE.lock().await; + let monitor_events = Self::reconcilable_monitor_events_from(&map, &monitor_events); let mut canonical = Self::canonical_mapping_records(&map)?; let monitor_state = Self::signed_monitor_state_bytes_from(&monitor_events); canonical.extend_from_slice(&(monitor_state.len() as u32).to_le_bytes()); @@ -425,6 +427,7 @@ impl NodeInfo { // node to validate blocks and avoid dialing deleted peers. let map = ADDRESS_MAP.lock().await; let monitor_events = MONITOR_EVENT_STATE.lock().await; + let monitor_events = Self::reconcilable_monitor_events_from(&map, &monitor_events); let mut mapping: Vec = Vec::with_capacity(map.len() * NODE_RECORD_FIXED_BYTES); for (address, node_info) in map.iter() { diff --git a/src/rpc/client/handshake.rs b/src/rpc/client/handshake.rs index ff94f67..69a9612 100644 --- a/src/rpc/client/handshake.rs +++ b/src/rpc/client/handshake.rs @@ -10,14 +10,36 @@ use crate::IpAddr; use crate::SocketAddr; use crate::TcpStream; use crate::{sleep, timeout, AsyncReadExt, AsyncWriteExt, Duration}; +use std::sync::atomic::{AtomicUsize, Ordering}; use tokio::net::TcpSocket; const OUTBOUND_CONNECT_TIMEOUT_SECONDS: u64 = 2; const HANDSHAKE_RESPONSE_TIMEOUT_SECONDS: u64 = 2; +static OUTBOUND_HANDSHAKES_ACTIVE: AtomicUsize = AtomicUsize::new(0); + +struct OutboundHandshakeGuard; + +impl OutboundHandshakeGuard { + fn acquire() -> Self { + OUTBOUND_HANDSHAKES_ACTIVE.fetch_add(1, Ordering::SeqCst); + Self + } +} + +impl Drop for OutboundHandshakeGuard { + fn drop(&mut self) { + OUTBOUND_HANDSHAKES_ACTIVE.fetch_sub(1, Ordering::SeqCst); + } +} + +pub fn outbound_handshake_in_progress() -> bool { + OUTBOUND_HANDSHAKES_ACTIVE.load(Ordering::SeqCst) != 0 +} pub async fn connect_and_handshake( params: Connect, ) -> Result<(), Box> { + let _handshake_guard = OutboundHandshakeGuard::acquire(); let original_params = params.clone(); let err = match connect_and_handshake_once(params).await { Ok(()) => return Ok(()), @@ -87,7 +109,7 @@ async fn connect_and_handshake_once( .map_err(|err| Box::new(err) as Box) } -fn sponsor_endpoint_from_error(error: &str) -> Option { +pub(crate) fn sponsor_endpoint_from_error(error: &str) -> Option { let (_, sponsor) = error.split_once("sponsor=")?; let endpoint = sponsor .split_whitespace() diff --git a/src/rpc/client/handshake_processing.rs b/src/rpc/client/handshake_processing.rs index ea6a525..2bdd31c 100644 --- a/src/rpc/client/handshake_processing.rs +++ b/src/rpc/client/handshake_processing.rs @@ -15,7 +15,7 @@ use crate::records::memory::connections::{ spawn_peer_setup_retry, startup_synced_peer_streams, CONNECTIONS, }; use crate::records::memory::network_mapping::NodeInfo; -use crate::records::memory::response_channels::{retire_peer_entries, Command}; +use crate::records::memory::response_channels::Command; use crate::records::memory::structs::Connection; use crate::rpc::client::genesis_compat::ensure_compatible_genesis; use crate::rpc::client::governance_state_sync::sync_governance_state; @@ -77,23 +77,32 @@ pub struct BootstrapParams { } pub fn spawn_bootstrap_peer_discovery(params: BootstrapParams) { - let Some(startup_guard) = try_start_startup_discovery() else { - info!("[startup] startup synchronization is already active; coalescing duplicate request"); - spawn_peer_setup_retry(params.connections_key, params.map); - return; - }; - tokio::spawn(async move { + let startup_guard = loop { + if let Some(guard) = try_start_startup_discovery() { + break guard; + } + + let current_stream = Connection::get_stream_from_memory(¶ms.connections_key).await; + if current_stream + .as_ref() + .map(|current| !Arc::ptr_eq(current, ¶ms.stream)) + .unwrap_or(true) + { + return; + } + + sleep(Duration::from_millis(250)).await; + }; let _startup_guard = startup_guard; let failed_stream = params.stream.clone(); let failed_key = params.connections_key.clone(); - let command_map = params.map.clone(); if let Err(e) = bootstrap_peer_discovery(params).await { warn!("[bootstrap] asynchronous peer setup failed for {failed_key}: {e}"); + // Request UIDs are allowed to expire independently. Retiring by + // endpoint here could cancel requests owned by a replacement + // stream connected to the same peer. remove_stream_from_memory(&failed_stream).await; - // The stream may already have been removed by the failing setup - // path. Retire its request waiters explicitly in either case. - retire_peer_entries(command_map, &failed_key).await; } }); } @@ -281,9 +290,9 @@ pub async fn bootstrap_peer_discovery(params: BootstrapParams) -> Result<(), Str } info!("[sync] post-sync checks complete, mining grace period started"); - for (peer_key, _) in startup_synced_peer_streams().await { - if !mark_peer_operational(&peer_key, params.map.clone()).await { - spawn_peer_setup_retry(peer_key, params.map.clone()); + for (peer_key, peer_stream) in startup_synced_peer_streams().await { + if !mark_peer_operational(&peer_key, &peer_stream, params.map.clone()).await { + spawn_peer_setup_retry(peer_key, peer_stream, params.map.clone()); } } sleep(Duration::from_secs(15)).await; @@ -477,7 +486,12 @@ pub async fn process_handshake_response( "Wallet registry sync failed after handshake: {err}" ))); } - mark_peer_wallet_registry_synced(&connections_key).await; + if !mark_peer_wallet_registry_synced(&connections_key, &stream).await { + remove_stream_from_memory(&stream).await; + return Err(io::Error::other( + "Wallet registry sync completed for a stale or missing connection", + )); + } if let Err(err) = register_connected_wallet( Arc::clone(&stream), @@ -556,33 +570,52 @@ pub async fn process_handshake_response( })?; if !params.first { - let matched = compare_network_mapping_digest( + let matched = match compare_network_mapping_digest( broadcast_stream.clone(), params.map.clone(), &connections_key, ) .await - .map_err(|err| io::Error::other(format!("Network-map digest comparison failed: {err}")))?; + { + Ok(matched) => matched, + Err(err) => { + remove_stream_from_memory(&stream).await; + return Err(io::Error::other(format!( + "Network-map digest comparison failed: {err}" + ))); + } + }; if !matched { info!("[network_map] reconciling signed mapping state with peer {connections_key}"); - reconcile_network_mapping_with_peer( + if let Err(err) = reconcile_network_mapping_with_peer( broadcast_stream.clone(), params.map.clone(), &connections_key, ) .await - .map_err(|err| io::Error::other(format!("Network-map reconciliation failed: {err}")))?; + { + remove_stream_from_memory(&stream).await; + return Err(io::Error::other(format!( + "Network-map reconciliation failed: {err}" + ))); + } - let reconciled = compare_network_mapping_digest( + let reconciled = match compare_network_mapping_digest( broadcast_stream.clone(), params.map.clone(), &connections_key, ) .await - .map_err(|err| { - io::Error::other(format!("Reconciled network-map comparison failed: {err}")) - })?; + { + Ok(reconciled) => reconciled, + Err(err) => { + remove_stream_from_memory(&stream).await; + return Err(io::Error::other(format!( + "Reconciled network-map comparison failed: {err}" + ))); + } + }; if !reconciled { remove_stream_from_memory(&stream).await; return Err(io::Error::other(format!( @@ -591,7 +624,12 @@ pub async fn process_handshake_response( } } } - mark_peer_network_map_synced(&connections_key).await; + if !mark_peer_network_map_synced(&connections_key, &stream).await { + remove_stream_from_memory(&stream).await; + return Err(io::Error::other( + "Network map sync completed for a stale or missing connection", + )); + } if params.first { let bsparams = BootstrapParams { @@ -605,14 +643,14 @@ pub async fn process_handshake_response( spawn_bootstrap_peer_discovery(bsparams); } else if is_normal_mode() { - if !mark_peer_operational(&connections_key, params.map.clone()).await { - spawn_peer_setup_retry(connections_key.clone(), params.map.clone()); + if !mark_peer_operational(&connections_key, &stream, params.map.clone()).await { + spawn_peer_setup_retry(connections_key.clone(), stream.clone(), params.map.clone()); } } else { // A concurrently discovered peer may finish setup after the startup // promotion snapshot. Retry until the canonical sync owner releases // rather than leaving that valid peer passive indefinitely. - spawn_peer_setup_retry(connections_key.clone(), params.map.clone()); + spawn_peer_setup_retry(connections_key.clone(), stream.clone(), params.map.clone()); } Ok(()) } diff --git a/src/rpc/client/syncing.rs b/src/rpc/client/syncing.rs index 268caa3..a7d2103 100644 --- a/src/rpc/client/syncing.rs +++ b/src/rpc/client/syncing.rs @@ -4,6 +4,7 @@ use crate::log::{error, info, warn}; use crate::orphans::structs::OrphanCheckup2; use crate::orphans::sync_check::sync_checkup; use crate::records::block_height::get_block_height::get_height; +use crate::records::memory::network_mapping::NodeInfo; use crate::records::memory::response_channels::reserve_entry_with_context; use crate::records::memory::response_channels::Command; use crate::records::memory::torrent_status::{set_torrent_status, TorrentStatus}; @@ -34,9 +35,20 @@ fn is_mapping_sync_race(err: &str) -> bool { || err.contains("Miner wallet address is not registered") } +fn is_sync_transport_error(err: &str) -> bool { + err.contains("No connected miner peers available") + || err.contains("No available peer could provide remaining pieces") + || err.contains("Timed out waiting for torrent response") + || err.contains("No torrent response received") + || err.contains("Timed out waiting for piece reply") + || err.contains("Piece reply channel closed") + || err.contains("Peer could not provide piece") + || err.contains("Failed to send torrent request") +} + #[cfg(test)] mod tests { - use super::is_mapping_sync_race; + use super::{is_mapping_sync_race, is_sync_transport_error}; #[test] fn missing_or_not_yet_eligible_miner_is_a_mapping_race() { @@ -49,6 +61,17 @@ mod tests { assert!(!is_mapping_sync_race("Invalid miner proof.")); assert!(!is_mapping_sync_race("Incorrect previous_block_hash.")); } + + #[test] + fn peer_loss_is_not_sent_to_orphan_recovery() { + assert!(is_sync_transport_error( + "No connected miner peers available for block 15070" + )); + assert!(is_sync_transport_error( + "Timed out waiting for torrent response at height 15070" + )); + assert!(!is_sync_transport_error("Incorrect previous_block_hash.")); + } } fn is_transient_sync_error(err: &str) -> bool { @@ -144,6 +167,15 @@ async fn handle_sync_error( ) -> io::Result<()> { warn!("[sync] error saving block: height={height} err={err}"); + // Transport loss is not evidence of a fork. Release this sync owner so + // a replacement connection can resume from the persisted chain height. + if is_sync_transport_error(err) { + return Err(io::Error::new( + io::ErrorKind::ConnectionAborted, + format!("sync transport failed at height {height}: {err}"), + )); + } + if err.contains("Invalid reward for the Rewards Transaction") || err.contains("This address is not eligable to mine") || err.contains("This miner address is not registered") @@ -329,6 +361,9 @@ pub async fn node_syncing( retry_counts.remove(&(next_to_save - 1)); } abort_pending_downloads(&mut pending); + NodeInfo::persist_mined_counts(db) + .await + .map_err(io::Error::other)?; info!("[sync] node syncing complete, awaiting post-sync checks before mining resumes"); Ok(()) } diff --git a/src/rpc/commands/add_network_node.rs b/src/rpc/commands/add_network_node.rs index d032dbb..56b606f 100644 --- a/src/rpc/commands/add_network_node.rs +++ b/src/rpc/commands/add_network_node.rs @@ -1,6 +1,6 @@ +use crate::records::memory::enums::ClientType; use crate::records::memory::network_mapping::structs::{AddAddressParams, SignedNodeEdit}; use crate::records::memory::network_mapping::NodeInfo; -use crate::records::memory::enums::ClientType; use crate::records::memory::response_channels::Command; use crate::rpc::read_bytes_from_stream; use crate::rpc::responses::RpcResponse; diff --git a/src/rpc/server/command_loop_state.rs b/src/rpc/server/command_loop_state.rs index 08591bd..0d3d9e9 100644 --- a/src/rpc/server/command_loop_state.rs +++ b/src/rpc/server/command_loop_state.rs @@ -5,9 +5,9 @@ use crate::records::memory::connections::{ }; use crate::records::memory::enums::ClientType; use crate::rpc::command_maps::{ - RPC_BLOCK_HEIGHT, RPC_BLOCK_PIECE, RPC_NETWORK_MEMBERSHIP_RECONCILE, - RPC_NETWORK_MONITOR_ADD, RPC_NETWORK_MONITOR_REMOVE, RPC_REPLY, RPC_SUBMIT_TORRENT, - RPC_SUBMIT_TRANSACTION, RPC_TORRENT_BY_HEIGHT, + RPC_BLOCK_HEIGHT, RPC_BLOCK_PIECE, RPC_NETWORK_MEMBERSHIP_RECONCILE, RPC_NETWORK_MONITOR_ADD, + RPC_NETWORK_MONITOR_REMOVE, RPC_REPLY, RPC_SUBMIT_TORRENT, RPC_SUBMIT_TRANSACTION, + RPC_TORRENT_BY_HEIGHT, }; use crate::rpc::server::connection_memory_manager::remove_stream_from_memory; use crate::rpc::server::flood_protection::check_request_frequency_with_client_type; @@ -108,9 +108,7 @@ pub async fn next_incoming_command( let accepts_relay = if matches!( command, - RPC_NETWORK_MONITOR_ADD - | RPC_NETWORK_MONITOR_REMOVE - | RPC_NETWORK_MEMBERSHIP_RECONCILE + RPC_NETWORK_MONITOR_ADD | RPC_NETWORK_MONITOR_REMOVE | RPC_NETWORK_MEMBERSHIP_RECONCILE ) { peer_accepts_mapping_relay(connections_key).await } else { diff --git a/src/rpc/server/connection_memory_manager.rs b/src/rpc/server/connection_memory_manager.rs index 29de541..9a02e08 100644 --- a/src/rpc/server/connection_memory_manager.rs +++ b/src/rpc/server/connection_memory_manager.rs @@ -1,12 +1,12 @@ use crate::common::binary_conversions::{binary_to_ip, ip_to_binary}; use crate::log::warn; use crate::records::memory::connections::{ - miner_connection_count, retire_peer_requests_from_runtime, set_stream_health, - spawn_outage_recovery_from_runtime, spawn_retry_dropped_outgoing, CONNECTIONS, + miner_connection_count, set_stream_health, spawn_outage_recovery_from_runtime, + spawn_retry_dropped_outgoing, CONNECTIONS, }; use crate::records::memory::enums::{ClientType, ConnectionType}; use crate::records::memory::response_channels::{ - delete_entry, reserve_entry_with_context, retire_peer_entries, Command, + delete_entry, reserve_entry_with_context, Command, }; use crate::records::memory::structs::{ConnectionHealth, StoreConnectionParams}; use crate::rpc::command_maps::RPC_BLOCK_HEIGHT; @@ -167,7 +167,6 @@ async fn purge_stale_duplicate_miner(ip: &str, command_map: Arc>) ); connection.drop_connection(connection_type, duplicate_ip, duplicate_port); drop(guard); - retire_peer_entries(command_map, &duplicate_key).await; true } @@ -337,7 +336,6 @@ pub async fn remove_stream_from_memory(stream: &Arc>) { let ip = crate::common::binary_conversions::binary_to_ip(ip_bytes); let dropped = connection.drop_connection(connection_type, ip.clone(), port); drop(connection_instance); - retire_peer_requests_from_runtime(&format!("{ip}:{port}")).await; if connection_type == ConnectionType::Outgoing && dropped .as_ref() diff --git a/src/rpc/server/flood_protection.rs b/src/rpc/server/flood_protection.rs index 8e0f3dd..a5fc1de 100644 --- a/src/rpc/server/flood_protection.rs +++ b/src/rpc/server/flood_protection.rs @@ -2,8 +2,8 @@ use crate::records::ip_score::enums::InfractionType; use crate::records::ip_score::score::update_ip_score; use crate::records::memory::enums::ClientType; use crate::rpc::command_maps::{ - RPC_ADD_NETWORK_NODE, RPC_NETWORK_MAPPING_HASH, RPC_NETWORK_MONITOR_ADD, - RPC_NETWORK_MEMBERSHIP_RECONCILE, RPC_NETWORK_MONITOR_REMOVE, RPC_NETWORK_MONITOR_STATE, + RPC_ADD_NETWORK_NODE, RPC_NETWORK_MAPPING_HASH, RPC_NETWORK_MEMBERSHIP_RECONCILE, + RPC_NETWORK_MONITOR_ADD, RPC_NETWORK_MONITOR_REMOVE, RPC_NETWORK_MONITOR_STATE, RPC_SETUP_COMPLETE, }; use crate::rpc::server::structs::RpcFloodState; diff --git a/src/rpc/server/handshake.rs b/src/rpc/server/handshake.rs index 96a9e35..0804741 100644 --- a/src/rpc/server/handshake.rs +++ b/src/rpc/server/handshake.rs @@ -8,7 +8,7 @@ use crate::orphans::torrent_candidates::hydrate_torrent_candidates; use crate::records::block_height::get_block_height::get_height; use crate::records::memory::connections::{ mark_peer_catching_up, mark_peer_network_map_synced, mark_peer_operational, - mark_peer_wallet_registry_synced, spawn_peer_setup_retry, + mark_peer_wallet_registry_synced, peer_connection_count, spawn_peer_setup_retry, }; use crate::records::memory::enums::ClientType; use crate::records::memory::response_channels::generate_uid; @@ -16,7 +16,9 @@ use crate::records::memory::response_channels::Command; use crate::records::memory::structs::Connection; use crate::rpc::client::genesis_compat::ensure_compatible_genesis; use crate::rpc::client::governance_state_sync::sync_governance_state; +use crate::rpc::client::handshake::{connect_and_handshake, sponsor_endpoint_from_error}; use crate::rpc::client::register_wallet::register_connected_wallet; +use crate::rpc::client::structs::Connect; use crate::rpc::client::syncing::node_syncing; use crate::rpc::client::wallet_registry_sync::sync_wallet_registry_with_retries; use crate::rpc::responses::RpcResponse; @@ -28,7 +30,7 @@ use crate::rpc::server::tests::{endpoint_port, is_port_open}; use crate::sled::Db; use crate::sleep; use crate::startup::network_broadcast::{ - announce_self_to_network, get_network_mapping_for_address, + announce_self_to_network, get_network_mapping_for_address, reconcile_network_mapping_with_peer, }; use crate::startup::remote_height::request_remote_height; use crate::wallets::structures::Wallet; @@ -41,7 +43,6 @@ use crate::TcpStream; use crate::Utc; const LIVE_ORPHAN_WINDOW: u32 = 10; -const PASSIVE_CATCH_UP_BLOCK_LIMIT: u32 = 20; fn peer_reached_catch_up_target( local_height: u32, @@ -52,11 +53,6 @@ fn peer_reached_catch_up_target( && local_height.saturating_sub(remote_height) <= LIVE_ORPHAN_WINDOW } -fn passive_catch_up_expired(local_height: u32, remote_height: u32, catch_up_target: u32) -> bool { - local_height.saturating_sub(catch_up_target) > PASSIVE_CATCH_UP_BLOCK_LIMIT - && local_height.saturating_sub(remote_height) > LIVE_ORPHAN_WINDOW -} - async fn drop_failed_handshake(stream: &Arc>) { // Failed handshakes are never stored in connection memory, but the // accepted TCP socket should still be closed immediately. @@ -234,43 +230,32 @@ fn spawn_incoming_peer_promotion_watcher( }; if peer_reached_catch_up_target(local_height, remote_height, catch_up_target) { - if mark_peer_operational(&connections_key, map.clone()).await { + if mark_peer_operational(&connections_key, &stream, map.clone()).await { break; } continue; } - - if passive_catch_up_expired(local_height, remote_height, catch_up_target) { - warn!( - "[startup] incoming peer failed to catch up before passive limit; disconnecting: peer={connections_key} target_height={catch_up_target} local_height={local_height} remote_height={remote_height}" - ); - remove_stream_from_memory(&stream).await; - break; - } } }); } -fn needs_reverse_bootstrap_announcement( - self_add_allowed: bool, +fn needs_reverse_membership_announcement( local_membership_active: bool, + has_operational_peer: bool, ) -> bool { - self_add_allowed && !local_membership_active + !local_membership_active || !has_operational_peer } #[cfg(test)] mod tests { - use super::{ - needs_reverse_bootstrap_announcement, passive_catch_up_expired, - peer_reached_catch_up_target, - }; + use super::{needs_reverse_membership_announcement, peer_reached_catch_up_target}; #[test] - fn reverse_bootstrap_announcement_is_only_for_an_unregistered_first_node() { - assert!(needs_reverse_bootstrap_announcement(true, false)); - assert!(!needs_reverse_bootstrap_announcement(true, true)); - assert!(!needs_reverse_bootstrap_announcement(false, false)); - assert!(!needs_reverse_bootstrap_announcement(false, true)); + fn inactive_local_membership_requires_peer_sponsorship_at_every_height() { + assert!(needs_reverse_membership_announcement(false, false)); + assert!(needs_reverse_membership_announcement(false, true)); + assert!(needs_reverse_membership_announcement(true, false)); + assert!(!needs_reverse_membership_announcement(true, true)); } #[test] @@ -283,12 +268,6 @@ mod tests { assert!(!peer_reached_catch_up_target(66_091, 66_090, 66_091)); assert!(!peer_reached_catch_up_target(66_120, 66_091, 66_091)); } - - #[test] - fn stale_passive_peer_expires_only_after_falling_outside_live_window() { - assert!(!passive_catch_up_expired(66_112, 66_102, 66_091)); - assert!(passive_catch_up_expired(66_112, 66_100, 66_091)); - } } async fn complete_incoming_miner_setup( @@ -319,7 +298,13 @@ async fn complete_incoming_miner_setup( ); return; } - mark_peer_wallet_registry_synced(connections_key).await; + if !mark_peer_wallet_registry_synced(connections_key, &stream).await { + warn!( + "[startup] incoming peer wallet sync completed for a stale or missing stream: peer={connections_key}" + ); + remove_stream_from_memory(&stream).await; + return; + } if let Err(err) = register_connected_wallet( stream.clone(), @@ -371,20 +356,16 @@ async fn complete_incoming_miner_setup( remove_stream_from_memory(&stream).await; return; } - let self_add_allowed = - crate::records::memory::network_mapping::NodeInfo::self_add_allowed_at_height( - remote_height, - ); let short_address = wallet.saved.short_address.clone(); let local_membership_active = crate::records::memory::network_mapping::NodeInfo::is_active_address(&short_address).await; + let has_operational_peer = peer_connection_count().await > 0; - if needs_reverse_bootstrap_announcement(self_add_allowed, local_membership_active) { - // Before the sponsored-add gate, reverse announcement lets tiny early - // networks bootstrap from one live peer. It is only valid while this - // receiving node has no active membership of its own. An established - // node must never import even its own record from a newly connected - // peer. + if needs_reverse_membership_announcement(local_membership_active, has_operational_peer) { + // An inactive local node announces itself to the connected peer at + // every chain height. The peer sponsors and signs the membership; this + // node never adds itself. The receiver's normal sponsor-eligibility + // rules still apply after the 10,000-block boundary. if let Err(err) = announce_self_to_network( stream.clone(), &short_address, @@ -395,7 +376,41 @@ async fn complete_incoming_miner_setup( ) .await { - error!("[startup] incoming peer network map sync failed: {err}"); + let Some(sponsor_endpoint) = sponsor_endpoint_from_error(&err) else { + error!("[startup] incoming peer network map sync failed: {err}"); + remove_stream_from_memory(&stream).await; + return; + }; + let Ok(sponsor_addr) = sponsor_endpoint.parse() else { + error!( + "[startup] incoming peer returned an invalid sponsor endpoint: {sponsor_endpoint}" + ); + remove_stream_from_memory(&stream).await; + return; + }; + + warn!( + "[startup] incoming peer cannot sponsor local membership; connecting to eligible sponsor {sponsor_endpoint}" + ); + let sponsor = Connect { + addr: sponsor_addr, + node_ip: sponsor_endpoint.clone(), + db: db.clone(), + wallet: wallet.clone(), + map: map.clone(), + first: true, + }; + if let Err(sponsor_err) = connect_and_handshake(sponsor).await { + error!( + "[startup] eligible sponsor connection failed: original={err}; sponsor={sponsor_endpoint}; error={sponsor_err}" + ); + remove_stream_from_memory(&stream).await; + return; + } + + // A sponsor rejection is a terminal response on the original + // connection. The redirected outgoing connection now owns local + // startup and mapping synchronization. remove_stream_from_memory(&stream).await; return; } @@ -413,7 +428,27 @@ async fn complete_incoming_miner_setup( return; } } - mark_peer_network_map_synced(connections_key).await; + + // The server side never replaces its established map with a connecting + // peer's snapshot. It does, however, send its signed membership and + // monitor state to that peer. On additional connections the outgoing side + // performs the reciprocal push, allowing deterministic record validation + // to converge both maps without assigning either connection direction + // unconditional authority. + if let Err(err) = + reconcile_network_mapping_with_peer(stream.clone(), map.clone(), connections_key).await + { + error!("[startup] incoming peer network-map reconciliation failed: {err}"); + remove_stream_from_memory(&stream).await; + return; + } + if !mark_peer_network_map_synced(connections_key, &stream).await { + warn!( + "[startup] incoming peer mapping sync completed for a stale or missing stream: peer={connections_key}" + ); + remove_stream_from_memory(&stream).await; + return; + } let (operational, chain_sync_guard, catch_up_target) = match sync_incoming_peer_before_operational( @@ -434,8 +469,8 @@ async fn complete_incoming_miner_setup( }; if operational { - if !mark_peer_operational(connections_key, map.clone()).await { - spawn_peer_setup_retry(connections_key.to_string(), map.clone()); + if !mark_peer_operational(connections_key, &stream, map.clone()).await { + spawn_peer_setup_retry(connections_key.to_string(), stream.clone(), map.clone()); } sleep(Duration::from_secs(15)).await; if let Some(guard) = chain_sync_guard { diff --git a/src/rpc/server/handshake_verifications.rs b/src/rpc/server/handshake_verifications.rs index 55816ca..d5484f1 100644 --- a/src/rpc/server/handshake_verifications.rs +++ b/src/rpc/server/handshake_verifications.rs @@ -3,7 +3,7 @@ use crate::records::memory::enums::ClientType; use crate::records::memory::response_channels::{generate_uid, Command}; use crate::rpc::responses::RpcResponse; use crate::rpc::server::structs::HandshakeTestParams; -use crate::rpc::server::tests::{ip_test, is_within_one_second}; +use crate::rpc::server::tests::{ip_test, is_within_two_seconds}; use crate::wallets::structures::Wallet; use crate::Arc; use crate::Mutex; @@ -57,7 +57,7 @@ pub async fn verify_timestamp( peer_time: u32, timestamp: u32, ) -> bool { - if !is_within_one_second(timestamp, peer_time) { + if !is_within_two_seconds(timestamp, peer_time) { // Tight timestamp checks keep peers from replaying stale signed // handshakes after the connection attempt has passed. let hashmap_key = generate_uid(); @@ -66,7 +66,7 @@ pub async fn verify_timestamp( let response_bytes = RpcResponse::Binary({ let msg = format!( - "error: Handshake Failed: The time on your computer must be within 1 second of the time of our server. Your local time: {peer_time}. Our server time: {timestamp}. Please consider installing NTP to ensure proper timestamps." + "error: Handshake Failed: The time on your computer must be within 2 seconds of the time of our server. Your local time: {peer_time}. Our server time: {timestamp}. Please consider installing NTP to ensure proper timestamps." ) .as_bytes() .to_vec(); diff --git a/src/rpc/server/rpc_command_loop.rs b/src/rpc/server/rpc_command_loop.rs index 8df45fb..00ac447 100644 --- a/src/rpc/server/rpc_command_loop.rs +++ b/src/rpc/server/rpc_command_loop.rs @@ -640,7 +640,8 @@ pub async fn start_loop( stream_locked.clone(), ) .await?; - mark_peer_remote_setup_complete(&connections_key, map.clone()).await; + mark_peer_remote_setup_complete(&connections_key, &stream_locked, map.clone()) + .await; let result = responses::RpcResponse::Binary(b"setup_complete_ack".to_vec()); result .send(&stream_locked, Some(&connections_key), uid) diff --git a/src/rpc/server/tests.rs b/src/rpc/server/tests.rs index 32e11f0..4e8dd9a 100644 --- a/src/rpc/server/tests.rs +++ b/src/rpc/server/tests.rs @@ -48,8 +48,21 @@ pub async fn is_port_open(ip_port: &str) -> Result bool { +// Verify connecting client time is within two seconds of our own. +pub fn is_within_two_seconds(timestamp1: u32, timestamp2: u32) -> bool { let duration = timestamp1.wrapping_sub(timestamp2); - duration <= 1 || timestamp2.wrapping_sub(timestamp1) <= 1 + duration <= 2 || timestamp2.wrapping_sub(timestamp1) <= 2 +} + +#[cfg(test)] +mod timestamp_tests { + use super::is_within_two_seconds; + + #[test] + fn handshake_clock_tolerance_includes_two_seconds_only() { + assert!(is_within_two_seconds(100, 98)); + assert!(is_within_two_seconds(98, 100)); + assert!(!is_within_two_seconds(100, 97)); + assert!(!is_within_two_seconds(97, 100)); + } } diff --git a/src/startup/connections.rs b/src/startup/connections.rs index fee6787..eb6cd72 100644 --- a/src/startup/connections.rs +++ b/src/startup/connections.rs @@ -3,12 +3,13 @@ use crate::common::network_startup::{get_ip_and_port, get_node_connections}; use crate::log::{error, info, warn}; use crate::miner::flag::{is_mining_stop_requested, is_normal_mode}; use crate::records::memory::connections::{ - operational_peer_wallets, peer_connection_count, ready_outgoing_connection_count, - refill_outgoing_connections_once, + miner_connection_count, operational_peer_wallets, peer_connection_count, + ready_outgoing_connection_count, refill_outgoing_connections_once, }; use crate::records::memory::network_mapping::NodeInfo; use crate::records::memory::response_channels::Command; use crate::rpc::client::handshake::connect_and_handshake; +use crate::rpc::client::handshake::outbound_handshake_in_progress; use crate::rpc::client::structs::Connect; use crate::sled::Db; use crate::sleep; @@ -65,6 +66,7 @@ pub async fn handle_connections( // dialing the list. Only the first two nodes establishing genesis require // reciprocal sponsorship; mature restarts need an operational peer. info!("No existing network peer is available. Waiting for connections."); + let mut retry_seconds = 0_u64; loop { let peer_wallets = operational_peer_wallets().await; let sponsorship_ready = if first_two_node_startup { @@ -80,6 +82,17 @@ pub async fn handle_connections( } return Ok(()); } + + // The normal isolated-recovery task is started only after this + // function returns. While startup is still waiting, periodically run + // that same single-owner recovery path so a temporarily unavailable + // configured peer cannot strand the node forever. A live or + // still-setting-up miner socket suppresses the attempt. + retry_seconds += 1; + if outgoing_connections > 0 && retry_seconds >= 60 { + retry_seconds = 0; + spawn_outage_recovery(db.clone(), wallet.clone(), map.clone()); + } sleep(Duration::from_secs(1)).await; } } @@ -198,7 +211,7 @@ pub fn spawn_outage_recovery(db: Db, wallet: Arc, map: Arc Result<(), String> { .try_into() .map_err(|_| "Invalid torrent info_hash length".to_string())?, ); - // Keep polling until every piece has either been downloaded or the // target block becomes obsolete because the chain tip advanced. loop { @@ -322,8 +321,8 @@ pub async fn download_block_pieces(params: DownloadSave) -> Result<(), String> { )); } } - // Poll more frequently so a fully downloaded block moves into - // combine/verify/save without an extra 1-second stall. + // Many block downloads can be active concurrently. Keep this poll + // bounded so waiting tasks do not spin and starve network processing. sleep(Duration::from_millis(50)).await; } Ok(()) diff --git a/src/torrent/torrenting_system/torrent_requests.rs b/src/torrent/torrenting_system/torrent_requests.rs index f2230e7..6d27f3b 100644 --- a/src/torrent/torrenting_system/torrent_requests.rs +++ b/src/torrent/torrenting_system/torrent_requests.rs @@ -30,8 +30,14 @@ pub async fn send_request_torrent_message( message.extend(request_torrent_binary); message.extend(hashmap_key); message.extend(get_height_binary); - RpcResponse::send_raw(&stream, Some(&connections_key), &message).await; - Ok(()) + if RpcResponse::send_raw(&stream, Some(&connections_key), &message).await { + Ok(()) + } else { + Err(io::Error::new( + io::ErrorKind::ConnectionAborted, + format!("Failed to send torrent request to {connections_key}"), + )) + } } pub async fn handle_response_and_save_torrent(