gossip.rs (8655B)
1 use crate::domain::{Block, ChainSnapshot, Transaction}; 2 3 use super::{ 4 BLOCK_REQUEST_LIMIT, GossipEnvelope, NETWORK_ID, NodeCore, PROTOCOL_VERSION, ProtocolHello, 5 TRANSACTION_BATCH_LIMIT, now_ms, types::BlockInventory, 6 }; 7 8 impl NodeCore { 9 pub fn mempool_gossip(&mut self) -> Vec<GossipEnvelope> { 10 let mut gossip = Vec::new(); 11 let mine_actions = self 12 .ledger 13 .pending() 14 .iter() 15 .filter(|transaction| matches!(transaction, Transaction::Mine { .. })) 16 .cloned() 17 .collect::<Vec<_>>(); 18 gossip.extend(mine_actions.chunks(TRANSACTION_BATCH_LIMIT).map(|chunk| { 19 GossipEnvelope::MineActions { 20 transactions: chunk.to_vec(), 21 } 22 })); 23 gossip.extend( 24 self.ledger 25 .pending_blinded_transactions() 26 .chunks(TRANSACTION_BATCH_LIMIT) 27 .map(|chunk| GossipEnvelope::BlindedTransactions { 28 transactions: chunk.to_vec(), 29 }), 30 ); 31 gossip.extend( 32 self.ledger 33 .pending_blinded_reveals() 34 .chunks(TRANSACTION_BATCH_LIMIT) 35 .map(|chunk| GossipEnvelope::BlindedReveals { 36 reveals: chunk.to_vec(), 37 }), 38 ); 39 gossip.extend( 40 self.usable_reveal_bundles() 41 .chunks(TRANSACTION_BATCH_LIMIT) 42 .map(|chunk| GossipEnvelope::RevealBundles { 43 bundles: chunk.to_vec(), 44 }), 45 ); 46 gossip 47 } 48 49 pub fn chain_snapshot(&self) -> ChainSnapshot { 50 self.ledger.snapshot() 51 } 52 53 pub fn hello(&self, listen_addr: Option<String>, node_id: Option<String>) -> GossipEnvelope { 54 GossipEnvelope::Hello(ProtocolHello { 55 protocol_version: PROTOCOL_VERSION, 56 network_id: NETWORK_ID.to_string(), 57 genesis_hash: self.ledger.genesis_hash().to_string(), 58 listen_addr, 59 node_id, 60 height: self.ledger.height(), 61 tip_hash: self.ledger.tip_hash().to_string(), 62 time_ms: now_ms(), 63 }) 64 } 65 66 pub fn peer_status(&self) -> GossipEnvelope { 67 GossipEnvelope::PeerStatus { 68 height: self.ledger.height(), 69 tip_hash: self.ledger.tip_hash().to_string(), 70 time_ms: now_ms(), 71 } 72 } 73 74 pub fn blocks_from(&self, from_height: u64, limit: usize) -> Vec<Block> { 75 self.ledger.blocks_from(from_height, limit) 76 } 77 78 pub fn blocks_by_hash(&self, hashes: &[String]) -> Vec<Block> { 79 hashes 80 .iter() 81 .filter_map(|hash| self.ledger.block_by_hash(hash)) 82 .collect() 83 } 84 85 pub fn missing_inventory_requests(&self, blocks: &[BlockInventory]) -> Vec<GossipEnvelope> { 86 let local_height = self.ledger.height(); 87 let first_height_gap = blocks 88 .iter() 89 .filter(|block| !self.ledger.has_block(&block.hash)) 90 .filter(|block| block.height > local_height + 1) 91 .map(|block| block.height) 92 .min(); 93 let missing_blocks = blocks 94 .iter() 95 .filter(|block| !self.ledger.has_block(&block.hash)) 96 .filter(|block| first_height_gap.is_none_or(|gap| block.height < gap)) 97 .map(|block| block.hash.clone()) 98 .collect::<Vec<_>>(); 99 100 let mut requests = Vec::new(); 101 if !missing_blocks.is_empty() { 102 requests.push(GossipEnvelope::BlockRequest { 103 hashes: missing_blocks, 104 }); 105 } 106 if first_height_gap.is_some() { 107 requests.push(GossipEnvelope::BlockRangeRequest { 108 from_height: local_height + 1, 109 limit: BLOCK_REQUEST_LIMIT, 110 }); 111 } 112 requests 113 } 114 } 115 116 #[cfg(test)] 117 mod tests { 118 use std::collections::BTreeMap; 119 120 use crate::{ 121 app::{BLOCK_REQUEST_LIMIT, BlockInventory, GossipEnvelope, NodeCore}, 122 domain::{Amount, GenesisBurn, Ledger, MICRO_IUNA, Transaction, Wallet}, 123 }; 124 125 #[test] 126 fn mempool_gossip_includes_blinded_transactions() { 127 let alice = Wallet::from_seed("blinded-gossip-alice"); 128 let mut genesis = BTreeMap::new(); 129 genesis.insert(alice.address().to_string(), 10 * MICRO_IUNA); 130 let ledger = Ledger::new(genesis, 1); 131 let blinded = ledger.build_blinded_burn(&alice, MICRO_IUNA, 7, 3).unwrap(); 132 let mut sender = NodeCore::from_ledger(alice.clone(), ledger.clone(), 0); 133 let mut receiver = NodeCore::from_ledger(alice, ledger, 0); 134 135 sender 136 .receive_blinded_transaction(blinded.transaction.clone()) 137 .unwrap(); 138 for envelope in sender.mempool_gossip() { 139 receiver.receive(envelope).unwrap(); 140 } 141 142 assert_eq!( 143 receiver.ledger().pending_blinded_transactions(), 144 std::slice::from_ref(&blinded.transaction) 145 ); 146 } 147 148 #[test] 149 fn mempool_gossip_includes_public_mine_actions() { 150 let alice = Wallet::from_seed("mine-gossip-alice"); 151 let ledger = Ledger::new(BTreeMap::new(), 1); 152 let mine = ledger.build_mine(alice.address()).unwrap(); 153 let mut sender = NodeCore::from_ledger(alice.clone(), ledger.clone(), 0); 154 let mut receiver = NodeCore::from_ledger(alice, ledger, 0); 155 156 sender.submit_public_mine_action(mine.clone()).unwrap(); 157 for envelope in sender.mempool_gossip() { 158 receiver.receive(envelope).unwrap(); 159 } 160 161 assert_eq!(receiver.ledger().pending(), std::slice::from_ref(&mine)); 162 assert!(receiver.ledger().pending_blinded_transactions().is_empty()); 163 } 164 165 #[test] 166 fn inventory_requests_only_missing_objects() { 167 let alice = Wallet::from_seed("missing-inv-alice"); 168 let bob = Wallet::from_seed("missing-inv-bob"); 169 let allocations = allocations(&[alice.clone(), bob], 1_000); 170 let mut local = node("local", alice.clone(), allocations.clone()); 171 let mut remote = node("remote", alice.clone(), allocations); 172 queue_plaintext_burn(&mut local, &alice, 1); 173 let block = local.mine_one_at(1).unwrap(); 174 let inventory = [BlockInventory { 175 height: block.height, 176 hash: block.hash.clone(), 177 }]; 178 179 let requests = remote.missing_inventory_requests(&inventory); 180 assert_eq!(requests.len(), 1); 181 assert!(matches!(requests[0], GossipEnvelope::BlockRequest { .. })); 182 183 remote.receive(GossipEnvelope::Block(block)).unwrap(); 184 assert!(remote.missing_inventory_requests(&inventory).is_empty()); 185 } 186 187 #[test] 188 fn inventory_gap_requests_range_instead_of_orphan_block() { 189 let alice = Wallet::from_seed("gap-inv-alice"); 190 let bob = Wallet::from_seed("gap-inv-bob"); 191 let allocations = allocations(&[alice.clone(), bob.clone()], 1_000); 192 let mut local = node("local", alice.clone(), allocations.clone()); 193 let remote = node("remote", bob, allocations); 194 195 let mut latest = None; 196 for height in 1..=3 { 197 queue_plaintext_burn(&mut local, &alice, 1); 198 latest = Some(local.mine_one_at(height).unwrap()); 199 } 200 let latest = latest.unwrap(); 201 202 let requests = remote.missing_inventory_requests(&[BlockInventory { 203 height: latest.height, 204 hash: latest.hash, 205 }]); 206 207 assert_eq!(requests.len(), 1); 208 match &requests[0] { 209 GossipEnvelope::BlockRangeRequest { from_height, limit } => { 210 assert_eq!(*from_height, 1); 211 assert_eq!(*limit, BLOCK_REQUEST_LIMIT); 212 } 213 other => panic!("expected block range request, got {other:?}"), 214 } 215 } 216 217 fn node(_network_key: &str, wallet: Wallet, allocations: BTreeMap<String, Amount>) -> NodeCore { 218 let ledger = Ledger::new_with_genesis_burns( 219 allocations, 220 vec![GenesisBurn::new(wallet.address(), 1)], 221 25, 222 ) 223 .unwrap(); 224 NodeCore::from_ledger(wallet, ledger, 0) 225 } 226 227 fn queue_plaintext_burn(node: &mut NodeCore, wallet: &Wallet, amount: Amount) -> Transaction { 228 let tx = node.ledger().build_burn(wallet, amount, 0).unwrap(); 229 node.receive_transaction(tx.clone()).unwrap(); 230 tx 231 } 232 233 fn allocations(wallets: &[Wallet], amount: Amount) -> BTreeMap<String, Amount> { 234 wallets 235 .iter() 236 .map(|wallet| (wallet.address().to_string(), amount)) 237 .collect() 238 } 239 }