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