syncing fixes

.gitignore

@@ -8,3 +8,4 @@ wallets/
 *.wallet
 *.key
 *.pdb
+*.zip

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<Torrent> {
     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,

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.

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,

src/records/memory/response_channels.rs

@@ -20,7 +20,7 @@ pub type Byte3 = [u8; 3];
 
 pub type Command = HashMap<Byte3, ChannelPair>;
 
-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();

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<VecDeque<ImmatureReward>> = 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")
-    for block_height in 0..=cutoff_height {
+        .map_err(|e| format!("Failed to store finalized reward marker: {e}"))?;
-        if tree
+    Ok(())
-            .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(
+                rebuilt.push_back(ImmatureReward {
-                    db,
+                    block_height,
-                    &miner,
+                    miner,
-                    rewards.unsigned.value,
+                    amount: 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}"))?;
                 break;
             }
         }
     }
 
+    while rebuilt.len() > IMMATURE_REWARD_WINDOW {
+        rebuilt.pop_front();
+    }
+
+    let mut rewards = IMMATURE_REWARDS.lock().await;
+    *rewards = rebuilt;
     Ok(())
 }
 

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,9 +217,18 @@ pub async fn save_block(params: SaveBlockParams) -> Result<(), String> {
             map,
         })
         .await?;
+    }
+
+    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,7 +469,14 @@ async fn save_binary_data(params: SaveBinaryDataParams<'_>) -> Result<(), String
         return Err(err);
     }
 
