startup sync bug fixes

src/orphans/orphan_checkup.rs

@@ -120,6 +120,7 @@ async fn candidate_attaches_before_rollback(
         block_number: height,
         allow_during_reorg: true,
         allow_historical: true,
+        historical_sync: true,
         allow_startup_peers: params.node_syncing,
         db: params.db.clone(),
         verification_service: std::sync::Arc::new(verification_service),

src/orphans/save_blocks.rs

@@ -12,7 +12,7 @@ use crate::torrent::torrenting_system::save_torrent::{
     list_staged_torrents_for_height, read_staged_torrent,
 };
 use crate::torrent::torrenting_system::torrent_requests::{
-    handle_response_and_save_torrent, send_request_torrent_message,
+    handle_historical_response_and_save_torrent, send_request_torrent_message,
 };
 use crate::wallets::structures::Wallet;
 use crate::Arc;
@@ -69,7 +69,7 @@ pub async fn save_new_blocks(
             // Height advancement is the proof that the candidate actually
             // extended the chain rather than merely parsing successfully.
             let local_height_before = get_height(&params.db);
-            match handle_response_and_save_torrent(
+            match handle_historical_response_and_save_torrent(
                 true_start_height,
                 &params.db,
                 torrent.clone(),
@@ -186,7 +186,7 @@ pub async fn save_new_blocks(
             let local_height_before = get_height(&params.db);
             // Save through the normal torrent path so all validation and
             // record updates stay identical to a live broadcast.
-            handle_response_and_save_torrent(
+            handle_historical_response_and_save_torrent(
                 true_start_height,
                 &params.db,
                 torrent,

src/orphans/sync_check.rs

@@ -21,7 +21,7 @@ use crate::torrent::structs::Torrent;
 use crate::torrent::torrenting_system::save_torrent::{
     list_staged_torrents, read_staged_torrent, remove_staged_torrent,
 };
-use crate::torrent::torrenting_system::torrent_requests::handle_response_and_save_torrent;
+use crate::torrent::torrenting_system::torrent_requests::handle_historical_response_and_save_torrent;
 use crate::wallets::structures::Wallet;
 use crate::Arc;
 
@@ -119,7 +119,7 @@ async fn replay_staged_torrents(
 
             // Reuse the normal torrent save/verify pipeline; staged replay
             // should behave exactly like receiving the torrent live.
-            match handle_response_and_save_torrent(
+            match handle_historical_response_and_save_torrent(
                 expected_height,
                 &params.db,
                 torrent,

src/records/memory/network_mapping/queries.rs

@@ -18,6 +18,36 @@ impl NodeInfo {
             || (node_info.deleted_block > 0 && block_number < node_info.deleted_block)
     }
 
+    fn eligible_at_historical_sync_block(
+        address: &str,
+        node_info: &NodeInfo,
+        block_number: u32,
+        block_timestamp: u32,
+    ) -> bool {
+        if Self::eligible_at_block(node_info, block_number) {
+            return true;
+        }
+
+        let block_timestamp_ms = (block_timestamp as u64).saturating_mul(1_000);
+
+        // A node can mine a block and disconnect before another peer receives
+        // that block. Deletion records use the current chain height, so the
+        // valid block and the later deletion can share a height. During
+        // historical replay, the timestamps disambiguate that boundary.
+        if node_info.deleted_block > 0
+            && block_number == node_info.deleted_block
+            && block_timestamp_ms < node_info.deleted_timestamp
+        {
+            return true;
+        }
+
+        // A reconnect replaces the current membership with an inactive
+        // pending lifecycle. That current state must not invalidate blocks
+        // mined before the reconnect occurred.
+        Self::can_pending_member_add_monitor(address, node_info)
+            && block_timestamp_ms < node_info.added_timestamp
+    }
+
     pub async fn has_reciprocal_sponsorship(
         local_address: &str,
         operational_peer_wallets: &[String],
@@ -53,8 +83,7 @@ impl NodeInfo {
             .await
             .get(address)
             .map(|node| {
-                node.deleted_timestamp == 0
+                node.deleted_timestamp == 0 || Self::can_pending_member_add_monitor(address, node)
-                    || Self::can_pending_member_add_monitor(address, node)
             })
             .unwrap_or(false)
     }
@@ -318,6 +347,24 @@ impl NodeInfo {
         false
     }
 
+    pub async fn historical_sync_address_checkup(
+        address: &str,
+        block_number: u32,
+        block_timestamp: u32,
+    ) -> bool {
+        let map = ADDRESS_MAP.lock().await;
+        map.get(address)
+            .map(|node_info| {
+                Self::eligible_at_historical_sync_block(
+                    address,
+                    node_info,
+                    block_number,
+                    block_timestamp,
+                )
+            })
+            .unwrap_or(false)
+    }
+
     pub async fn find_address_by_ip(ip: &str) -> Option<String> {
         let map = ADDRESS_MAP.lock().await;
         for (address, node_info) in map.iter() {
@@ -485,6 +532,74 @@ mod tests {
         assert!(NodeInfo::eligible_at_block(&pending, 10_000));
     }
 
+    #[test]
+    fn historical_sync_accepts_only_blocks_before_pending_reconnect() {
+        let address = "node.cltc";
+        let mut pending = NodeInfo::new(
+            "198.51.100.10".to_string(),
+            50050,
+            0,
+            "sponsor.cltc".to_string(),
+            300_000,
+            "signature".to_string(),
+        );
+        pending.deleted_timestamp = 300_000;
+        pending.deleted_block = 0;
+
+        assert!(NodeInfo::eligible_at_historical_sync_block(
+            address, &pending, 10, 299
+        ));
+        assert!(!NodeInfo::eligible_at_historical_sync_block(
+            address, &pending, 11, 300
+        ));
+        assert!(!NodeInfo::eligible_at_block(&pending, 10));
+    }
+
+    #[test]
+    fn historical_sync_accepts_block_mined_before_same_height_deletion() {
+        let address = "node.cltc";
+        let mut deleted = NodeInfo::new(
+            "198.51.100.10".to_string(),
+            50050,
+            0,
+            "sponsor.cltc".to_string(),
+            100_000,
+            "signature".to_string(),
+        );
+        deleted.deleted_timestamp = 300_000;
+        deleted.deleted_block = 50;
+
+        assert!(!NodeInfo::eligible_at_block(&deleted, 50));
+        assert!(NodeInfo::eligible_at_historical_sync_block(
+            address, &deleted, 50, 299
+        ));
+        assert!(!NodeInfo::eligible_at_historical_sync_block(
+            address, &deleted, 50, 300
+        ));
+        assert!(!NodeInfo::eligible_at_historical_sync_block(
+            address, &deleted, 51, 299
+        ));
+    }
+
+    #[test]
+    fn historical_sync_does_not_activate_self_sponsored_pending_member() {
+        let address = "node.cltc";
+        let mut pending = NodeInfo::new(
+            "198.51.100.10".to_string(),
+            50050,
+            0,
+            address.to_string(),
+            300_000,
+            "signature".to_string(),
+        );
+        pending.deleted_timestamp = 300_000;
+        pending.deleted_block = 0;
+
+        assert!(!NodeInfo::eligible_at_historical_sync_block(
+            address, &pending, 10, 299
+        ));
+    }
+
     #[tokio::test]
     async fn bootstrap_snapshot_keeps_deletion_and_monitor_state() {
         let address = "1111111111111111111111111111111111111111.cltc";

src/rpc/commands/receive_torrent.rs

@@ -280,6 +280,7 @@ pub async fn torrent_submission(
                         staged_path,
                         false,
                         false,
+                        false,
                         db_clone.clone(),
                         map_for_download,
                     )

src/torrent/structs.rs

@@ -57,6 +57,7 @@ pub struct DownloadSave {
     pub block_number: u32,
     pub allow_during_reorg: bool,
     pub allow_historical: bool,
+    pub historical_sync: bool,
     pub allow_startup_peers: bool,
     pub db: Db,
     pub verification_service: Arc<VerificationService>,

src/torrent/torrenting_system/save_block.rs

@@ -100,10 +100,16 @@ async fn verify_and_save_block_payload(
     }
 
     // Run full block verification before allowing the chain save path to persist the downloaded block.
-    let signatures = match loaded_block
+    let verification = if params.historical_sync {
-        .verify(&params.db, params.verification_service.clone())
+        loaded_block
-        .await
+            .verify_historical_sync(&params.db, params.verification_service.clone())
-    {
+            .await
+    } else {
+        loaded_block
+            .verify(&params.db, params.verification_service.clone())
+            .await
+    };
+    let signatures = match verification {
         Ok(signatures) => signatures,
         Err(err) => {
             error!(

src/torrent/torrenting_system/setup_block_download.rs

@@ -18,6 +18,7 @@ pub async fn setup_download(
     torrent: Torrent,
     staged_path: String,
     allow_during_reorg: bool,
+    historical_sync: bool,
     allow_startup_peers: bool,
     db: Db,
     verification_service: Arc<VerificationService>,
@@ -43,6 +44,7 @@ pub async fn setup_download(
         block_number,
         allow_during_reorg,
         allow_historical: false,
+        historical_sync,
         allow_startup_peers,
         db: db.clone(),
         verification_service,
@@ -78,6 +80,7 @@ pub async fn download_sync_candidate(
         block_number,
         allow_during_reorg: false,
         allow_historical: false,
+        historical_sync: true,
         allow_startup_peers,
         db,
         verification_service,

src/torrent/torrenting_system/torrent_requests.rs

@@ -43,6 +43,55 @@ pub async fn handle_response_and_save_torrent(
     allow_during_reorg: bool,
     allow_startup_peers: bool,
     rebroadcast: bool,
+) -> Result<(), String> {
+    handle_response_and_save_torrent_internal(
+        height,
+        db,
+        torrent,
+        wallet,
+        map,
+        allow_during_reorg,
+        allow_startup_peers,
+        rebroadcast,
+        false,
+    )
+    .await
+}
+
+pub async fn handle_historical_response_and_save_torrent(
+    height: u32,
+    db: &Db,
+    torrent: Torrent,
+    wallet: Arc<Wallet>,
+    map: Arc<Mutex<Command>>,
+    allow_during_reorg: bool,
+    allow_startup_peers: bool,
+    rebroadcast: bool,
+) -> Result<(), String> {
+    handle_response_and_save_torrent_internal(
+        height,
+        db,
+        torrent,
+        wallet,
+        map,
+        allow_during_reorg,
+        allow_startup_peers,
+        rebroadcast,
+        true,
+    )
+    .await
+}
+
+async fn handle_response_and_save_torrent_internal(
+    height: u32,
+    db: &Db,
+    torrent: Torrent,
+    wallet: Arc<Wallet>,
+    map: Arc<Mutex<Command>>,
+    allow_during_reorg: bool,
+    allow_startup_peers: bool,
+    rebroadcast: bool,
+    historical_sync: bool,
 ) -> Result<(), String> {
     let Some((torrent, staged_path)) =
         stage_and_verify_torrent(height, db, torrent, wallet, true).await?
@@ -56,6 +105,7 @@ pub async fn handle_response_and_save_torrent(
         torrent,
         staged_path,
         allow_during_reorg,
+        historical_sync,
         allow_startup_peers,
         db.clone(),
         map.clone(),
@@ -101,6 +151,7 @@ pub async fn process_torrent_response(params: ProcessTorrentResponse) -> Result<
         staged_path,
         params.allow_during_reorg,
         false,
+        false,
         params.db,
         params.map.clone(),
     )
@@ -146,6 +197,7 @@ pub async fn setup_download_for_torrent(
     torrent: Torrent,
     staged_path: String,
     allow_during_reorg: bool,
+    historical_sync: bool,
     allow_startup_peers: bool,
     db: Db,
     map: Arc<Mutex<Command>>,
@@ -160,6 +212,7 @@ pub async fn setup_download_for_torrent(
         torrent,
         staged_path,
         allow_during_reorg,
+        historical_sync,
         allow_startup_peers,
         db,
         Arc::new(verification_service),

src/verifications/async_funcs/verify_block.rs

@@ -24,6 +24,23 @@ impl Block {
         &self,
         db: &Db,
         verification_service: Arc<VerificationService>,
+    ) -> Result<Vec<String>, String> {
+        self.verify_internal(db, verification_service, false).await
+    }
+
+    pub async fn verify_historical_sync(
+        &self,
+        db: &Db,
+        verification_service: Arc<VerificationService>,
+    ) -> Result<Vec<String>, String> {
+        self.verify_internal(db, verification_service, true).await
+    }
+
+    async fn verify_internal(
+        &self,
+        db: &Db,
+        verification_service: Arc<VerificationService>,
+        historical_sync: bool,
     ) -> Result<Vec<String>, String> {
         // block verification checks header validity first, then
         // delegates transaction verification to the shared service
@@ -51,13 +68,19 @@ impl Block {
             .ok_or_else(|| "This miner address is not registered".to_string())?;
         let miner_pubkey_hex = encode(&miner_pubkey);
         let block_number = current_chain_height(db).await + 1;
+        let block_timestamp = header.unmined_block.timestamp;
 
-        if !NodeInfo::address_checkup(miner, block_number).await {
+        let miner_is_eligible = if historical_sync {
+            NodeInfo::historical_sync_address_checkup(miner, block_number, block_timestamp).await
+        } else {
+            NodeInfo::address_checkup(miner, block_number).await
+        };
+        if !miner_is_eligible {
             return Err("This address is not eligable to mine".to_string());
         }
 
         // get variables from the block
-        let timestamp = header.unmined_block.timestamp;
+        let timestamp = block_timestamp;
         let previous_hash = header.unmined_block.previous_hash.clone();
         let difficulty = header.unmined_block.next_block_difficulty;
         let vrf = header.vrf;