fetch.rs (8411B)
1 use std::net::SocketAddr; 2 3 use anyhow::{Context, Result}; 4 use tokio::{ 5 io::AsyncWriteExt, 6 net::{TcpStream, tcp::OwnedReadHalf}, 7 time::timeout, 8 }; 9 10 use crate::{ 11 app::{GossipEnvelope, NETWORK_ID, PROTOCOL_VERSION, now_ms}, 12 domain::{Block, ChainSnapshot, Ledger, verify_vdf}, 13 }; 14 15 use super::{ 16 GossipNetwork, JOIN_RESPONSE_TIMEOUT, LimitedLineReader, MAX_JOIN_RESPONSE_ENVELOPES, 17 PeerStatus, parse_envelope, 18 }; 19 20 pub async fn fetch_snapshot(peer: &str) -> Result<ChainSnapshot> { 21 fetch_snapshot_with_announcement(peer, None).await 22 } 23 24 pub async fn fetch_peer_height(peer: &str) -> Result<u64> { 25 fetch_peer_status(peer).await.map(|status| status.height) 26 } 27 28 async fn fetch_peer_status(peer: &str) -> Result<PeerStatus> { 29 let stream = TcpStream::connect(peer) 30 .await 31 .with_context(|| format!("connecting to peer {peer}"))?; 32 let (reader, _writer) = stream.into_split(); 33 let mut reader = LimitedLineReader::new(reader); 34 let line = reader 35 .read_line() 36 .await? 37 .with_context(|| format!("peer {peer} closed before sending its peer status"))?; 38 match parse_envelope(&line)? { 39 GossipEnvelope::Hello(hello) => { 40 if hello.protocol_version != PROTOCOL_VERSION { 41 anyhow::bail!( 42 "unsupported protocol version {}; expected {}", 43 hello.protocol_version, 44 PROTOCOL_VERSION 45 ); 46 } 47 if hello.network_id != NETWORK_ID { 48 anyhow::bail!( 49 "wrong network {}; expected {}", 50 hello.network_id, 51 NETWORK_ID 52 ); 53 } 54 Ok(PeerStatus::with_time( 55 hello.height, 56 hello.tip_hash, 57 hello.time_ms, 58 )) 59 } 60 GossipEnvelope::PeerStatus { 61 height, 62 tip_hash, 63 time_ms, 64 } => Ok(PeerStatus::from_envelope(height, tip_hash, time_ms)), 65 other => anyhow::bail!("peer {peer} sent {other:?} instead of peer status"), 66 } 67 } 68 69 pub async fn fetch_snapshot_with_announcement( 70 peer: &str, 71 _advertised_addr: Option<SocketAddr>, 72 ) -> Result<ChainSnapshot> { 73 let stream = TcpStream::connect(peer) 74 .await 75 .with_context(|| format!("connecting to join peer {peer}"))?; 76 let (reader, mut writer) = stream.into_split(); 77 let mut reader = LimitedLineReader::new(reader); 78 let line = reader 79 .read_line() 80 .await? 81 .with_context(|| format!("join peer {peer} closed before sending its peer status"))?; 82 match parse_envelope(&line)? { 83 GossipEnvelope::Hello(hello) => { 84 if hello.protocol_version != PROTOCOL_VERSION { 85 anyhow::bail!( 86 "unsupported protocol version {}; expected {}", 87 hello.protocol_version, 88 PROTOCOL_VERSION 89 ); 90 } 91 if hello.network_id != NETWORK_ID { 92 anyhow::bail!( 93 "wrong network {}; expected {}", 94 hello.network_id, 95 NETWORK_ID 96 ); 97 } 98 } 99 GossipEnvelope::PeerStatus { .. } => {} 100 other => anyhow::bail!("join peer {peer} sent {other:?} instead of peer status"), 101 } 102 103 let line = serde_json::to_string(&GossipEnvelope::ChainSnapshotRequest)?; 104 writer.write_all(line.as_bytes()).await?; 105 writer.write_all(b"\n").await?; 106 let snapshot = read_join_snapshot_response(peer, &mut reader).await?; 107 108 Ok(snapshot) 109 } 110 111 async fn read_join_snapshot_response( 112 peer: &str, 113 reader: &mut LimitedLineReader<OwnedReadHalf>, 114 ) -> Result<ChainSnapshot> { 115 for _ in 0..MAX_JOIN_RESPONSE_ENVELOPES { 116 let line = timeout(JOIN_RESPONSE_TIMEOUT, reader.read_line()) 117 .await 118 .with_context(|| format!("join peer {peer} timed out waiting for a chain snapshot"))?? 119 .with_context(|| format!("join peer {peer} closed before sending a chain snapshot"))?; 120 match join_snapshot_response(peer, parse_envelope(&line)?)? { 121 Some(snapshot) => return Ok(snapshot), 122 None => continue, 123 } 124 } 125 126 anyhow::bail!("join peer {peer} sent too many non-snapshot envelopes while joining") 127 } 128 129 pub(super) fn join_snapshot_response( 130 peer: &str, 131 envelope: GossipEnvelope, 132 ) -> Result<Option<ChainSnapshot>> { 133 match envelope { 134 GossipEnvelope::ChainSnapshot(snapshot) => Ok(Some(snapshot)), 135 GossipEnvelope::Hello(_) 136 | GossipEnvelope::PeerStatus { .. } 137 | GossipEnvelope::PeerList { .. } 138 | GossipEnvelope::PeerVerificationChallenge { .. } 139 | GossipEnvelope::PeerVerificationResponse { .. } 140 | GossipEnvelope::Inventory { .. } => Ok(None), 141 other => anyhow::bail!("join peer {peer} sent {other:?} instead of a chain snapshot"), 142 } 143 } 144 145 pub(super) async fn validate_snapshot_extension( 146 mut ledger: Ledger, 147 snapshot: ChainSnapshot, 148 now_ms: u64, 149 ) -> Result<Ledger> { 150 if ledger.is_setup_placeholder() { 151 return tokio::task::spawn_blocking(move || Ledger::from_snapshot_at(snapshot, now_ms)) 152 .await 153 .context("chain snapshot adoption worker failed")?; 154 } 155 let missing_blocks = ledger.missing_snapshot_blocks(&snapshot)?; 156 verify_blocks_vdf(missing_blocks).await?; 157 158 tokio::task::spawn_blocking(move || { 159 ledger.extend_from_preverified_snapshot_at(snapshot, now_ms)?; 160 Ok(ledger) 161 }) 162 .await 163 .context("chain snapshot extension worker failed")? 164 } 165 166 pub(super) async fn validate_blocks_extension( 167 mut ledger: Ledger, 168 blocks: Vec<Block>, 169 now_ms: u64, 170 ) -> Result<Ledger> { 171 if blocks.is_empty() { 172 return Ok(ledger); 173 } 174 verify_blocks_vdf(blocks.clone()).await?; 175 176 tokio::task::spawn_blocking(move || { 177 for block in blocks { 178 ledger.apply_preverified_block_at(block, now_ms)?; 179 } 180 Ok(ledger) 181 }) 182 .await 183 .context("block batch extension worker failed")? 184 } 185 186 pub(super) async fn network_adjusted_time_ms(network: &GossipNetwork) -> u64 { 187 let local_time_ms = now_ms(); 188 network 189 .inner 190 .peers 191 .lock() 192 .await 193 .adjusted_time_ms_at(local_time_ms) 194 } 195 196 pub(super) async fn verify_block_vdf(block: Block) -> Result<Block> { 197 let seed = block.vdf_seed(); 198 let rounds = block.vdf_rounds; 199 let solution = block.vdf_output.clone(); 200 let valid = tokio::task::spawn_blocking(move || verify_vdf(&seed, rounds, &solution)) 201 .await 202 .context("VDF verification worker failed")?; 203 if !valid { 204 anyhow::bail!("block VDF output is invalid"); 205 } 206 207 Ok(block) 208 } 209 210 async fn verify_blocks_vdf(blocks: Vec<Block>) -> Result<()> { 211 let mut tasks = tokio::task::JoinSet::new(); 212 for block in blocks { 213 tasks.spawn_blocking(move || { 214 if !verify_vdf(&block.vdf_seed(), block.vdf_rounds, &block.vdf_output) { 215 anyhow::bail!("block {} VDF output is invalid", block.height); 216 } 217 Ok::<(), anyhow::Error>(()) 218 }); 219 } 220 221 while let Some(result) = tasks.join_next().await { 222 result.context("VDF verification worker failed")??; 223 } 224 225 Ok(()) 226 } 227 228 #[cfg(test)] 229 mod tests { 230 use crate::{app::GossipEnvelope, domain::Wallet}; 231 232 use super::super::test_support::{allocations, node}; 233 use super::join_snapshot_response; 234 235 #[test] 236 fn join_snapshot_response_ignores_status_noise_before_snapshot() { 237 assert!( 238 join_snapshot_response( 239 "127.0.0.1:9544", 240 GossipEnvelope::PeerStatus { 241 height: 0, 242 tip_hash: "tip".to_string(), 243 time_ms: 1_000, 244 } 245 ) 246 .unwrap() 247 .is_none() 248 ); 249 250 let alice = Wallet::from_seed("join-noise-alice"); 251 let snapshot = node( 252 "alice", 253 alice.clone(), 254 allocations(std::slice::from_ref(&alice), 1_000), 255 ) 256 .chain_snapshot(); 257 let parsed = join_snapshot_response( 258 "127.0.0.1:9544", 259 GossipEnvelope::ChainSnapshot(snapshot.clone()), 260 ) 261 .unwrap(); 262 263 assert_eq!(parsed, Some(snapshot)); 264 } 265 }