receive.rs (17953B)
1 use anyhow::{Result, bail}; 2 3 use crate::domain::{ 4 BlindedReveal, BlindedTransaction, Block, ChainSnapshot, Ledger, RevealBundle, Transaction, 5 TransactionSubmitOutcome, 6 }; 7 8 use super::{ 9 GossipEnvelope, IMPORT_REBROADCAST_LIMIT, NodeCore, helpers::transaction_input_outpoints, 10 }; 11 12 impl NodeCore { 13 pub fn receive_transaction(&mut self, tx: Transaction) -> Result<TransactionSubmitOutcome> { 14 let outcome = self.ledger.submit_transaction_with_outcome(tx.clone())?; 15 Ok(outcome) 16 } 17 18 pub fn receive_mine_action(&mut self, tx: Transaction) -> Result<()> { 19 if !matches!(tx, Transaction::Mine { .. }) { 20 bail!("only mine actions may be gossiped as plaintext"); 21 } 22 if self 23 .ledger 24 .submit_transaction_with_outcome(tx.clone())? 25 .added() 26 { 27 self.outbox.push(GossipEnvelope::MineAction(tx)); 28 } 29 Ok(()) 30 } 31 32 pub fn receive_blinded_transaction(&mut self, tx: BlindedTransaction) -> Result<()> { 33 if self.blinded_transaction_conflicts_with_local_anchor(&tx) { 34 return Ok(()); 35 } 36 if self.ledger.submit_blinded_transaction(tx.clone())? { 37 self.outbox.push(GossipEnvelope::BlindedTransaction(tx)); 38 } 39 Ok(()) 40 } 41 42 fn blinded_transaction_conflicts_with_local_anchor(&self, tx: &BlindedTransaction) -> bool { 43 let Some((height, burn)) = &self.local_block_anchor_burn else { 44 return false; 45 }; 46 if *height != self.ledger.height() || self.ledger.has_transaction(burn.signature()) { 47 return false; 48 } 49 let anchor_inputs = transaction_input_outpoints(burn); 50 tx.inputs 51 .iter() 52 .any(|input| anchor_inputs.contains(&input.outpoint)) 53 } 54 55 pub fn receive_blinded_reveal(&mut self, reveal: BlindedReveal) -> Result<()> { 56 self.receive_blinded_reveal_without_bundle_publish(reveal)?; 57 Ok(()) 58 } 59 60 fn receive_blinded_reveal_without_bundle_publish( 61 &mut self, 62 reveal: BlindedReveal, 63 ) -> Result<bool> { 64 if self.ledger.submit_blinded_reveal(reveal.clone())? { 65 self.outbox.push(GossipEnvelope::BlindedReveal(reveal)); 66 return Ok(true); 67 } 68 Ok(false) 69 } 70 71 pub fn receive_reveal_bundle(&mut self, bundle: RevealBundle) -> Result<()> { 72 let next_height = self.ledger.height().saturating_add(1); 73 if bundle.height <= self.ledger.height() { 74 return Ok(()); 75 } 76 if bundle.height > next_height { 77 return Ok(()); 78 } 79 let key = (bundle.height, bundle.slot); 80 if self.equivocated_reveal_bundle_slots.contains(&key) { 81 return Ok(()); 82 } 83 if let Some(existing) = self.reveal_bundles.get(&key) { 84 if existing.canonical() != bundle.canonical() { 85 self.reveal_bundles.remove(&key); 86 self.equivocated_reveal_bundle_slots.insert(key); 87 } 88 return Ok(()); 89 } 90 self.ledger 91 .validate_next_block_reveal_bundles(vec![bundle.clone()])?; 92 self.reveal_bundles.insert(key, bundle.clone()); 93 self.outbox.push(GossipEnvelope::RevealBundle(bundle)); 94 Ok(()) 95 } 96 97 pub fn receive(&mut self, envelope: GossipEnvelope) -> Result<()> { 98 match envelope { 99 GossipEnvelope::Hello(_) 100 | GossipEnvelope::PeerStatus { .. } 101 | GossipEnvelope::ChainSnapshotRequest 102 | GossipEnvelope::BlockRangeRequest { .. } 103 | GossipEnvelope::BlockRequest { .. } 104 | GossipEnvelope::Inventory { .. } => Ok(()), 105 GossipEnvelope::BlindedTransaction(tx) => self.receive_blinded_transaction(tx), 106 GossipEnvelope::BlindedTransactions { transactions } => { 107 for tx in transactions { 108 self.receive_blinded_transaction(tx)?; 109 } 110 Ok(()) 111 } 112 GossipEnvelope::MineAction(tx) => self.receive_mine_action(tx), 113 GossipEnvelope::MineActions { transactions } => { 114 for tx in transactions { 115 self.receive_mine_action(tx)?; 116 } 117 Ok(()) 118 } 119 GossipEnvelope::BlindedReveal(reveal) => self.receive_blinded_reveal(reveal), 120 GossipEnvelope::BlindedReveals { reveals } => { 121 for reveal in reveals { 122 self.receive_blinded_reveal_without_bundle_publish(reveal)?; 123 } 124 Ok(()) 125 } 126 GossipEnvelope::RevealBundle(bundle) => self.receive_reveal_bundle(bundle), 127 GossipEnvelope::RevealBundles { bundles } => { 128 for bundle in bundles { 129 self.receive_reveal_bundle(bundle)?; 130 } 131 Ok(()) 132 } 133 GossipEnvelope::Block(block) => { 134 let previous_height = self.ledger.height(); 135 self.ledger.apply_block(block.clone())?; 136 if self.ledger.height() > previous_height { 137 self.clear_stale_local_block_anchor(); 138 self.clear_stale_reveal_bundle_collection(); 139 self.prune_reveal_bundles(); 140 self.prune_owned_blinded_payloads_for_block(&block); 141 self.publish_owned_reveals_for_block(&block)?; 142 self.outbox.push(GossipEnvelope::Block(block)); 143 } 144 Ok(()) 145 } 146 GossipEnvelope::Blocks { blocks } => { 147 let mut imported = Vec::new(); 148 for block in blocks { 149 let previous_height = self.ledger.height(); 150 self.ledger.apply_block(block.clone())?; 151 if self.ledger.height() > previous_height { 152 self.clear_stale_local_block_anchor(); 153 self.clear_stale_reveal_bundle_collection(); 154 self.prune_reveal_bundles(); 155 self.prune_owned_blinded_payloads_for_block(&block); 156 self.publish_owned_reveals_for_block(&block)?; 157 imported.push(block); 158 } 159 } 160 for block in imported { 161 self.outbox.push(GossipEnvelope::Block(block)); 162 } 163 Ok(()) 164 } 165 GossipEnvelope::ChainSnapshot(snapshot) => self.import_chain_snapshot(snapshot), 166 GossipEnvelope::PeerAnnouncement { .. } 167 | GossipEnvelope::PeerVerificationChallenge { .. } 168 | GossipEnvelope::PeerVerificationResponse { .. } 169 | GossipEnvelope::PeerList { .. } => Ok(()), 170 } 171 } 172 173 pub(crate) fn receive_preverified_block_at(&mut self, block: Block, now_ms: u64) -> Result<()> { 174 let previous_height = self.ledger.height(); 175 self.ledger 176 .apply_preverified_block_at(block.clone(), now_ms)?; 177 if self.ledger.height() > previous_height { 178 self.clear_stale_local_block_anchor(); 179 self.clear_stale_reveal_bundle_collection(); 180 self.prune_reveal_bundles(); 181 self.prune_owned_blinded_payloads_for_block(&block); 182 self.publish_owned_reveals_for_block(&block)?; 183 self.outbox.push(GossipEnvelope::Block(block)); 184 } 185 Ok(()) 186 } 187 188 pub(crate) fn block_requires_vdf_verification_at( 189 &self, 190 block: &Block, 191 now_ms: u64, 192 ) -> Result<bool> { 193 self.ledger 194 .block_requires_vdf_verification_at(block, now_ms) 195 } 196 197 pub fn import_chain_snapshot(&mut self, snapshot: ChainSnapshot) -> Result<()> { 198 let previous_height = self.ledger.height(); 199 let imported = self.ledger.extend_from_snapshot(snapshot)?; 200 if imported { 201 self.reset_automatic_mining_progress(); 202 self.clear_stale_local_block_anchor(); 203 self.clear_stale_reveal_bundle_collection(); 204 self.prune_reveal_bundles(); 205 self.enqueue_imported_blocks(previous_height)?; 206 } 207 Ok(()) 208 } 209 210 pub(crate) fn import_verified_ledger(&mut self, ledger: Ledger) -> Result<bool> { 211 let replaces_setup_placeholder = self.ledger.is_setup_placeholder() 212 && ledger.genesis_hash() != self.ledger.genesis_hash(); 213 if ledger.genesis_hash() != self.ledger.genesis_hash() && !replaces_setup_placeholder { 214 anyhow::bail!("chain snapshot genesis does not match local chain"); 215 } 216 let previous_height = self.ledger.height(); 217 if !replaces_setup_placeholder && ledger.height() <= previous_height { 218 return Ok(false); 219 } 220 221 self.ledger = ledger; 222 self.reset_automatic_mining_progress(); 223 self.clear_stale_local_block_anchor(); 224 self.clear_stale_reveal_bundle_collection(); 225 self.prune_reveal_bundles(); 226 self.enqueue_imported_blocks(previous_height)?; 227 Ok(true) 228 } 229 230 pub fn drain_outbox(&mut self) -> Vec<GossipEnvelope> { 231 std::mem::take(&mut self.outbox) 232 } 233 234 fn enqueue_imported_blocks(&mut self, previous_height: u64) -> Result<()> { 235 if self.ledger.height() <= previous_height { 236 return Ok(()); 237 } 238 let blocks = self 239 .ledger 240 .blocks_from(previous_height + 1, IMPORT_REBROADCAST_LIMIT); 241 for block in &blocks { 242 self.prune_reveal_bundles(); 243 self.clear_stale_reveal_bundle_collection(); 244 self.prune_owned_blinded_payloads_for_block(block); 245 self.publish_owned_reveals_for_block(block)?; 246 } 247 if !blocks.is_empty() { 248 self.outbox.push(GossipEnvelope::Blocks { blocks }); 249 } 250 Ok(()) 251 } 252 } 253 254 #[cfg(test)] 255 mod tests { 256 use std::collections::BTreeMap; 257 258 use crate::{ 259 app::{ 260 GossipEnvelope, NodeCore, REVEAL_BUNDLE_COLLECTION_MS, 261 helpers::transaction_input_outpoints, 262 }, 263 domain::{GenesisBurn, Ledger, MICRO_IUNA, Wallet}, 264 }; 265 266 fn wallet_for_address<'a>(wallets: &'a [Wallet], address: &str) -> &'a Wallet { 267 wallets 268 .iter() 269 .find(|wallet| wallet.address() == address) 270 .unwrap_or_else(|| panic!("missing wallet for address {address}")) 271 } 272 273 #[test] 274 fn receiving_blinded_reveal_batch_waits_before_signing_committee_bundle() { 275 let alice = Wallet::from_seed("immediate-bundle-alice"); 276 let bob = Wallet::from_seed("immediate-bundle-bob"); 277 let carol = Wallet::from_seed("immediate-bundle-carol"); 278 let dave = Wallet::from_seed("immediate-bundle-dave"); 279 let finalizers = [alice.clone(), bob.clone()]; 280 let mut allocations = BTreeMap::new(); 281 allocations.insert(alice.address().to_string(), 10 * MICRO_IUNA); 282 allocations.insert(bob.address().to_string(), 10 * MICRO_IUNA); 283 allocations.insert(carol.address().to_string(), 10 * MICRO_IUNA); 284 allocations.insert(dave.address().to_string(), 10 * MICRO_IUNA); 285 let mut ledger = Ledger::new_with_genesis_burns( 286 allocations, 287 finalizers 288 .iter() 289 .map(|wallet| GenesisBurn::new(wallet.address(), MICRO_IUNA)) 290 .collect(), 291 1, 292 ) 293 .unwrap(); 294 let first = ledger 295 .build_blinded_burn(&carol, 3, 100, ledger.height() + 4) 296 .unwrap(); 297 let second = ledger 298 .build_blinded_burn(&dave, 4, 100, ledger.height() + 4) 299 .unwrap(); 300 ledger 301 .submit_blinded_transaction(first.transaction.clone()) 302 .unwrap(); 303 ledger 304 .submit_blinded_transaction(second.transaction.clone()) 305 .unwrap(); 306 let leader = ledger.expected_leader_for_next_block().unwrap(); 307 let leader_wallet = wallet_for_address(&finalizers, &leader); 308 let burn = ledger.build_burn(leader_wallet, 1, 0).unwrap(); 309 ledger.submit_transaction(burn).unwrap(); 310 let commit_block = ledger.mine_next_block(leader_wallet, 1).unwrap(); 311 ledger.apply_locally_mined_block(commit_block).unwrap(); 312 313 let committee = ledger.reveal_committee_for_next_block(); 314 let committee_wallet = committee 315 .iter() 316 .filter_map(|member| { 317 finalizers 318 .iter() 319 .find(|wallet| wallet.address() == member.owner) 320 }) 321 .next() 322 .expect("test finalizer should be in reveal committee"); 323 let mut committee_node = NodeCore::from_ledger_with_burn_fee_and_enabled( 324 committee_wallet.clone(), 325 ledger, 326 true, 327 0, 328 0, 329 ); 330 331 committee_node 332 .receive(GossipEnvelope::BlindedReveals { 333 reveals: vec![first.reveal.clone()], 334 }) 335 .unwrap(); 336 let outbox = committee_node.drain_outbox(); 337 338 assert!(outbox.iter().any(|envelope| matches!( 339 envelope, 340 GossipEnvelope::BlindedReveal(reveal) if reveal.commitment == first.reveal.commitment 341 ))); 342 assert!( 343 !outbox 344 .iter() 345 .any(|envelope| matches!(envelope, GossipEnvelope::RevealBundle(_))) 346 ); 347 348 let early = committee_node.prepare_automatic_finalization(2); 349 assert!(early.work.is_none()); 350 assert!( 351 early 352 .skipped_reason 353 .as_deref() 354 .unwrap_or_default() 355 .contains("collecting blinded reveals") 356 ); 357 assert!( 358 !committee_node 359 .drain_outbox() 360 .iter() 361 .any(|envelope| matches!(envelope, GossipEnvelope::RevealBundle(_))) 362 ); 363 364 committee_node 365 .receive(GossipEnvelope::BlindedReveals { 366 reveals: vec![second.reveal.clone()], 367 }) 368 .unwrap(); 369 let outbox = committee_node.drain_outbox(); 370 assert!(outbox.iter().any(|envelope| matches!( 371 envelope, 372 GossipEnvelope::BlindedReveal(reveal) if reveal.commitment == second.reveal.commitment 373 ))); 374 assert!( 375 !outbox 376 .iter() 377 .any(|envelope| matches!(envelope, GossipEnvelope::RevealBundle(_))) 378 ); 379 380 let ready = committee_node.prepare_automatic_finalization(REVEAL_BUNDLE_COLLECTION_MS + 3); 381 let _ = ready; 382 let outbox = committee_node.drain_outbox(); 383 assert!(outbox.iter().any(|envelope| matches!( 384 envelope, 385 GossipEnvelope::RevealBundle(bundle) 386 if bundle.member == committee_wallet.address() 387 && bundle.reveals.len() == 2 388 && bundle.reveals.iter().any(|reveal| reveal.commitment == first.reveal.commitment) 389 && bundle.reveals.iter().any(|reveal| reveal.commitment == second.reveal.commitment) 390 ))); 391 } 392 393 #[test] 394 fn inbound_blinded_transaction_conflicting_with_local_anchor_is_not_queued() { 395 let alice = Wallet::from_seed("local-anchor-inbound-alice"); 396 let bob = Wallet::from_seed("local-anchor-inbound-bob"); 397 let finalizers = [alice.clone(), bob.clone()]; 398 let mut allocations = BTreeMap::new(); 399 allocations.insert(alice.address().to_string(), 10 * MICRO_IUNA); 400 allocations.insert(bob.address().to_string(), 10 * MICRO_IUNA); 401 let ledger = Ledger::new_with_genesis_burns( 402 allocations, 403 finalizers 404 .iter() 405 .map(|wallet| GenesisBurn::new(wallet.address(), MICRO_IUNA)) 406 .collect(), 407 1, 408 ) 409 .unwrap(); 410 let leader = ledger.expected_leader_for_next_block().unwrap(); 411 let leader_wallet = finalizers 412 .iter() 413 .find(|wallet| wallet.address() == leader) 414 .unwrap() 415 .clone(); 416 let mut node = NodeCore::from_ledger_with_burn_fee_and_enabled( 417 leader_wallet.clone(), 418 ledger, 419 true, 420 MICRO_IUNA / 10, 421 1, 422 ); 423 424 let plan = node.prepare_automatic_finalization(1); 425 assert!(plan.burned.is_some()); 426 assert_eq!(node.ledger().pending_blinded_transactions().len(), 1); 427 let automatic_burn_commitment = node.ledger().pending_blinded_transactions()[0] 428 .commitment 429 .clone(); 430 assert_eq!( 431 node.ledger().pending_blinded_transactions()[0].commitment, 432 automatic_burn_commitment 433 ); 434 node.drain_outbox(); 435 let (_, anchor_burn) = node 436 .local_block_anchor_burn 437 .clone() 438 .expect("leader burn should be held as a local block anchor"); 439 let anchor_inputs = transaction_input_outpoints(&anchor_burn) 440 .into_iter() 441 .collect::<Vec<_>>(); 442 let conflicting_payload = node 443 .ledger() 444 .build_transfer_with_inputs(&leader_wallet, bob.address(), 1, 0, &anchor_inputs) 445 .unwrap(); 446 let conflicting = node 447 .ledger() 448 .build_blinded_transaction(&leader_wallet, conflicting_payload, node.chain_height() + 4) 449 .unwrap(); 450 451 node.receive_blinded_transaction(conflicting.transaction) 452 .unwrap(); 453 454 assert_eq!(node.ledger().pending_blinded_transactions().len(), 1); 455 assert_eq!( 456 node.ledger().pending_blinded_transactions()[0].commitment, 457 automatic_burn_commitment 458 ); 459 assert!(node.drain_outbox().is_empty()); 460 assert!(node.prepare_automatic_finalization(1).work.is_some()); 461 } 462 }