receive.rs (16623B)
1 use anyhow::Result; 2 3 use crate::domain::{ 4 Block, BurnBundle, ChainSnapshot, Ledger, Transaction, TransactionSubmitOutcome, TransactionV2, 5 ValidationError, decode_hex, error_has_validation, hex_encode, 6 }; 7 8 use super::{GossipEnvelope, IMPORT_REBROADCAST_LIMIT, NodeCore}; 9 10 impl NodeCore { 11 pub fn receive_transaction(&mut self, tx: Transaction) -> Result<TransactionSubmitOutcome> { 12 let outcome = self.ledger.submit_transaction_with_outcome(tx.clone())?; 13 Ok(outcome) 14 } 15 16 pub fn receive_gossiped_transaction(&mut self, tx: Transaction) -> Result<()> { 17 let outcome = match self.ledger.submit_transaction_with_outcome(tx.clone()) { 18 Ok(outcome) => outcome, 19 Err(error) 20 if matches!(&tx, Transaction::Mine { .. }) 21 && error_has_validation(&error, |kind| { 22 kind == ValidationError::MineAnchorLimitReached 23 }) => 24 { 25 return Ok(()); 26 } 27 Err(error) => return Err(error), 28 }; 29 if outcome.added() { 30 self.outbox.push(GossipEnvelope::Transaction(tx)); 31 } 32 Ok(()) 33 } 34 35 pub fn receive_gossiped_transaction_v2(&mut self, envelope: String) -> Result<()> { 36 let encoded = decode_hex(&envelope)?; 37 let transaction = self.ledger.decode_transaction_v2(&encoded)?; 38 let outcome = self.ledger.submit_transaction_v2(transaction)?; 39 if outcome.added() { 40 self.outbox.push(GossipEnvelope::TransactionV2 { 41 envelope: hex_encode(encoded), 42 }); 43 } 44 Ok(()) 45 } 46 47 pub(super) fn submit_public_transaction_v2( 48 &mut self, 49 transaction: TransactionV2, 50 ) -> Result<TransactionV2> { 51 let domain = self.ledger.transaction_v2_domain()?; 52 let envelope = hex_encode(transaction.encode(&domain)?); 53 if self 54 .ledger 55 .submit_transaction_v2(transaction.clone())? 56 .added() 57 { 58 self.outbox.push(GossipEnvelope::TransactionV2 { envelope }); 59 } 60 Ok(transaction) 61 } 62 63 pub(super) fn submit_prioritized_public_transaction_v2( 64 &mut self, 65 transaction: TransactionV2, 66 ) -> Result<TransactionV2> { 67 let domain = self.ledger.transaction_v2_domain()?; 68 let envelope = hex_encode(transaction.encode(&domain)?); 69 let mut candidate = self.ledger.clone(); 70 if candidate.prioritize_transaction_v2_for_block_building(transaction.clone())? { 71 self.ledger = candidate; 72 self.outbox.push(GossipEnvelope::TransactionV2 { envelope }); 73 } 74 Ok(transaction) 75 } 76 77 pub fn receive_burn_bundle(&mut self, bundle: BurnBundle) -> Result<()> { 78 let next_height = self.ledger.height().saturating_add(1); 79 if bundle.height <= self.ledger.height() { 80 return Ok(()); 81 } 82 if bundle.height > next_height { 83 return Ok(()); 84 } 85 let key = (bundle.height, bundle.slot, bundle.member.clone()); 86 self.ledger.precheck_next_block_burn_bundle(&bundle)?; 87 for burn in &bundle.burns { 88 self.receive_gossiped_transaction(burn.clone())?; 89 } 90 for envelope in &bundle.burns_v2 { 91 self.receive_gossiped_transaction_v2(envelope.clone())?; 92 } 93 self.ledger 94 .validate_next_block_burn_bundles(vec![bundle.clone()])?; 95 if self.equivocated_burn_bundle_slots.contains(&key) { 96 return Ok(()); 97 } 98 if let Some(existing) = self.burn_bundles.get(&key) { 99 if existing.canonical() != bundle.canonical() { 100 self.burn_bundles.remove(&key); 101 self.equivocated_burn_bundle_slots.insert(key); 102 } 103 return Ok(()); 104 } 105 self.burn_bundles.insert(key, bundle.clone()); 106 self.outbox.push(GossipEnvelope::BurnBundle(bundle)); 107 Ok(()) 108 } 109 110 pub fn receive(&mut self, envelope: GossipEnvelope) -> Result<()> { 111 match envelope { 112 GossipEnvelope::Hello(_) 113 | GossipEnvelope::PeerStatus { .. } 114 | GossipEnvelope::ChainBootstrapRequest 115 | GossipEnvelope::ChainBootstrap(_) 116 | GossipEnvelope::BlockLocatorRequest { .. } 117 | GossipEnvelope::BlockRangeRequest { .. } 118 | GossipEnvelope::BlockRequest { .. } 119 | GossipEnvelope::Inventory { .. } => Ok(()), 120 GossipEnvelope::Transaction(tx) => self.receive_gossiped_transaction(tx), 121 GossipEnvelope::Transactions { transactions } => { 122 for tx in transactions { 123 self.receive_gossiped_transaction(tx)?; 124 } 125 Ok(()) 126 } 127 GossipEnvelope::TransactionV2 { envelope } => { 128 self.receive_gossiped_transaction_v2(envelope) 129 } 130 GossipEnvelope::TransactionsV2 { envelopes } => { 131 for envelope in envelopes { 132 self.receive_gossiped_transaction_v2(envelope)?; 133 } 134 Ok(()) 135 } 136 GossipEnvelope::BurnBundle(bundle) => self.receive_burn_bundle(bundle), 137 GossipEnvelope::BurnBundles { bundles } => { 138 for bundle in bundles { 139 self.receive_burn_bundle(bundle)?; 140 } 141 Ok(()) 142 } 143 GossipEnvelope::BurnBundleRequest { .. } => Ok(()), 144 GossipEnvelope::Block(block) => { 145 let previous_height = self.ledger.height(); 146 self.ledger.apply_block(block.clone())?; 147 if self.ledger.height() > previous_height { 148 self.clear_stale_local_block_anchor(); 149 self.clear_stale_burn_bundle_collection(); 150 self.prune_burn_bundles(); 151 self.outbox.push(GossipEnvelope::Block(block)); 152 } 153 Ok(()) 154 } 155 GossipEnvelope::Blocks { blocks } => { 156 let mut imported = Vec::new(); 157 for block in blocks { 158 let previous_height = self.ledger.height(); 159 self.ledger.apply_block(block.clone())?; 160 if self.ledger.height() > previous_height { 161 self.clear_stale_local_block_anchor(); 162 self.clear_stale_burn_bundle_collection(); 163 self.prune_burn_bundles(); 164 imported.push(block); 165 } 166 } 167 for block in imported { 168 self.outbox.push(GossipEnvelope::Block(block)); 169 } 170 Ok(()) 171 } 172 GossipEnvelope::PeerAnnouncement { .. } 173 | GossipEnvelope::PeerVerificationChallenge { .. } 174 | GossipEnvelope::PeerVerificationResponse { .. } 175 | GossipEnvelope::PeerList { .. } => Ok(()), 176 } 177 } 178 179 pub(crate) fn receive_preverified_block_at(&mut self, block: Block, now_ms: u64) -> Result<()> { 180 let previous_height = self.ledger.height(); 181 self.ledger 182 .apply_preverified_block_at(block.clone(), now_ms)?; 183 if self.ledger.height() > previous_height { 184 self.clear_stale_local_block_anchor(); 185 self.clear_stale_burn_bundle_collection(); 186 self.prune_burn_bundles(); 187 self.outbox.push(GossipEnvelope::Block(block)); 188 } 189 Ok(()) 190 } 191 192 pub(crate) fn block_requires_vdf_verification_at( 193 &self, 194 block: &Block, 195 now_ms: u64, 196 ) -> Result<bool> { 197 self.ledger 198 .block_requires_vdf_verification_at(block, now_ms) 199 } 200 201 pub fn import_chain_snapshot(&mut self, snapshot: ChainSnapshot) -> Result<()> { 202 if self.network_migration_from().is_some() { 203 anyhow::bail!("local chain reset is required before joining the new network"); 204 } 205 let previous_height = self.ledger.height(); 206 let imported = self.ledger.extend_from_snapshot(snapshot)?; 207 if imported { 208 self.reset_automatic_mining_progress(); 209 self.clear_stale_local_block_anchor(); 210 self.clear_stale_burn_bundle_collection(); 211 self.prune_burn_bundles(); 212 self.enqueue_imported_blocks(previous_height)?; 213 } 214 Ok(()) 215 } 216 217 pub(crate) fn import_verified_ledger(&mut self, ledger: Ledger) -> Result<bool> { 218 if self.network_migration_from().is_some() { 219 anyhow::bail!("local chain reset is required before joining the new network"); 220 } 221 let replaces_setup_placeholder = self.ledger.is_setup_placeholder() 222 && ledger.genesis_hash() != self.ledger.genesis_hash(); 223 if ledger.genesis_hash() != self.ledger.genesis_hash() && !replaces_setup_placeholder { 224 anyhow::bail!("chain snapshot genesis does not match local chain"); 225 } 226 let previous_height = self.ledger.height(); 227 if !replaces_setup_placeholder { 228 if ledger.height() < previous_height { 229 return Ok(false); 230 } 231 if ledger.height() == previous_height && ledger.tip_hash() == self.ledger.tip_hash() { 232 return Ok(false); 233 } 234 } 235 236 self.ledger = ledger; 237 self.reset_automatic_mining_progress(); 238 self.clear_stale_local_block_anchor(); 239 self.clear_stale_burn_bundle_collection(); 240 self.prune_burn_bundles(); 241 self.enqueue_imported_blocks(previous_height)?; 242 Ok(true) 243 } 244 245 pub fn drain_outbox(&mut self) -> Vec<GossipEnvelope> { 246 std::mem::take(&mut self.outbox) 247 } 248 249 fn enqueue_imported_blocks(&mut self, previous_height: u64) -> Result<()> { 250 if self.ledger.height() <= previous_height { 251 return Ok(()); 252 } 253 let blocks = self 254 .ledger 255 .blocks_from(previous_height + 1, IMPORT_REBROADCAST_LIMIT); 256 for _ in &blocks { 257 self.prune_burn_bundles(); 258 self.clear_stale_burn_bundle_collection(); 259 } 260 if !blocks.is_empty() { 261 self.outbox.push(GossipEnvelope::Blocks { blocks }); 262 } 263 Ok(()) 264 } 265 } 266 267 #[cfg(test)] 268 mod tests { 269 use std::collections::BTreeMap; 270 271 use crate::{ 272 app::NodeCore, 273 domain::{GenesisBurn, Ledger, MICRO_IUNA, Wallet}, 274 }; 275 276 fn funded_ledger(wallets: &[Wallet]) -> Ledger { 277 let allocations = wallets 278 .iter() 279 .map(|wallet| (wallet.address().to_string(), 10 * MICRO_IUNA)) 280 .collect::<BTreeMap<_, _>>(); 281 let genesis_burns = wallets 282 .iter() 283 .map(|wallet| GenesisBurn::new(wallet.address(), MICRO_IUNA)) 284 .collect::<Vec<_>>(); 285 Ledger::new_with_genesis_burns(allocations, genesis_burns, 1).unwrap() 286 } 287 288 #[test] 289 fn verified_same_height_fork_replaces_local_ledger() { 290 let wallet = Wallet::from_seed("verified-same-height-fork"); 291 let parent = funded_ledger(std::slice::from_ref(&wallet)); 292 let mut local = parent.clone(); 293 let burn = local.build_burn(&wallet, 1, 1).unwrap(); 294 local.submit_transaction(burn.clone()).unwrap(); 295 let local_block = local.mine_next_block(&wallet, 1).unwrap(); 296 local.apply_locally_mined_block(local_block).unwrap(); 297 298 let mut remote = parent; 299 remote.submit_transaction(burn).unwrap(); 300 let remote_block = remote.mine_next_block(&wallet, 2).unwrap(); 301 remote 302 .apply_locally_mined_block(remote_block.clone()) 303 .unwrap(); 304 assert_eq!(local.height(), remote.height()); 305 assert_ne!(local.tip_hash(), remote.tip_hash()); 306 307 let mut node = NodeCore::from_ledger(wallet, local, 0); 308 assert!(node.import_verified_ledger(remote).unwrap()); 309 assert_eq!(node.ledger().tip_hash(), remote_block.hash); 310 } 311 312 #[test] 313 fn gossiped_mine_over_anchor_limit_is_silently_ignored() { 314 let wallets = (0..4) 315 .map(|index| { 316 let seed = format!("gossip-mine-limit-{index}"); 317 Wallet::from_seed(&seed) 318 }) 319 .collect::<Vec<_>>(); 320 let ledger = Ledger::new(BTreeMap::new(), 1); 321 let mine_actions = wallets 322 .iter() 323 .map(|wallet| ledger.build_mine(wallet.address()).unwrap()) 324 .collect::<Vec<_>>(); 325 let mut node = NodeCore::from_ledger(wallets[0].clone(), ledger, 0); 326 327 node.receive_gossiped_transaction(mine_actions[0].clone()) 328 .unwrap(); 329 node.receive_gossiped_transaction(mine_actions[1].clone()) 330 .unwrap(); 331 node.receive_gossiped_transaction(mine_actions[2].clone()) 332 .unwrap(); 333 334 assert_eq!(node.ledger().pending().len(), 2); 335 assert!( 336 node.ledger() 337 .pending() 338 .iter() 339 .all(|transaction| transaction.signature() != mine_actions[2].signature()) 340 ); 341 let error = node 342 .receive_transaction(mine_actions[3].clone()) 343 .unwrap_err(); 344 assert_eq!(error.to_string(), "mine transaction anchor limit reached"); 345 } 346 347 #[test] 348 fn burn_bundle_imports_new_signed_burns_to_mempool() { 349 let alice = Wallet::from_seed("bundle-import-new-burn-alice"); 350 let bob = Wallet::from_seed("bundle-import-new-burn-bob"); 351 let wallets = [alice.clone(), bob.clone()]; 352 let ledger = funded_ledger(&wallets); 353 let finalizer = ledger.expected_leader_for_next_block().unwrap(); 354 let signer = wallets 355 .iter() 356 .find(|wallet| wallet.address() == finalizer) 357 .expect("test ledger should include selected finalizer") 358 .clone(); 359 let burner = wallets 360 .iter() 361 .find(|wallet| wallet.address() != signer.address()) 362 .expect("test ledger should include a non-finalizer") 363 .clone(); 364 let burn = ledger.build_burn(&burner, 1, 1).unwrap(); 365 let mut signer_ledger = ledger.clone(); 366 signer_ledger.submit_transaction(burn.clone()).unwrap(); 367 let bundle = signer_ledger.build_burn_bundle(&signer).unwrap().unwrap(); 368 let mut receiver = NodeCore::from_ledger(signer, ledger, 0); 369 370 receiver.receive_burn_bundle(bundle).unwrap(); 371 372 assert!( 373 receiver 374 .ledger() 375 .pending() 376 .iter() 377 .any(|transaction| transaction.signature() == burn.signature()) 378 ); 379 } 380 381 #[test] 382 fn oversized_burn_bundle_does_not_import_embedded_burns() { 383 let alice = Wallet::from_seed("oversized-bundle-alice"); 384 let bob = Wallet::from_seed("oversized-bundle-bob"); 385 let wallets = [alice.clone(), bob.clone()]; 386 let ledger = funded_ledger(&wallets); 387 let finalizer = ledger.expected_leader_for_next_block().unwrap(); 388 let signer = wallets 389 .iter() 390 .find(|wallet| wallet.address() == finalizer) 391 .expect("test ledger should include selected finalizer") 392 .clone(); 393 let burner = wallets 394 .iter() 395 .find(|wallet| wallet.address() != signer.address()) 396 .expect("test ledger should include a non-finalizer") 397 .clone(); 398 let mut signer_ledger = ledger.clone(); 399 let mut burns = Vec::new(); 400 let oversized_bundle = loop { 401 let burn = signer_ledger.build_burn(&burner, 1, 1).unwrap(); 402 signer_ledger.submit_transaction(burn.clone()).unwrap(); 403 burns.push(burn); 404 let bundle = signer_ledger.test_burn_bundle(&signer, burns.clone()); 405 if bundle.serialized_size_bytes().unwrap() > 10_000 { 406 break bundle; 407 } 408 }; 409 let first_burn_signature = oversized_bundle.burns[0].signature().to_string(); 410 let mut receiver = NodeCore::from_ledger(signer, ledger, 0); 411 412 let error = receiver.receive_burn_bundle(oversized_bundle).unwrap_err(); 413 414 assert!( 415 error.to_string().contains("burn bundle exceeds max size"), 416 "{error:#}" 417 ); 418 assert!( 419 receiver 420 .ledger() 421 .pending() 422 .iter() 423 .all(|transaction| transaction.signature() != first_burn_signature) 424 ); 425 assert!(receiver.drain_outbox().is_empty()); 426 } 427 }