-    let torrent_bytes = match metadata_from_file(
+    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,
@@ -469,7 +487,7 @@ async fn save_binary_data(params: SaveBinaryDataParams<'_>) -> Result<(), String
         )
         .await
         {
-        Ok(torrent_bytes) => torrent_bytes,
+            Ok(torrent_bytes) => Some(torrent_bytes),
             Err(err) => {
                 cleanup_block_indexes(db, block_header_number, header_hash);
                 cleanup_torrent_file(next_number);
@@ -480,6 +498,7 @@ async fn save_binary_data(params: SaveBinaryDataParams<'_>) -> Result<(), String
                 cleanup_block_file(&file_name);
                 return Err(err);
             }
+        }
     };
 
     if next_number != 0 {
@@ -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;
+            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;
     }
 

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::{
+use crate::torrent::torrenting_system::save_block::verify_and_save_downloaded_block;
-    handle_response_and_save_torrent, send_request_torrent_message,
+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<Result<crate::torrent::structs::DownloadSave, String>>,
+    >,
+) {
+    for (_, handle) in pending.iter() {
+        handle.abort();
+    }
+    pending.clear();
+}
+
+fn spawn_sync_download(
+    height: u32,
+    stream: Arc<Mutex<TcpStream>>,
+    db: Db,
+    map: Arc<Mutex<Command>>,
+    connections_key: String,
+    allow_startup_peers: bool,
+) -> crate::tokio::task::JoinHandle<Result<crate::torrent::structs::DownloadSave, String>> {
+    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<Mutex<TcpStream>>,
+    db: &Db,
+    remote_height: u32,
+    map: Arc<Mutex<Command>>,
+    node_syncing: bool,
+    wallet: Arc<Wallet>,
+    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<Mutex<TcpStream>>,
@@ -35,112 +157,100 @@ pub async fn node_syncing(
         local_height += 1;
     }
 
-    // Walk forward block-by-block until the local chain catches up to
+    let mut next_to_schedule = local_height;
-    // the advertised remote height.
+    let mut next_to_save = local_height;
-    while remote_height >= local_height {
+    let mut pending = BTreeMap::new();
-        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?;
 
-        // Wait for the requested torrent bytes to come back through the
+    // Download a small window of candidates ahead, but validate/save only
-        // shared response map entry created for this request.
+    // the exact next height so parent, difficulty, and balance rules still
-        let mut rx = torrent_rx.lock().await;
+    // see the canonical local chain state.
-        if let Some(torrent_bytes) =
+    while remote_height >= next_to_save {
-            timeout(Duration::from_secs(30), rx.recv())
+        while pending.len() < SYNC_PREFETCH_WINDOW && next_to_schedule <= remote_height {
-                .await
+            let handle = spawn_sync_download(
-                .map_err(|_| {
+                next_to_schedule,
-                    io::Error::new(
+                stream.clone(),
-                        io::ErrorKind::TimedOut,
+                db.clone(),
-                        "Timed out waiting for torrent response",
+                map.clone(),
-                    )
+                connections_key.clone(),
-                })?
+                node_syncing,
-        {
+            );
-            if let Ok(text) = String::from_utf8(torrent_bytes.clone()) {
+            pending.insert(next_to_schedule, handle);
-                let trimmed = text.trim();
+            next_to_schedule += 1;
-                if !trimmed.is_empty() {
+        }
-                    warn!("[sync] received textual torrent response at height={local_height} err={trimmed}");
+
-                    if local_height == 0 && !genesis_checkup().await {
+        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;
                 }
-            }
+                handle_sync_error(
-
+                    &err,
-            let torrent = Torrent::from_bytes(&torrent_bytes).await?;
+                    next_to_save,
-            match handle_response_and_save_torrent(
+                    stream.clone(),
-                local_height,
                     db,
-                torrent,
+                    remote_height,
-                wallet.clone(),
                     map.clone(),
-                false,
                     node_syncing,
-                false,
+                    wallet.clone(),
+                    connections_key.clone(),
                 )
-            .await
+                .await?;
-            {
+                next_to_save = get_height(db);
-                Ok(()) => {}
+                if next_to_save > 0 || genesis_checkup().await {
+                    next_to_save += 1;
+                }
+                next_to_schedule = next_to_save;
+                continue;
+            }
             Err(err) => {
-                    warn!("[sync] error saving block: height={local_height} err={err}");
+                abort_pending_downloads(&mut pending);
-
-                    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,
+                    io::ErrorKind::Other,
-                            format!(
+                    format!("Sync download task failed at height {next_to_save}: {err}"),
-                                "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 {
+        if let Err(err) = verify_and_save_downloaded_block(download, wallet.clone()).await {
-                        Ok(()) => {}
+            abort_pending_downloads(&mut pending);
-                        Err(err) => error!("[sync] orphan check returned error: {err}"),
+            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;
         }
 
-                    // Sync correction may roll back several blocks, so the
+        next_to_save += 1;
-                    // next loop iteration needs a fresh local height.
-                    local_height = get_height(db);
-                }
-            }
-        } else {
-            return Err(io::Error::new(
-                io::ErrorKind::UnexpectedEof,
-                format!("No torrent response received for height {local_height}"),
-            ));
-        }
-        local_height += 1;
     }
+    abort_pending_downloads(&mut pending);
     info!("[sync] node syncing complete, awaiting post-sync checks before mining resumes");
     Ok(())
 }

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,

src/rpc/server/handshake.rs

@@ -210,6 +210,11 @@ async fn complete_incoming_miner_setup(
         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(),
@@ -225,6 +230,7 @@ async fn complete_incoming_miner_setup(
             remove_key_from_memory(connections_key).await;
             return;
         }
+    }
     mark_peer_network_map_synced(connections_key).await;
 
     let operational = match sync_incoming_peer_before_operational(

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_LARGEST_TX_FEE, RPC_LATEST_ADDRESS_HISTORY, RPC_LATEST_ADDRESS_TRANSACTIONS,
-    RPC_MEMPOOL_TX_BY_ADDRESS, RPC_MEMPOOL_TX_BY_SIGNATURE, RPC_MEMPOOL_TX_COUNT, RPC_NETWORK_INFO, 
+    RPC_LOAN_CONTRACT, RPC_MEMPOOL_TX_BY_ADDRESS, RPC_MEMPOOL_TX_BY_SIGNATURE,
-    RPC_NFT_DETAILS, RPC_NFT_LIST, RPC_REGISTER_WALLET, RPC_TIME, RPC_TOKEN_DETAILS, RPC_TOKEN_LIST,
+    RPC_MEMPOOL_TX_COUNT, RPC_NETWORK_INFO, RPC_NFT_DETAILS, RPC_NFT_LIST, RPC_REGISTER_WALLET,
-    RPC_TORRENT_BY_HEIGHT, RPC_TOTAL_CONFIRMED_TX, RPC_TRANSACTION_BY_TXID, RPC_UNBLOCK_IP,
+    RPC_TIME, RPC_TOKEN_DETAILS, RPC_TOKEN_LIST, RPC_TORRENT_BY_HEIGHT, RPC_TOTAL_CONFIRMED_TX,
-    RPC_VANITY_LOOKUP, RPC_VANITY_OWNER_LOOKUP, RPC_WALLET_REGISTRATION_STATUS,
+    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('|')
+            let (address, limit) = command_input.split_once('|').ok_or_else(|| {
-                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput,
+                io::Error::new(
-                                              "Latest address transactions input must be address|limit"))?;
+                    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"))?;
+                .ok_or_else(|| {
-            let limit = limit.parse::<u32>()
+                    io::Error::new(io::ErrorKind::InvalidInput, "Invalid wallet address")
-                .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "Invalid history limit"))?;
+                })?;
+            let limit = limit.parse::<u32>().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);

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<Wallet>, 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.

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<Mutex<Command>>,
 ) {
+    if is_syncing_mode() {
+        return;
+    }
+
     // Command byte for "submit torrent".
     let command: u8 = RPC_SUBMIT_TORRENT;
 

src/torrent/structs.rs

@@ -23,6 +23,8 @@ pub struct DownloadedPieceJob {
     pub db: Db,
     pub torrent: Torrent,
     pub torrent_map: Arc<Mutex<TorrentMap>>,
+    pub in_memory_pieces: Arc<Mutex<HashMap<u8, Vec<u8>>>>,
+    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<Mutex<TorrentMap>>,
+    pub in_memory_pieces: Arc<Mutex<HashMap<u8, Vec<u8>>>>,
+    pub use_in_memory_pieces: bool,
     pub torrent: Torrent,
     pub staged_path: String,
     pub block_number: u32,

src/torrent/torrenting_system/download_pieces.rs

@@ -103,6 +103,10 @@ async fn handle_downloaded_piece(job: DownloadedPieceJob, data: Vec<u8>) {
     };
 
     if received_hash == expected_hash {
+        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,
@@ -119,8 +123,9 @@ async fn handle_downloaded_piece(job: DownloadedPieceJob, data: Vec<u8>) {
                 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<u8> = (1..=piece_count)
         .map(|i| u8::try_from(i).map_err(|_| "Torrent piece index exceeds u8 limit".to_string()))
         .collect::<Result<Vec<_>, _>>()?;
+    if !params.use_in_memory_pieces {
         hydrate_cached_pieces(&params, &pieces).await?;
+    }
 
     // Piece requests send the 128-bit info hash as little-endian bytes.
     let info_hash_bytes = decode(&params.torrent.info.info_hash)
@@ -197,12 +204,14 @@ pub async fn download_block_pieces(params: DownloadSave) -> Result<(), String> {
                 current_height,
                 expected_height
             );
+            if !params.use_in_memory_pieces {
                 let _ = remove_block_pieces_from_db(
                     &params.db,
                     params.block_number,
                     &params.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,

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(
         &params.db,
@@ -23,45 +31,32 @@ async fn cleanup_download_pieces(params: &DownloadSave) {
     .await;
 }
 
-pub async fn verify_and_save_block(params: DownloadSave) -> Result<(), String> {
+async fn combine_in_memory_pieces(params: &DownloadSave) -> Result<Vec<u8>, String> {
-    if is_reorganizing_mode() && !params.allow_during_reorg {
+    let piece_count = params.torrent.info.pieces.len();
-        // Normal torrent downloads pause during reorg unless the caller is the reorg path itself.
+    if piece_count > u8::MAX as usize {
-        warn!(
+        return Err("Torrent piece count exceeds u8 limit".to_string());
-            "[download] deferring verify/save during reorg: block_number={}",
+    }
+
+    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
-        );
-        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(&params.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(&params).await;
-        return Err("Incoming block is no longer the next expected height.".to_string());
-    }
-
-    // Reassemble the downloaded pieces and verify the combined payload
-    // against the info hash advertised in the torrent metadata.
-    let result = combine_pieces(
-        &params.db,
-        params.block_number,
-        &params.torrent.info.info_hash,
             )
-    .await?;
+        })?;
+        combined_data.extend_from_slice(data);
+    }
+    Ok(combined_data)
+}
 
+async fn verify_and_save_block_payload(
+    params: DownloadSave,
+    result: Vec<u8>,
+) -> 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(&params.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(&params).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<Wallet>,
+) -> Result<(), String> {
+    ensure_next_expected_height(&params).await?;
+    params
+        .torrent
+        .verify(params.block_number, &params.db, wallet)
+        .await?;
+
+    let result = if params.use_in_memory_pieces {
+        combine_in_memory_pieces(&params).await?
+    } else {
+        combine_pieces(
+            &params.db,
+            params.block_number,
+            &params.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(&params).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(&params).await?
+    } else {
+        combine_pieces(
+            &params.db,
+            params.block_number,
+            &params.torrent.info.info_hash,
+        )
+        .await?
+    };
+
+    verify_and_save_block_payload(params, result).await
+}

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<VerificationService>,
+    map: Arc<Mutex<Command>>,
+) -> Result<DownloadSave, String> {
+    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)
+}

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