1use crate::config::{
2 P2P_BLOCK_PEER_TIMEOUT_SECS, P2P_CONNECT_TIMEOUT_SECS, P2P_HANDSHAKE_TIMEOUT_SECS,
3 P2P_HEADERS_TIMEOUT_SECS, P2P_HEIGHT_SLACK, P2P_PIPELINE_PER_PEER, P2P_POLL_MESSAGES,
4 P2P_POLL_READ_TIMEOUT_MILLIS, P2P_PROTOCOL_VERSION, P2P_READ_TIMEOUT_SECS, P2P_USER_AGENT,
5};
6use crate::error::{Context, Error, ensure};
7use crate::p2p::wire::{self, Message};
8use bitcoin::BlockHash;
9use bitcoin::block::Header;
10use bitcoin::consensus::encode::deserialize;
11use bitcoin::hashes::Hash;
12use bitcoin::p2p::ServiceFlags;
13use bitcoin::p2p::address::Address;
14use bitcoin::p2p::message::NetworkMessage;
15use bitcoin::p2p::message_blockdata::{GetHeadersMessage, Inventory};
16use bitcoin::p2p::message_network::VersionMessage;
17use std::io::{self, ErrorKind};
18use std::net::{SocketAddr, TcpStream};
19use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
20use urma_chain::observation::Chain;
21
22pub struct Peer {
23 stream: TcpStream,
24 magic: [u8; 4],
25 pub address: SocketAddr,
26 pub services: ServiceFlags,
27 pub start_height: u64,
28 pub learned: Vec<Gossiped>,
29 pub version: Option<VersionMessage>,
30 pub(super) proven_height: Option<u64>,
31 pub(super) arrived: Vec<Vec<u8>>,
32 pub(super) lacking: Vec<BlockHash>,
33 requested: Vec<BlockHash>,
34}
35
36#[derive(Clone, Copy, Debug, PartialEq, Eq)]
37pub struct Gossiped {
38 pub address: SocketAddr,
39 pub time: u32,
40}
41
42enum Waited<T> {
43 Done(T),
44 Keep,
45}
46
47#[derive(Clone, Debug)]
48pub struct Hello {
49 pub services: ServiceFlags,
50 pub user_agent: String,
51 pub relay: bool,
52 pub negotiate: bool,
53}
54
55impl Hello {
56 pub fn light() -> Self {
57 Self {
58 services: ServiceFlags::NONE,
59 user_agent: P2P_USER_AGENT.to_owned(),
60 relay: false,
61 negotiate: false,
62 }
63 }
64}
65
66impl Peer {
67 pub fn connect(chain: Chain, address: SocketAddr, our_height: u64) -> Result<Self, Error> {
68 Self::connect_with(chain, address, our_height, &Hello::light())
69 }
70
71 pub fn connect_with(
72 chain: Chain,
73 address: SocketAddr,
74 our_height: u64,
75 hello: &Hello,
76 ) -> Result<Self, Error> {
77 let stream =
78 TcpStream::connect_timeout(&address, Duration::from_secs(P2P_CONNECT_TIMEOUT_SECS))?;
79 stream.set_read_timeout(Some(Duration::from_secs(P2P_READ_TIMEOUT_SECS)))?;
80 stream.set_write_timeout(Some(Duration::from_secs(P2P_READ_TIMEOUT_SECS)))?;
81 stream.set_nodelay(true)?;
82 let mut peer = Self {
83 stream,
84 magic: chain.params().magic,
85 address,
86 services: ServiceFlags::NONE,
87 start_height: 0,
88 learned: Vec::new(),
89 version: None,
90 proven_height: None,
91 arrived: Vec::new(),
92 lacking: Vec::new(),
93 requested: Vec::new(),
94 };
95 peer.handshake(our_height, hello)?;
96 Ok(peer)
97 }
98
99 fn handshake(&mut self, our_height: u64, hello: &Hello) -> Result<(), Error> {
100 let elapsed = SystemTime::now()
101 .duration_since(UNIX_EPOCH)
102 .map_err(|error| {
103 Error::Invalid(format!("system clock precedes the unix epoch: {error}"))
104 })?;
105 let now = i64::try_from(elapsed.as_secs())?;
106 let mut version = VersionMessage::new(
107 hello.services,
108 now,
109 Address::new(&self.address, ServiceFlags::NONE),
110 Address::new(&"0.0.0.0:0".parse::<SocketAddr>()?, hello.services),
111 rand::random(),
112 hello.user_agent.clone(),
113 i32::try_from(our_height)?,
114 );
115 version.version = P2P_PROTOCOL_VERSION;
116 version.relay = hello.relay;
117 self.send(NetworkMessage::Version(version))?;
118 let deadline = Instant::now() + Duration::from_secs(P2P_HANDSHAKE_TIMEOUT_SECS);
119 let mut versioned = false;
120 let mut acknowledged = false;
121 while !(versioned && acknowledged) {
122 ensure!(Instant::now() < deadline, "peer handshake timed out");
123 match self.receive()? {
124 Message::Version(theirs) => {
125 ensure!(!versioned, "peer sent version twice");
126 ensure!(
127 theirs.services.has(ServiceFlags::WITNESS)
128 && (theirs.services.has(ServiceFlags::NETWORK)
129 || theirs.services.has(ServiceFlags::NETWORK_LIMITED)),
130 "peer {} lacks block and witness service bits",
131 self.address
132 );
133 self.services = theirs.services;
134 self.start_height = u64::try_from(theirs.start_height.max(0))?;
135 self.version = Some(theirs);
136 if hello.negotiate {
137 self.send(NetworkMessage::WtxidRelay)?;
138 self.send(NetworkMessage::SendAddrV2)?;
139 }
140 self.send(NetworkMessage::Verack)?;
141 versioned = true;
142 }
143 Message::Verack => acknowledged = true,
144 other => self.absorb(other)?,
145 }
146 }
147 if hello.negotiate {
148 self.send(NetworkMessage::SendHeaders)?;
149 }
150 self.send(NetworkMessage::GetAddr)?;
151 Ok(())
152 }
153
154 pub(super) fn height_hint(&self, tip: u64) -> u64 {
155 match self.proven_height {
156 None => self.start_height.min(tip + P2P_HEIGHT_SLACK),
157 Some(proven) => proven.max(self.start_height.min(tip)),
158 }
159 }
160
161 pub(super) fn limited(&self) -> bool {
162 !self.services.has(ServiceFlags::NETWORK)
163 }
164
165 fn send(&mut self, payload: NetworkMessage) -> Result<(), Error> {
166 tracing::trace!(peer = %self.address, command = payload.cmd(), "p2p send");
167 wire::send(&mut self.stream, self.magic, payload)
168 }
169
170 pub fn receive(&mut self) -> Result<Message, Error> {
171 let received = wire::receive(&mut self.stream, self.magic);
172 match &received {
173 Ok(message) => {
174 tracing::trace!(peer = %self.address, command = %message.command(), "p2p receive")
175 }
176 Err(error) => tracing::trace!(peer = %self.address, %error, "p2p receive failed"),
177 }
178 received
179 }
180
181 pub(super) fn poll(&mut self) -> Result<(), Error> {
182 let mut byte = [0u8; 1];
183 for _round in 0..P2P_POLL_MESSAGES {
184 self.stream.set_nonblocking(true)?;
185 let peeked = self.stream.peek(&mut byte);
186 self.stream.set_nonblocking(false)?;
187 match peeked {
188 Ok(0) => return Err(Error::Io(ErrorKind::UnexpectedEof.into())),
189 Ok(_waiting) => {
190 self.stream.set_read_timeout(Some(Duration::from_millis(
191 P2P_POLL_READ_TIMEOUT_MILLIS,
192 )))?;
193 let message = self.receive();
194 self.stream
195 .set_read_timeout(Some(Duration::from_secs(P2P_READ_TIMEOUT_SECS)))?;
196 self.absorb(message?)?;
197 }
198 Err(error) if error.kind() == ErrorKind::WouldBlock => return Ok(()),
199 Err(error) => return Err(error.into()),
200 }
201 }
202 Ok(())
203 }
204
205 pub fn absorb_next(&mut self) -> Result<(), Error> {
206 let message = self.receive()?;
207 self.absorb(message)
208 }
209
210 pub fn absorb(&mut self, message: Message) -> Result<(), Error> {
211 match message {
212 Message::Ping(nonce) => self.send(NetworkMessage::Pong(nonce)),
213 Message::Addr(rows) => {
214 for (time, address) in rows {
215 self.learn(address.socket_addr(), address.services, time);
216 }
217 Ok(())
218 }
219 Message::AddrV2(rows) => {
220 for row in rows {
221 self.learn(row.socket_addr(), row.services, row.time);
222 }
223 Ok(())
224 }
225 Message::Version(_version) => Err(Error::Invalid("peer repeated its version".into())),
226 Message::Headers(rows) => {
227 tracing::debug!(peer = %self.address, count = rows.len(), "unsolicited headers ignored");
228 Ok(())
229 }
230 Message::Block(raw) => {
231 let hash = block_hash(&raw)?;
232 match self.requested.contains(&hash) {
233 true => {
234 self.requested.retain(|asked| *asked != hash);
235 self.arrived.push(raw);
236 }
237 false => {
238 tracing::debug!(peer = %self.address, bytes = raw.len(), "unsolicited block dropped")
239 }
240 }
241 Ok(())
242 }
243 Message::Pong(nonce) => {
244 tracing::trace!(peer = %self.address, nonce, "pong");
245 Ok(())
246 }
247 Message::Inv(items) => {
248 tracing::trace!(peer = %self.address, count = items.len(), "inventory announcement ignored");
249 Ok(())
250 }
251 Message::NotFound(items) => {
252 for item in items {
253 if let Inventory::WitnessBlock(hash) | Inventory::Block(hash) = item
254 && self.requested.contains(&hash)
255 {
256 self.requested.retain(|asked| *asked != hash);
257 self.lacking.push(hash);
258 }
259 }
260 Ok(())
261 }
262 Message::Verack => Ok(()),
263 Message::Other(command) => {
264 tracing::trace!(peer = %self.address, %command, "ignored peer message");
265 Ok(())
266 }
267 }
268 }
269
270 fn learn(
271 &mut self,
272 address: Result<SocketAddr, bitcoin::io::Error>,
273 services: ServiceFlags,
274 time: u32,
275 ) {
276 match address {
277 Ok(address) => {
278 if services.has(ServiceFlags::WITNESS)
279 && (services.has(ServiceFlags::NETWORK)
280 || services.has(ServiceFlags::NETWORK_LIMITED))
281 {
282 self.learned.push(Gossiped { address, time });
283 }
284 }
285 Err(error) => tracing::debug!(%error, "gossiped address is not reachable by socket"),
286 }
287 }
288
289 fn wait<T>(
290 &mut self,
291 deadline: Instant,
292 mut matcher: impl FnMut(&mut Self, Message) -> Result<Waited<T>, Error>,
293 ) -> Result<T, Error> {
294 loop {
295 if Instant::now() >= deadline {
296 return Err(Error::Io(io::Error::new(
297 ErrorKind::TimedOut,
298 format!("peer {} did not answer in time", self.address),
299 )));
300 }
301 let message = self.receive()?;
302 match matcher(self, message)? {
303 Waited::Done(value) => return Ok(value),
304 Waited::Keep => (),
305 }
306 }
307 }
308
309 pub(super) fn request_headers(
310 &mut self,
311 locator: Vec<BlockHash>,
312 ) -> Result<Vec<Header>, Error> {
313 let mut request = GetHeadersMessage::new(locator, BlockHash::all_zeros());
314 request.version = P2P_PROTOCOL_VERSION;
315 self.send(NetworkMessage::GetHeaders(request))?;
316 let patience = Duration::from_secs(P2P_HEADERS_TIMEOUT_SECS);
317 self.stream.set_read_timeout(Some(patience))?;
318 let answered = self.wait(Instant::now() + patience, |peer, message| match message {
319 Message::Headers(rows) => Ok(Waited::Done(rows)),
320 other => {
321 peer.absorb(other)?;
322 Ok(Waited::Keep)
323 }
324 });
325 self.stream
326 .set_read_timeout(Some(Duration::from_secs(P2P_READ_TIMEOUT_SECS)))?;
327 answered
328 }
329
330 pub(super) fn outstanding(&self) -> usize {
331 self.requested.len()
332 }
333
334 pub(super) fn ask_blocks(&mut self, hashes: &[BlockHash]) -> Result<(), Error> {
335 ensure!(
336 self.requested.len() + hashes.len() <= P2P_PIPELINE_PER_PEER + 1,
337 "peer {} already holds its block request window",
338 self.address
339 );
340 let items = hashes
341 .iter()
342 .copied()
343 .map(Inventory::WitnessBlock)
344 .collect();
345 self.send(NetworkMessage::GetData(items))?;
346 self.requested.extend_from_slice(hashes);
347 Ok(())
348 }
349
350 pub(super) fn await_block(&mut self, hash: BlockHash) -> Result<Vec<u8>, Error> {
351 if self.lacking.contains(&hash) {
352 self.lacking.retain(|missing| *missing != hash);
353 return Err(self.missing(hash));
354 }
355 for (position, raw) in self.arrived.iter().enumerate() {
356 if block_hash(raw)? == hash {
357 return Ok(self.arrived.remove(position));
358 }
359 }
360 let patience = Duration::from_secs(P2P_BLOCK_PEER_TIMEOUT_SECS);
361 self.stream.set_read_timeout(Some(patience))?;
362 let answered = self.wait(Instant::now() + patience, |peer, message| match message {
363 Message::Block(raw) if block_hash(&raw)? == hash => {
364 peer.requested.retain(|asked| *asked != hash);
365 Ok(Waited::Done(raw))
366 }
367 Message::NotFound(rows)
368 if rows.iter().any(|item| {
369 matches!(item, Inventory::WitnessBlock(missing) | Inventory::Block(missing) if *missing == hash)
370 }) =>
371 {
372 peer.requested.retain(|asked| *asked != hash);
373 Err(peer.missing(hash))
374 }
375 other => {
376 peer.absorb(other)?;
377 Ok(Waited::Keep)
378 }
379 });
380 self.stream
381 .set_read_timeout(Some(Duration::from_secs(P2P_READ_TIMEOUT_SECS)))?;
382 answered
383 }
384
385 fn missing(&self, hash: BlockHash) -> Error {
386 Error::Missing(format!("peer {} does not have block {hash}", self.address))
387 }
388}
389
390pub(super) fn block_hash(raw: &[u8]) -> Result<BlockHash, Error> {
391 let header: Header = deserialize(
392 raw.get(..Header::SIZE)
393 .context("block shorter than a header")?,
394 )?;
395 Ok(header.block_hash())
396}