contractless/Contractless
contractless
/
Contractless
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

network startup fixes

This commit is contained in:
contractless 2026-09-06 14:10:28 -06:00
parent b843afe353
commit 7ca01b9e69
20 changed files with 480 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/bin/lookup_online_node_count.rs
View File

@ -56,8 +56,7 @@ fn online_node_count(response: &[u8]) -> Result<usize, String> {
.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;

184
src/records/memory/connections.rs
View File

@ -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<Mutex<TcpStream>>,
) -> 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<Mutex<TcpStream>>,
) -> 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<Mutex<TcpStream>>,
command_map: Arc<Mutex<Command>>,
) -> Option<(Arc<Mutex<TcpStream>>, 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<Mutex<TcpStream>>,
) -> 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<Mutex<TcpStream>>,
command_map: Arc<Mutex<Command>>,
) -> 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<Mutex<TcpStream>>,
command_map: Arc<Mutex<Command>>,
) -> 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<Mutex<TcpStream>>) -> 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<Mutex<TcpStream>>) -> 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<u32> {
})
}
pub async fn mark_peer_operational(key: &str, map: Arc<Mutex<Command>>) -> bool {
pub async fn mark_peer_operational(
key: &str,
stream: &Arc<Mutex<TcpStream>>,
map: Arc<Mutex<Command>>,
) -> bool {
if peer_is_operational(key).await {
return true;
}
@ -1309,11 +1348,18 @@ pub async fn mark_peer_operational(key: &str, map: Arc<Mutex<Command>>) -> 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<Mutex<Command>>) -> 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<Mutex<Command>>) -> 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<Mutex<Command>>) -> bool
false
}
pub async fn mark_peer_remote_setup_complete(key: &str, map: Arc<Mutex<Command>>) -> bool {
pub async fn mark_peer_remote_setup_complete(
key: &str,
stream: &Arc<Mutex<TcpStream>>,
map: Arc<Mutex<Command>>,
) -> 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<Mutex<Command>>) {
pub fn spawn_peer_setup_retry(
key: String,
stream: Arc<Mutex<TcpStream>>,
map: Arc<Mutex<Command>>,
) {
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(&current, &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));
}
}

8
src/records/memory/network_mapping/mined_counts.rs
View File

@ -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());

49
src/records/memory/network_mapping/monitor.rs
View File

@ -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<String, NodeInfo>,
event_state: &HashMap<String, SignedMonitorEdit>,
) -> HashMap<String, SignedMonitorEdit> {
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<u8> {
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)
}

3
src/records/memory/network_mapping/queries.rs
View File

@ -95,6 +95,7 @@ impl NodeInfo {
pub(crate) async fn signed_mapping_state() -> (Vec<SyncedNodeState>, Vec<SignedMonitorEdit>) {
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<SyncedNodeState> = map
.iter()
@ -221,6 +222,7 @@ impl NodeInfo {
pub async fn canonical_mapping_digest() -> Result<Vec<u8>, 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<u8> = Vec::with_capacity(map.len() * NODE_RECORD_FIXED_BYTES);
for (address, node_info) in map.iter() {

24
src/rpc/client/handshake.rs
View File

@ -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<dyn std::error::Error + Send + Sync>> {
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<dyn std::error::Error + Send + Sync>)
}
fn sponsor_endpoint_from_error(error: &str) -> Option<String> {
pub(crate) fn sponsor_endpoint_from_error(error: &str) -> Option<String> {
let (_, sponsor) = error.split_once("sponsor=")?;
let endpoint = sponsor
.split_whitespace()

92
src/rpc/client/handshake_processing.rs
View File

@ -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(&params.connections_key).await;
if current_stream
.as_ref()
.map(|current| !Arc::ptr_eq(current, &params.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(())
}

37
src/rpc/client/syncing.rs
View File

@ -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(())
}

2
src/rpc/commands/add_network_node.rs
View File

@ -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;

10
src/rpc/server/command_loop_state.rs
View File

@ -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 {

8
src/rpc/server/connection_memory_manager.rs
View File

@ -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<Mutex<Command>>)
);
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<Mutex<TcpStream>>) {
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()

4
src/rpc/server/flood_protection.rs
View File

@ -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;

135
src/rpc/server/handshake.rs
View File

@ -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<Mutex<TcpStream>>) {
// 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 {

6
src/rpc/server/handshake_verifications.rs
View File

@ -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();

3
src/rpc/server/rpc_command_loop.rs
View File

@ -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)

19
src/rpc/server/tests.rs
View File

@ -48,8 +48,21 @@ pub async fn is_port_open(ip_port: &str) -> Result<bool, Box<dyn std::error::Err
}
}
// verify connecting client time is within 1 second of our own.
pub fn is_within_one_second(timestamp1: u32, timestamp2: u32) -> 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));
}
}

19
src/startup/connections.rs
View File

@ -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<Wallet>, map: Arc<Mutex<Command
// A raw socket that never completed setup is not network recovery.
// Only an operational miner peer should suppress configured-peer
// outage recovery.
if peer_connection_count().await != 0 {
if outbound_handshake_in_progress() || miner_connection_count().await != 0 {
return;
}

9
src/startup/network_broadcast.rs
View File

@ -273,11 +273,10 @@ pub async fn reconcile_network_mapping_with_peer(
.await?;
}
// Reconciliation on an additional/discovered connection is intentionally
// one-way. This node already joined through its bootstrap owner, so the
// newly connected peer may merge our signed state but must never replace
// this node's established mapping. Missing records continue to arrive
// through the normal redundant network broadcasts.
// This operation is deliberately a signed-state push, never a snapshot
// replacement. Incoming setup performs the corresponding push in the
// other direction, so each node validates and merges the peer's records
// using the same deterministic rules.
Ok(())
}

5
src/torrent/torrenting_system/download_pieces.rs
View File

@ -186,7 +186,6 @@ pub async fn download_block_pieces(params: DownloadSave) -> 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(())

10
src/torrent/torrenting_system/torrent_requests.rs
View File

@ -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(