diff --git a/src/orphans/orphan_checkup.rs b/src/orphans/orphan_checkup.rs index 4752b1f..a1d0f3a 100644 --- a/src/orphans/orphan_checkup.rs +++ b/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), diff --git a/src/orphans/save_blocks.rs b/src/orphans/save_blocks.rs index eaed5cf..fdd197f 100644 --- a/src/orphans/save_blocks.rs +++ b/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(¶ms.db); - match handle_response_and_save_torrent( + match handle_historical_response_and_save_torrent( true_start_height, ¶ms.db, torrent.clone(), @@ -186,7 +186,7 @@ pub async fn save_new_blocks( let local_height_before = get_height(¶ms.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, ¶ms.db, torrent, diff --git a/src/orphans/sync_check.rs b/src/orphans/sync_check.rs index 0c30893..7f881e3 100644 --- a/src/orphans/sync_check.rs +++ b/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, ¶ms.db, torrent, diff --git a/src/records/memory/network_mapping/queries.rs b/src/records/memory/network_mapping/queries.rs index 0199473..48de7d6 100644 --- a/src/records/memory/network_mapping/queries.rs +++ b/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 - || Self::can_pending_member_add_monitor(address, node) + node.deleted_timestamp == 0 || 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 { 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"; diff --git a/src/rpc/commands/receive_torrent.rs b/src/rpc/commands/receive_torrent.rs index fa48fd5..3b41557 100644 --- a/src/rpc/commands/receive_torrent.rs +++ b/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, ) diff --git a/src/torrent/structs.rs b/src/torrent/structs.rs index 886dfb4..d45d1ff 100644 --- a/src/torrent/structs.rs +++ b/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, diff --git a/src/torrent/torrenting_system/save_block.rs b/src/torrent/torrenting_system/save_block.rs index 532050b..6724614 100644 --- a/src/torrent/torrenting_system/save_block.rs +++ b/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 - .verify(¶ms.db, params.verification_service.clone()) - .await - { + let verification = if params.historical_sync { + loaded_block + .verify_historical_sync(¶ms.db, params.verification_service.clone()) + .await + } else { + loaded_block + .verify(¶ms.db, params.verification_service.clone()) + .await + }; + let signatures = match verification { Ok(signatures) => signatures, Err(err) => { error!( diff --git a/src/torrent/torrenting_system/setup_block_download.rs b/src/torrent/torrenting_system/setup_block_download.rs index 59773e1..76ec31e 100644 --- a/src/torrent/torrenting_system/setup_block_download.rs +++ b/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, @@ -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, diff --git a/src/torrent/torrenting_system/torrent_requests.rs b/src/torrent/torrenting_system/torrent_requests.rs index 9d64eb9..f2230e7 100644 --- a/src/torrent/torrenting_system/torrent_requests.rs +++ b/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, + map: Arc>, + 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, + map: Arc>, + 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>, @@ -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), diff --git a/src/verifications/async_funcs/verify_block.rs b/src/verifications/async_funcs/verify_block.rs index d363ccc..d5c3db7 100644 --- a/src/verifications/async_funcs/verify_block.rs +++ b/src/verifications/async_funcs/verify_block.rs @@ -24,6 +24,23 @@ impl Block { &self, db: &Db, verification_service: Arc, + ) -> Result, String> { + self.verify_internal(db, verification_service, false).await + } + + pub async fn verify_historical_sync( + &self, + db: &Db, + verification_service: Arc, + ) -> Result, String> { + self.verify_internal(db, verification_service, true).await + } + + async fn verify_internal( + &self, + db: &Db, + verification_service: Arc, + historical_sync: bool, ) -> Result, 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;