From 504207c4248a32c729b436e3cc2658f113b561f0 Mon Sep 17 00:00:00 2001 From: viraladmin <00purple@gmail.com> Date: Thu, 18 Jun 2026 11:38:52 -0600 Subject: [PATCH] syncing fixes --- .gitignore | 1 + src/orphans/orphan_checkup.rs | 3 + src/orphans/undo_transactions/undo_rewards.rs | 3 +- src/records/memory/connections.rs | 5 +- src/records/memory/response_channels.rs | 2 +- src/records/record_chain/rewards_tx.rs | 99 +++++- src/records/record_chain/save.rs | 90 +++--- src/rpc/client/syncing.rs | 306 ++++++++++++------ src/rpc/commands/route_reply.rs | 4 +- src/rpc/server/handshake.rs | 34 +- .../connections/sending_request.rs | 29 +- src/startup/node_runtime.rs | 4 + src/torrent/create_metadata.rs | 5 + src/torrent/structs.rs | 4 + .../torrenting_system/download_pieces.rs | 57 ++-- src/torrent/torrenting_system/save_block.rs | 139 ++++++-- .../torrenting_system/setup_block_download.rs | 37 +++ src/verifications/async_funcs/verify_block.rs | 2 +- 18 files changed, 583 insertions(+), 241 deletions(-) diff --git a/.gitignore b/.gitignore index 93d7517..cc895f3 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,4 @@ wallets/ *.wallet *.key *.pdb +*.zip diff --git a/src/orphans/orphan_checkup.rs b/src/orphans/orphan_checkup.rs index c24f7f5..0cb8eb3 100644 --- a/src/orphans/orphan_checkup.rs +++ b/src/orphans/orphan_checkup.rs @@ -23,6 +23,7 @@ use crate::torrent::unpack_local_torrent::load_torrent; use crate::verifications::verification_service::global_verification_service; use crate::wallets::structures::Wallet; use crate::Arc; +use crate::Mutex; async fn staged_candidates_for_height(height: u32) -> Vec { let mut candidates = Vec::new(); @@ -125,6 +126,8 @@ async fn candidate_attaches_before_rollback( let torrent_map = create_torrent_map(torrent).await?; let download_save_params = DownloadSave { torrent_map, + in_memory_pieces: Arc::new(Mutex::new(crate::HashMap::new())), + use_in_memory_pieces: false, torrent: torrent.clone(), staged_path: String::new(), block_number: height, diff --git a/src/orphans/undo_transactions/undo_rewards.rs b/src/orphans/undo_transactions/undo_rewards.rs index 9686a21..dc11103 100644 --- a/src/orphans/undo_transactions/undo_rewards.rs +++ b/src/orphans/undo_transactions/undo_rewards.rs @@ -3,7 +3,7 @@ use crate::common::network_paths_and_settings::block_extension_and_paths; use crate::decode; use crate::records::balance_sheet::operations::balance_sheet_operation_with_db; use crate::records::record_chain::rewards_tx::{ - remove_reward_credit_marker, reward_credit_applied, + remove_immature_reward, remove_reward_credit_marker, reward_credit_applied, }; use crate::records::record_chain::wallet_tx_index::remove_wallet_transaction_index; use crate::sled::Db; @@ -30,6 +30,7 @@ pub async fn undo_rewards_transaction( ) = block_extension_and_paths(); let value = transaction.unsigned.value; + remove_immature_reward(block_height).await; // Rewards are only spendable after finalization, so rollback subtracts // them only when that delayed credit has actually been applied. diff --git a/src/records/memory/connections.rs b/src/records/memory/connections.rs index 7bbfe49..028003c 100644 --- a/src/records/memory/connections.rs +++ b/src/records/memory/connections.rs @@ -1,6 +1,7 @@ use crate::common::binary_conversions::{binary_to_ip, ip_to_binary}; use crate::lazy_static; use crate::log::{info, warn}; +use crate::miner::flag::is_normal_mode; use crate::records::memory::enums::{ClientType, ConnectionType}; use crate::records::memory::network_mapping::monitor::{MONITOR_ACTION_ADD, MONITOR_ACTION_REMOVE}; use crate::records::memory::network_mapping::structs::{MonitorAddressParams, SignedMonitorEdit}; @@ -293,7 +294,9 @@ impl Connection { }; let removed = self.connection_map.remove(&connection_key); if let Some(connection_info) = removed.as_ref() { - if ClientType::from_bytes(&connection_info.client_type) == Some(ClientType::Miner) { + if ClientType::from_bytes(&connection_info.client_type) == Some(ClientType::Miner) + && is_normal_mode() + { spawn_monitor_update( ip.clone(), MONITOR_ACTION_REMOVE, diff --git a/src/records/memory/response_channels.rs b/src/records/memory/response_channels.rs index 755c5a2..aeeb6bd 100644 --- a/src/records/memory/response_channels.rs +++ b/src/records/memory/response_channels.rs @@ -20,7 +20,7 @@ pub type Byte3 = [u8; 3]; pub type Command = HashMap; -const SLOW_RPC_TRACE_MS: u128 = 1_000; +pub const SLOW_RPC_TRACE_MS: u128 = 3_000; fn random_3_byte_number() -> [u8; 3] { let mut rng = rand::thread_rng(); diff --git a/src/records/record_chain/rewards_tx.rs b/src/records/record_chain/rewards_tx.rs index d7f8c4c..71d1f48 100644 --- a/src/records/record_chain/rewards_tx.rs +++ b/src/records/record_chain/rewards_tx.rs @@ -1,7 +1,9 @@ use crate::blocks::rewards::RewardsTransaction; use crate::common::types::Transaction; use crate::decode; +use crate::lazy_static; use crate::records::balance_sheet::operations::balance_sheet_operation_with_db; +use crate::records::block_height::get_block_height::get_height; use crate::records::memory::mempool::BASECOIN; use crate::records::record_chain::pending_effects::PendingEffects; use crate::records::record_chain::wallet_tx_index::index_wallet_transaction; @@ -9,8 +11,21 @@ use crate::records::unpack_block::load_by_block_number::load_block; use crate::sled::Db; use crate::Arc; use crate::Mutex; +use std::collections::VecDeque; const FINALIZED_REWARD_TREE: &str = "finalized_rewards"; +const IMMATURE_REWARD_WINDOW: usize = 100; + +#[derive(Clone)] +struct ImmatureReward { + block_height: u32, + miner: String, + amount: u64, +} + +lazy_static! { + static ref IMMATURE_REWARDS: Mutex> = Mutex::new(VecDeque::new()); +} fn reward_key(block_height: u32) -> [u8; 4] { block_height.to_le_bytes() @@ -29,19 +44,68 @@ pub fn remove_reward_credit_marker(db: &Db, block_height: u32) { } } -pub async fn finalize_rewards_through_height(db: &Db, cutoff_height: u32) -> Result<(), String> { +fn mark_reward_credit_applied(db: &Db, block_height: u32) -> Result<(), String> { let tree = db .open_tree(FINALIZED_REWARD_TREE) .map_err(|e| format!("Failed to open finalized reward tree: {e}"))?; + tree.insert(reward_key(block_height), b"true") + .map_err(|e| format!("Failed to store finalized reward marker: {e}"))?; + Ok(()) +} - for block_height in 0..=cutoff_height { - if tree - .contains_key(reward_key(block_height)) - .map_err(|e| format!("Failed to read finalized reward marker: {e}"))? - { - continue; +fn credit_matured_reward(db: &Db, reward: ImmatureReward) -> Result<(), String> { + if reward_credit_applied(db, reward.block_height) { + return Ok(()); + } + + if reward.amount > 0 { + balance_sheet_operation_with_db(db, &reward.miner, reward.amount, &BASECOIN, "addition") + .map_err(|e| format!("Failed to finalize reward balance: {e}"))?; + } + + mark_reward_credit_applied(db, reward.block_height) +} + +pub async fn record_saved_reward( + db: &Db, + block_height: u32, + miner: String, + amount: u64, +) -> Result<(), String> { + let matured = { + let mut rewards = IMMATURE_REWARDS.lock().await; + rewards.retain(|reward| reward.block_height != block_height); + rewards.push_back(ImmatureReward { + block_height, + miner, + amount, + }); + + if rewards.len() > IMMATURE_REWARD_WINDOW { + rewards.pop_front() + } else { + None } + }; + if let Some(reward) = matured { + credit_matured_reward(db, reward)?; + } + + Ok(()) +} + +pub async fn remove_immature_reward(block_height: u32) { + let mut rewards = IMMATURE_REWARDS.lock().await; + rewards.retain(|reward| reward.block_height != block_height); +} + +pub async fn rebuild_immature_reward_window(db: &Db) -> Result<(), String> { + let current_height = get_height(db); + let start_height = current_height.saturating_sub((IMMATURE_REWARD_WINDOW - 1) as u32); + let mut rebuilt = VecDeque::new(); + + for block_height in start_height..=current_height { let block = match load_block(block_height).await { Ok(block) => block, Err(_) => continue, @@ -50,21 +114,22 @@ pub async fn finalize_rewards_through_height(db: &Db, cutoff_height: u32) -> Res for transaction in block.transactions { if let Transaction::Rewards(rewards) = transaction { - balance_sheet_operation_with_db( - db, - &miner, - rewards.unsigned.value, - &BASECOIN, - "addition", - ) - .map_err(|e| format!("Failed to finalize reward balance: {e}"))?; - tree.insert(reward_key(block_height), b"true") - .map_err(|e| format!("Failed to store finalized reward marker: {e}"))?; + rebuilt.push_back(ImmatureReward { + block_height, + miner, + amount: rewards.unsigned.value, + }); break; } } } + while rebuilt.len() > IMMATURE_REWARD_WINDOW { + rebuilt.pop_front(); + } + + let mut rewards = IMMATURE_REWARDS.lock().await; + *rewards = rebuilt; Ok(()) } diff --git a/src/records/record_chain/save.rs b/src/records/record_chain/save.rs index ed89888..f400dbb 100644 --- a/src/records/record_chain/save.rs +++ b/src/records/record_chain/save.rs @@ -1,5 +1,6 @@ use crate::common::check_genesis::genesis_checkup; use crate::common::network_paths_and_settings::block_extension_and_paths; +use crate::common::types::Transaction; use crate::decode; use crate::fs; use crate::log::{error, info}; @@ -24,7 +25,7 @@ use crate::records::memory::torrent_status::prune_torrent_statuses_through_heigh use crate::records::record_chain::parse_transactions::handle_transactions; use crate::records::record_chain::pending_effects::PendingEffects; use crate::records::record_chain::previous_difficulty::previous_block_difficulty; -use crate::records::record_chain::rewards_tx::finalize_rewards_through_height; +use crate::records::record_chain::rewards_tx::record_saved_reward; use crate::records::record_chain::save_flags::SAVE_FLAG; use crate::records::record_chain::structs::{ SaveBinaryDataParams, SaveBinaryDataWithMempoolStreamParams, SaveBlockParams, SaveType, @@ -149,6 +150,14 @@ pub async fn save_block(params: SaveBlockParams) -> Result<(), String> { ) .await?; + let saved_reward_amount = block + .transactions + .iter() + .find_map(|transaction| match transaction { + Transaction::Rewards(rewards) => Some(rewards.unsigned.value), + _ => None, + }); + let start_index = { let index = index_mutex.lock().await; **index @@ -190,8 +199,6 @@ pub async fn save_block(params: SaveBlockParams) -> Result<(), String> { map, }) .await?; - - spawn_processed_cleanup(block_header_number); } else { save_binary_data(SaveBinaryDataParams { data: &binary_data, @@ -210,10 +217,19 @@ pub async fn save_block(params: SaveBlockParams) -> Result<(), String> { map, }) .await?; - - spawn_processed_cleanup(block_header_number); } + if let Some(amount) = saved_reward_amount { + if let Err(err) = record_saved_reward(&db, block_header_number, miner.clone(), amount).await + { + error!( + "Failed to update immature reward window for block {block_header_number}: {err}" + ); + } + } + + spawn_processed_cleanup(block_header_number); + Ok(()) } @@ -357,11 +373,6 @@ async fn save_binary_data_with_mempool_stream( } let _ = update_snapshot(db, next_number, map.clone()).await; if let Some(snapshot_height) = snapshot_height(db).await { - if let Err(err) = finalize_rewards_through_height(db, snapshot_height).await { - error!( - "Failed to finalize rewards through snapshot height {snapshot_height}: {err}" - ); - } prune_recent_torrents(snapshot_height).await; prune_torrent_statuses_through_height(snapshot_height).await; let _ = prune_staged_torrents(snapshot_height).await; @@ -458,27 +469,35 @@ async fn save_binary_data(params: SaveBinaryDataParams<'_>) -> Result<(), String return Err(err); } - let torrent_bytes = match metadata_from_file( - &file_name, - next_number, - difficulty, - timestamp, - header_hash, - previous_hash, - miner.clone(), - ) - .await - { - Ok(torrent_bytes) => torrent_bytes, - Err(err) => { - cleanup_block_indexes(db, block_header_number, header_hash); - cleanup_torrent_file(next_number); - let _ = restore_processed_by_signatures(signatures).await; - if let Err(rollback_err) = applied_effects.rollback(db) { - error!("Failed to roll back block effects: {rollback_err}"); + let uses_staged_sync_torrent = save_type.is_updating() && is_syncing_mode(); + let torrent_bytes = if uses_staged_sync_torrent { + // Startup/live catch-up already has verified incoming torrent metadata + // staged by the download path. Avoid regenerating the same torrent from + // the just-written block; the staged torrent is promoted after save. + None + } else { + match metadata_from_file( + &file_name, + next_number, + difficulty, + timestamp, + header_hash, + previous_hash, + miner.clone(), + ) + .await + { + Ok(torrent_bytes) => Some(torrent_bytes), + Err(err) => { + cleanup_block_indexes(db, block_header_number, header_hash); + cleanup_torrent_file(next_number); + let _ = restore_processed_by_signatures(signatures).await; + if let Err(rollback_err) = applied_effects.rollback(db) { + error!("Failed to roll back block effects: {rollback_err}"); + } + cleanup_block_file(&file_name); + return Err(err); } - cleanup_block_file(&file_name); - return Err(err); } }; @@ -495,20 +514,17 @@ async fn save_binary_data(params: SaveBinaryDataParams<'_>) -> Result<(), String } let _ = update_snapshot(db, next_number, map.clone()).await; if let Some(snapshot_height) = snapshot_height(db).await { - if let Err(err) = finalize_rewards_through_height(db, snapshot_height).await { - error!( - "Failed to finalize rewards through snapshot height {snapshot_height}: {err}" - ); - } prune_recent_torrents(snapshot_height).await; prune_torrent_statuses_through_height(snapshot_height).await; - let _ = prune_staged_torrents(snapshot_height).await; + if !uses_staged_sync_torrent { + let _ = prune_staged_torrents(snapshot_height).await; + } } } else { let _ = update_snapshot(db, next_number, map.clone()).await; } - if !is_syncing_mode() { + if let Some(torrent_bytes) = torrent_bytes.filter(|_| !is_syncing_mode()) { broadcast_new_torrent_to_peers(next_number, &torrent_bytes, map).await; } diff --git a/src/rpc/client/syncing.rs b/src/rpc/client/syncing.rs index f99c81d..0333db4 100644 --- a/src/rpc/client/syncing.rs +++ b/src/rpc/client/syncing.rs @@ -6,18 +6,140 @@ use crate::orphans::sync_check::sync_checkup; use crate::records::block_height::get_block_height::get_height; 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}; use crate::rpc::command_maps::RPC_TORRENT_BY_HEIGHT; use crate::sled::Db; use crate::timeout; use crate::torrent::structs::Torrent; -use crate::torrent::torrenting_system::torrent_requests::{ - handle_response_and_save_torrent, send_request_torrent_message, -}; +use crate::torrent::torrenting_system::save_block::verify_and_save_downloaded_block; +use crate::torrent::torrenting_system::save_torrent::save_staged_torrent; +use crate::torrent::torrenting_system::setup_block_download::download_sync_candidate; +use crate::torrent::torrenting_system::torrent_requests::send_request_torrent_message; +use crate::verifications::verification_service::global_verification_service; use crate::wallets::structures::Wallet; use crate::Arc; use crate::Duration; use crate::Mutex; use crate::TcpStream; +use std::collections::BTreeMap; + +const SYNC_PREFETCH_WINDOW: usize = 100; + +fn abort_pending_downloads( + pending: &mut BTreeMap< + u32, + crate::tokio::task::JoinHandle>, + >, +) { + for (_, handle) in pending.iter() { + handle.abort(); + } + pending.clear(); +} + +fn spawn_sync_download( + height: u32, + stream: Arc>, + db: Db, + map: Arc>, + connections_key: String, + allow_startup_peers: bool, +) -> crate::tokio::task::JoinHandle> { + crate::tokio::spawn(async move { + let (hashmap_key, _torrent_tx, torrent_rx) = reserve_entry_with_context( + map.clone(), + Some(RPC_TORRENT_BY_HEIGHT), + Some(connections_key.clone()), + ) + .await; + send_request_torrent_message(stream, height, hashmap_key, connections_key.clone()) + .await + .map_err(|err| err.to_string())?; + + let torrent_bytes = { + let mut rx = torrent_rx.lock().await; + timeout(Duration::from_secs(30), rx.recv()) + .await + .map_err(|_| format!("Timed out waiting for torrent response at height {height}"))? + .ok_or_else(|| format!("No torrent response received for height {height}"))? + }; + + if let Ok(text) = String::from_utf8(torrent_bytes.clone()) { + let trimmed = text.trim(); + if !trimmed.is_empty() { + return Err(format!( + "textual torrent response at height={height}: {trimmed}" + )); + } + } + + let torrent = Torrent::from_bytes(&torrent_bytes) + .await + .map_err(|err| err.to_string())?; + let staged_path = save_staged_torrent(height, &torrent_bytes) + .await + .map_err(|err| format!("Failed to save staged torrent: {err}"))?; + set_torrent_status(height, &torrent.info.info_hash, TorrentStatus::Pending).await; + + let verification_service = global_verification_service() + .ok_or_else(|| "Verification service not initialized".to_string())?; + download_sync_candidate( + height, + torrent, + staged_path, + allow_startup_peers, + db, + Arc::new(verification_service), + map, + ) + .await + }) +} + +async fn handle_sync_error( + err: &str, + height: u32, + stream: Arc>, + db: &Db, + remote_height: u32, + map: Arc>, + node_syncing: bool, + wallet: Arc, + connections_key: String, +) -> io::Result<()> { + warn!("[sync] error saving block: height={height} err={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") + || err.contains("Miner wallet address is not registered") + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("sync rejected canonical peer block at height {height}: {err}"), + )); + } + + // Refresh local chain state before triggering orphan handling so the + // reconciliation logic starts from the real saved height. + let local_height = get_height(db); + let orphan_checkup_params = OrphanCheckup2 { + stream, + db: db.clone(), + local_height, + remote_height, + recheck_from_height: Some(local_height), + map, + node_syncing, + connections_key, + }; + + if let Err(err) = sync_checkup(orphan_checkup_params, wallet).await { + error!("[sync] orphan check returned error: {err}"); + } + + Ok(()) +} pub async fn node_syncing( stream: Arc>, @@ -35,112 +157,100 @@ pub async fn node_syncing( local_height += 1; } - // Walk forward block-by-block until the local chain catches up to - // the advertised remote height. - while remote_height >= local_height { - let (hashmap_key, _torrent_tx, torrent_rx) = reserve_entry_with_context( - map.clone(), - Some(RPC_TORRENT_BY_HEIGHT), - Some(connections_key.clone()), - ) - .await; - send_request_torrent_message( - stream.clone(), - local_height, - hashmap_key, - connections_key.clone(), - ) - .await?; + let mut next_to_schedule = local_height; + let mut next_to_save = local_height; + let mut pending = BTreeMap::new(); - // Wait for the requested torrent bytes to come back through the - // shared response map entry created for this request. - let mut rx = torrent_rx.lock().await; - if let Some(torrent_bytes) = - timeout(Duration::from_secs(30), rx.recv()) - .await - .map_err(|_| { - io::Error::new( - io::ErrorKind::TimedOut, - "Timed out waiting for torrent response", - ) - })? - { - if let Ok(text) = String::from_utf8(torrent_bytes.clone()) { - let trimmed = text.trim(); - if !trimmed.is_empty() { - warn!("[sync] received textual torrent response at height={local_height} err={trimmed}"); - if local_height == 0 && !genesis_checkup().await { + // Download a small window of candidates ahead, but validate/save only + // the exact next height so parent, difficulty, and balance rules still + // see the canonical local chain state. + while remote_height >= next_to_save { + while pending.len() < SYNC_PREFETCH_WINDOW && next_to_schedule <= remote_height { + let handle = spawn_sync_download( + next_to_schedule, + stream.clone(), + db.clone(), + map.clone(), + connections_key.clone(), + node_syncing, + ); + pending.insert(next_to_schedule, handle); + next_to_schedule += 1; + } + + let Some(handle) = pending.remove(&next_to_save) else { + return Err(io::Error::new( + io::ErrorKind::NotFound, + format!("Missing scheduled sync download for height {next_to_save}"), + )); + }; + + let download = match handle.await { + Ok(Ok(download)) => download, + Ok(Err(err)) => { + abort_pending_downloads(&mut pending); + if err.starts_with("textual torrent response") { + warn!("[sync] received textual torrent response: {err}"); + if next_to_save == 0 && !genesis_checkup().await { warn!( "[sync] remote peer has no genesis torrent; local node will attempt genesis mining" ); } break; } - } - - let torrent = Torrent::from_bytes(&torrent_bytes).await?; - match handle_response_and_save_torrent( - local_height, - db, - torrent, - wallet.clone(), - map.clone(), - false, - node_syncing, - false, - ) - .await - { - Ok(()) => {} - Err(err) => { - warn!("[sync] error saving block: height={local_height} err={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") - || err.contains("Miner wallet address is not registered") - { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!( - "sync rejected canonical peer block at height {local_height}: {err}" - ), - )); - } - - // Refresh local chain state before triggering orphan - // handling so the reconciliation logic starts from the - // real saved height. - local_height = get_height(db); - let orphan_checkup_params = OrphanCheckup2 { - stream: stream.clone(), - db: db.clone(), - local_height, - remote_height, - recheck_from_height: Some(local_height), - map: map.clone(), - node_syncing, - connections_key: connections_key.clone(), - }; - - match sync_checkup(orphan_checkup_params, wallet.clone()).await { - Ok(()) => {} - Err(err) => error!("[sync] orphan check returned error: {err}"), - } - - // Sync correction may roll back several blocks, so the - // next loop iteration needs a fresh local height. - local_height = get_height(db); + handle_sync_error( + &err, + next_to_save, + stream.clone(), + db, + remote_height, + map.clone(), + node_syncing, + wallet.clone(), + connections_key.clone(), + ) + .await?; + next_to_save = get_height(db); + if next_to_save > 0 || genesis_checkup().await { + next_to_save += 1; } + next_to_schedule = next_to_save; + continue; } - } else { - return Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - format!("No torrent response received for height {local_height}"), - )); + Err(err) => { + abort_pending_downloads(&mut pending); + return Err(io::Error::new( + io::ErrorKind::Other, + format!("Sync download task failed at height {next_to_save}: {err}"), + )); + } + }; + + if let Err(err) = verify_and_save_downloaded_block(download, wallet.clone()).await { + abort_pending_downloads(&mut pending); + handle_sync_error( + &err, + next_to_save, + stream.clone(), + db, + remote_height, + map.clone(), + node_syncing, + wallet.clone(), + connections_key.clone(), + ) + .await?; + next_to_save = get_height(db); + if next_to_save > 0 || genesis_checkup().await { + next_to_save += 1; + } + next_to_schedule = next_to_save; + continue; } - local_height += 1; + + next_to_save += 1; } + abort_pending_downloads(&mut pending); info!("[sync] node syncing complete, awaiting post-sync checks before mining resumes"); Ok(()) } diff --git a/src/rpc/commands/route_reply.rs b/src/rpc/commands/route_reply.rs index acf814c..978f965 100644 --- a/src/rpc/commands/route_reply.rs +++ b/src/rpc/commands/route_reply.rs @@ -1,7 +1,7 @@ use crate::log::warn; use crate::records::memory::enums::ClientType; use crate::records::memory::response_channels::{ - delete_entry, get_entry, is_retired_entry, trace_entry, Command, + delete_entry, get_entry, is_retired_entry, trace_entry, Command, SLOW_RPC_TRACE_MS, }; use crate::rpc::command_maps::MAX_RPC_REPLY_BYTES; use crate::rpc::read_bytes_from_stream; @@ -49,7 +49,7 @@ pub async fn route_reply( warn!("[rpc] reply receiver dropped before payload delivery: {uid:?}"); } if let Some(trace) = trace { - if trace.age_ms >= 1_000 { + if trace.age_ms >= SLOW_RPC_TRACE_MS { warn!( "[rpc_trace] slow reply routed: uid={uid:?} cmd={:?} peer={} age_ms={} payload_bytes={message_length}", trace.command, diff --git a/src/rpc/server/handshake.rs b/src/rpc/server/handshake.rs index 403c3c6..e609c42 100644 --- a/src/rpc/server/handshake.rs +++ b/src/rpc/server/handshake.rs @@ -210,20 +210,26 @@ async fn complete_incoming_miner_setup( return; } - let short_address = wallet.saved.short_address.clone(); - if let Err(err) = announce_self_to_network( - stream.clone(), - &short_address, - map.clone(), - db, - wallet.clone(), - connections_key, - ) - .await - { - error!("[startup] incoming peer network map sync failed: {err}"); - remove_key_from_memory(connections_key).await; - return; + if get_height(db) <= 10_000 { + // Before the sponsored-add gate, reverse announcement lets tiny early + // networks bootstrap from one live peer. After the gate activates, the + // incoming peer may have no local history yet, so only its outgoing + // self-announcement should be sponsored by this established node. + let short_address = wallet.saved.short_address.clone(); + if let Err(err) = announce_self_to_network( + stream.clone(), + &short_address, + map.clone(), + db, + wallet.clone(), + connections_key, + ) + .await + { + error!("[startup] incoming peer network map sync failed: {err}"); + remove_key_from_memory(connections_key).await; + return; + } } mark_peer_network_map_synced(connections_key).await; diff --git a/src/standalone_tools/connections/sending_request.rs b/src/standalone_tools/connections/sending_request.rs index 1e7e071..a5211f7 100644 --- a/src/standalone_tools/connections/sending_request.rs +++ b/src/standalone_tools/connections/sending_request.rs @@ -5,11 +5,12 @@ use crate::records::memory::response_channels::Byte3; use crate::rpc::command_maps::{ MAX_RPC_REPLY_BYTES, RPC_ADDRESS_HISTORY, RPC_ADDRESS_TOTAL_BALANCE, RPC_BLOCK_BY_HASH, RPC_BLOCK_BY_HEIGHT, RPC_BLOCK_HEIGHT, RPC_BLOCK_IP, RPC_CONTRACT_BY_ADDRESS, RPC_DIFFICULTY, - RPC_LARGEST_TX_FEE, RPC_LATEST_ADDRESS_HISTORY, RPC_LATEST_ADDRESS_TRANSACTIONS, RPC_LOAN_CONTRACT, - RPC_MEMPOOL_TX_BY_ADDRESS, RPC_MEMPOOL_TX_BY_SIGNATURE, RPC_MEMPOOL_TX_COUNT, RPC_NETWORK_INFO, - RPC_NFT_DETAILS, RPC_NFT_LIST, RPC_REGISTER_WALLET, RPC_TIME, RPC_TOKEN_DETAILS, RPC_TOKEN_LIST, - RPC_TORRENT_BY_HEIGHT, RPC_TOTAL_CONFIRMED_TX, RPC_TRANSACTION_BY_TXID, RPC_UNBLOCK_IP, - RPC_VANITY_LOOKUP, RPC_VANITY_OWNER_LOOKUP, RPC_WALLET_REGISTRATION_STATUS, + RPC_LARGEST_TX_FEE, RPC_LATEST_ADDRESS_HISTORY, RPC_LATEST_ADDRESS_TRANSACTIONS, + RPC_LOAN_CONTRACT, RPC_MEMPOOL_TX_BY_ADDRESS, RPC_MEMPOOL_TX_BY_SIGNATURE, + RPC_MEMPOOL_TX_COUNT, RPC_NETWORK_INFO, RPC_NFT_DETAILS, RPC_NFT_LIST, RPC_REGISTER_WALLET, + RPC_TIME, RPC_TOKEN_DETAILS, RPC_TOKEN_LIST, RPC_TORRENT_BY_HEIGHT, RPC_TOTAL_CONFIRMED_TX, + RPC_TRANSACTION_BY_TXID, RPC_UNBLOCK_IP, RPC_VANITY_LOOKUP, RPC_VANITY_OWNER_LOOKUP, + RPC_WALLET_REGISTRATION_STATUS, }; use crate::standalone_tools::transaction_creator::create_transaction_request; use crate::timeout; @@ -544,14 +545,20 @@ async fn build_request_bytes( 47 => { // NEW: latest address transaction bytes (RPC_LATEST_ADDRESS_TRANSACTIONS) let command_number: u8 = RPC_LATEST_ADDRESS_TRANSACTIONS; - let (address, limit) = command_input.split_once('|') - .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, - "Latest address transactions input must be address|limit"))?; + let (address, limit) = command_input.split_once('|').ok_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidInput, + "Latest address transactions input must be address|limit", + ) + })?; let address_bytes = Wallet::normalize_to_short_address(address) .and_then(|s| Wallet::short_address_to_bytes(&s)) - .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Invalid wallet address"))?; - let limit = limit.parse::() - .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "Invalid history limit"))?; + .ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "Invalid wallet address") + })?; + let limit = limit.parse::().map_err(|_| { + io::Error::new(io::ErrorKind::InvalidInput, "Invalid history limit") + })?; bin_msg.extend_from_slice(&command_number.to_le_bytes()); bin_msg.extend_from_slice(&hashmap_key); bin_msg.extend_from_slice(&address_bytes); diff --git a/src/startup/node_runtime.rs b/src/startup/node_runtime.rs index ef573cc..1e425ce 100644 --- a/src/startup/node_runtime.rs +++ b/src/startup/node_runtime.rs @@ -13,6 +13,7 @@ use crate::records::memory::chain_state::rebuild_chain_state_cache; use crate::records::memory::mempool::{init_db, setup_mempool}; use crate::records::memory::network_mapping::NodeInfo; use crate::records::memory::response_channels::Command; +use crate::records::record_chain::rewards_tx::rebuild_immature_reward_window; use crate::rpc::server::start_rpc::start_rpc; use crate::sled::Db; use crate::startup::connections::handle_connections; @@ -112,6 +113,9 @@ pub async fn run_unlocked_node(wallet: Arc, install_shutdown: bool) -> R if let Err(err) = rebuild_chain_state_cache(&db).await { error!("[chain_state] failed to rebuild startup chain state cache: {err}"); } + if let Err(err) = rebuild_immature_reward_window(&db).await { + error!("[rewards] failed to rebuild immature reward window: {err}"); + } if install_shutdown { // Console/daemon mode owns signal cleanup; Windows service shutdown is handled separately. diff --git a/src/torrent/create_metadata.rs b/src/torrent/create_metadata.rs index ccbdf66..daebe8a 100644 --- a/src/torrent/create_metadata.rs +++ b/src/torrent/create_metadata.rs @@ -2,6 +2,7 @@ use crate::blocks::block::{NONCE_OFFSET, VRF_OFFSET}; use crate::common::network_paths_and_settings::block_extension_and_paths; use crate::common::skein::skein_128_hash_bytes; use crate::log::error; +use crate::miner::flag::is_syncing_mode; use crate::records::memory::response_channels::{reserve_transient_entry_with_context, Command}; use crate::rpc::command_maps::RPC_SUBMIT_TORRENT; use crate::rpc::responses::RpcResponse; @@ -143,6 +144,10 @@ pub async fn broadcast_new_torrent_to_peers( torrent_bytes: &[u8], map: Arc>, ) { + if is_syncing_mode() { + return; + } + // Command byte for "submit torrent". let command: u8 = RPC_SUBMIT_TORRENT; diff --git a/src/torrent/structs.rs b/src/torrent/structs.rs index f42706f..afae2de 100644 --- a/src/torrent/structs.rs +++ b/src/torrent/structs.rs @@ -23,6 +23,8 @@ pub struct DownloadedPieceJob { pub db: Db, pub torrent: Torrent, pub torrent_map: Arc>, + pub in_memory_pieces: Arc>>>, + pub use_in_memory_pieces: bool, pub block_number: u32, pub piece: u8, pub ip: String, @@ -47,6 +49,8 @@ pub struct RequestPiece { #[derive(Clone)] pub struct DownloadSave { pub torrent_map: Arc>, + pub in_memory_pieces: Arc>>>, + pub use_in_memory_pieces: bool, pub torrent: Torrent, pub staged_path: String, pub block_number: u32, diff --git a/src/torrent/torrenting_system/download_pieces.rs b/src/torrent/torrenting_system/download_pieces.rs index 4c7b02a..1644474 100644 --- a/src/torrent/torrenting_system/download_pieces.rs +++ b/src/torrent/torrenting_system/download_pieces.rs @@ -103,24 +103,29 @@ async fn handle_downloaded_piece(job: DownloadedPieceJob, data: Vec) { }; if received_hash == expected_hash { - if let Err(err) = save_piece_to_db( - job.db.clone(), - job.block_number, - &job.torrent.info.info_hash, - job.piece, - &data, - ) - .await - { - error!( - "[download] failed to stage piece data: block_number={} piece={} err={}", - job.block_number, job.piece, err - ); - mark_piece_failed(job.torrent_map, job.piece, &job.ip).await; - return; + if job.use_in_memory_pieces { + let mut pieces = job.in_memory_pieces.lock().await; + pieces.insert(job.piece, data); + } else { + if let Err(err) = save_piece_to_db( + job.db.clone(), + job.block_number, + &job.torrent.info.info_hash, + job.piece, + &data, + ) + .await + { + error!( + "[download] failed to stage piece data: block_number={} piece={} err={}", + job.block_number, job.piece, err + ); + mark_piece_failed(job.torrent_map, job.piece, &job.ip).await; + return; + } } - // A hash match is only complete after the piece bytes are staged for final assembly. + // A hash match is only complete after the piece bytes are available for final assembly. { let mut torrent_map = job.torrent_map.lock().await; let _ = torrent_map.mark_piece_complete(job.piece); @@ -168,7 +173,9 @@ pub async fn download_block_pieces(params: DownloadSave) -> Result<(), String> { let pieces: Vec = (1..=piece_count) .map(|i| u8::try_from(i).map_err(|_| "Torrent piece index exceeds u8 limit".to_string())) .collect::, _>>()?; - hydrate_cached_pieces(¶ms, &pieces).await?; + if !params.use_in_memory_pieces { + hydrate_cached_pieces(¶ms, &pieces).await?; + } // Piece requests send the 128-bit info hash as little-endian bytes. let info_hash_bytes = decode(¶ms.torrent.info.info_hash) @@ -197,12 +204,14 @@ pub async fn download_block_pieces(params: DownloadSave) -> Result<(), String> { current_height, expected_height ); - let _ = remove_block_pieces_from_db( - ¶ms.db, - params.block_number, - ¶ms.torrent.info.info_hash, - ) - .await; + if !params.use_in_memory_pieces { + let _ = remove_block_pieces_from_db( + ¶ms.db, + params.block_number, + ¶ms.torrent.info.info_hash, + ) + .await; + } return Ok(()); } @@ -268,6 +277,8 @@ pub async fn download_block_pieces(params: DownloadSave) -> Result<(), String> { db, torrent, torrent_map, + in_memory_pieces: params.in_memory_pieces.clone(), + use_in_memory_pieces: params.use_in_memory_pieces, block_number, piece, ip: peer_key, diff --git a/src/torrent/torrenting_system/save_block.rs b/src/torrent/torrenting_system/save_block.rs index 388432c..1720bc5 100644 --- a/src/torrent/torrenting_system/save_block.rs +++ b/src/torrent/torrenting_system/save_block.rs @@ -11,9 +11,17 @@ use crate::torrent::structs::DownloadSave; use crate::torrent::torrenting_system::create_file::combine_pieces; use crate::torrent::torrenting_system::save_torrent::promote_staged_torrent; use crate::torrent::torrenting_system::temp_database_storage::remove_block_pieces_from_db; +use crate::wallets::structures::Wallet; +use crate::Arc; use crate::Path; async fn cleanup_download_pieces(params: &DownloadSave) { + if params.use_in_memory_pieces { + let mut pieces = params.in_memory_pieces.lock().await; + pieces.clear(); + return; + } + // Failed, obsolete, or deferred downloads should not leave staged piece rows behind. let _ = remove_block_pieces_from_db( ¶ms.db, @@ -23,45 +31,32 @@ async fn cleanup_download_pieces(params: &DownloadSave) { .await; } -pub async fn verify_and_save_block(params: DownloadSave) -> Result<(), String> { - if is_reorganizing_mode() && !params.allow_during_reorg { - // Normal torrent downloads pause during reorg unless the caller is the reorg path itself. - warn!( - "[download] deferring verify/save during reorg: block_number={}", - params.block_number - ); - return Err("Block download/save deferred while reorganizing.".to_string()); +async fn combine_in_memory_pieces(params: &DownloadSave) -> Result, String> { + let piece_count = params.torrent.info.pieces.len(); + if piece_count > u8::MAX as usize { + return Err("Torrent piece count exceeds u8 limit".to_string()); } - // Stop early if the chain has already advanced past this block while - // the download was still in progress. - let current_height = get_height(¶ms.db); - let expected_height = if current_height > 0 || genesis_checkup().await { - current_height + 1 - } else { - current_height - }; - - if params.block_number < expected_height { - warn!( - "[download] block is no longer next expected height: block_number={} current_height={} expected_height={}", - params.block_number, - current_height, - expected_height - ); - cleanup_download_pieces(¶ms).await; - return Err("Incoming block is no longer the next expected height.".to_string()); + let pieces = params.in_memory_pieces.lock().await; + let mut combined_data = Vec::with_capacity(params.torrent.info.length as usize); + for piece_number in 1..=piece_count { + let piece_number = u8::try_from(piece_number) + .map_err(|_| "Torrent piece index exceeds u8 limit".to_string())?; + let data = pieces.get(&piece_number).ok_or_else(|| { + format!( + "Downloaded in-memory piece missing: block_number={} piece={piece_number}", + params.block_number + ) + })?; + combined_data.extend_from_slice(data); } + Ok(combined_data) +} - // Reassemble the downloaded pieces and verify the combined payload - // against the info hash advertised in the torrent metadata. - let result = combine_pieces( - ¶ms.db, - params.block_number, - ¶ms.torrent.info.info_hash, - ) - .await?; - +async fn verify_and_save_block_payload( + params: DownloadSave, + result: Vec, +) -> Result<(), String> { if result.len() != params.torrent.info.length as usize { return Err("Downloaded candidate length does not match torrent metadata.".to_string()); } @@ -162,3 +157,77 @@ pub async fn verify_and_save_block(params: DownloadSave) -> Result<(), String> { Ok(()) } + +async fn ensure_next_expected_height(params: &DownloadSave) -> Result<(), String> { + if is_reorganizing_mode() && !params.allow_during_reorg { + // Normal torrent downloads pause during reorg unless the caller is the reorg path itself. + warn!( + "[download] deferring verify/save during reorg: block_number={}", + params.block_number + ); + return Err("Block download/save deferred while reorganizing.".to_string()); + } + + // Stop early if the chain has already advanced past this block while + // the download was still in progress. + let current_height = get_height(¶ms.db); + let expected_height = if current_height > 0 || genesis_checkup().await { + current_height + 1 + } else { + current_height + }; + + if params.block_number < expected_height { + warn!( + "[download] block is no longer next expected height: block_number={} current_height={} expected_height={}", + params.block_number, + current_height, + expected_height + ); + cleanup_download_pieces(¶ms).await; + return Err("Incoming block is no longer the next expected height.".to_string()); + } + Ok(()) +} + +pub async fn verify_and_save_downloaded_block( + params: DownloadSave, + wallet: Arc, +) -> Result<(), String> { + ensure_next_expected_height(¶ms).await?; + params + .torrent + .verify(params.block_number, ¶ms.db, wallet) + .await?; + + let result = if params.use_in_memory_pieces { + combine_in_memory_pieces(¶ms).await? + } else { + combine_pieces( + ¶ms.db, + params.block_number, + ¶ms.torrent.info.info_hash, + ) + .await? + }; + verify_and_save_block_payload(params, result).await +} + +pub async fn verify_and_save_block(params: DownloadSave) -> Result<(), String> { + ensure_next_expected_height(¶ms).await?; + + // Reassemble the downloaded pieces and verify the combined payload + // against the info hash advertised in the torrent metadata. + let result = if params.use_in_memory_pieces { + combine_in_memory_pieces(¶ms).await? + } else { + combine_pieces( + ¶ms.db, + params.block_number, + ¶ms.torrent.info.info_hash, + ) + .await? + }; + + verify_and_save_block_payload(params, result).await +} diff --git a/src/torrent/torrenting_system/setup_block_download.rs b/src/torrent/torrenting_system/setup_block_download.rs index 27e584f..aa821e0 100644 --- a/src/torrent/torrenting_system/setup_block_download.rs +++ b/src/torrent/torrenting_system/setup_block_download.rs @@ -1,3 +1,4 @@ +use crate::miner::flag::is_syncing_mode; use crate::records::memory::response_channels::Command; use crate::sled::Db; use crate::torrent::structs::{DownloadSave, Torrent}; @@ -7,6 +8,7 @@ use crate::torrent::torrenting_system::save_block::verify_and_save_block; use crate::torrent::torrenting_system::torrent_map::create_torrent_map; use crate::verifications::verification_service::VerificationService; use crate::Arc; +use crate::HashMap; use crate::Mutex; // Build the per-piece download state, fetch the block pieces, and then @@ -27,11 +29,14 @@ pub async fn setup_download( // Initialize in-memory status tracking for each piece in the torrent. let torrent_map = create_torrent_map(&torrent).await?; + let use_in_memory_pieces = is_syncing_mode() && !allow_during_reorg; // Carry the shared download context through the piece download and // final block verification stages. let download_save_params = DownloadSave { torrent_map, + in_memory_pieces: Arc::new(Mutex::new(HashMap::new())), + use_in_memory_pieces, torrent, staged_path, block_number, @@ -47,3 +52,35 @@ pub async fn setup_download( // The same context is reused to assemble, verify, save, and promote the staged torrent. verify_and_save_block(download_save_params).await } + +pub async fn download_sync_candidate( + block_number: u32, + torrent: Torrent, + staged_path: String, + allow_startup_peers: bool, + db: Db, + verification_service: Arc, + map: Arc>, +) -> Result { + let _download_guard = + acquire_candidate_download(block_number, &torrent.info.info_hash, false).await?; + + let torrent_map = create_torrent_map(&torrent).await?; + let download_save_params = DownloadSave { + torrent_map, + in_memory_pieces: Arc::new(Mutex::new(HashMap::new())), + use_in_memory_pieces: true, + torrent, + staged_path, + block_number, + allow_during_reorg: false, + allow_historical: false, + allow_startup_peers, + db, + verification_service, + map, + }; + + download_block_pieces(download_save_params.clone()).await?; + Ok(download_save_params) +} diff --git a/src/verifications/async_funcs/verify_block.rs b/src/verifications/async_funcs/verify_block.rs index a545a5b..f02f5cc 100644 --- a/src/verifications/async_funcs/verify_block.rs +++ b/src/verifications/async_funcs/verify_block.rs @@ -150,7 +150,7 @@ impl Block { // validate that the current timestamp is greater the previous // block timestamp plus 2 seconds let current_timestamp = Utc::now().timestamp() as u32; - if current_timestamp < previous_block.unmined_block.timestamp + 2 { + if current_timestamp < previous_block.unmined_block.timestamp + 1 { return Err("Mining to quickly".to_string()); }