1use crate::{
3 Error,
4 peer::{PeerSource, Pending, bind_response},
5 peer_limits::ESTABLISHED_PER_DIRECTION,
6 peer_protocol::{PARENT_LIMIT, Reply, Response},
7 source::{CommitSource, check_cancelled, verify_parent},
8};
9use bitcoin::{BlockHash, Txid};
10use libp2p::PeerId;
11use std::{
12 sync::atomic::{AtomicBool, AtomicUsize, Ordering},
13 time::{Duration, Instant},
14};
15use urma_chain::observation::Chain;
16
17pub(crate) enum Parent {
19 Hex(String),
20 Absent,
22 Busy,
23}
24
25pub(crate) fn validate_parent(
28 genesis: BlockHash,
29 peer: PeerId,
30 pending: &Pending,
31 response: Response,
32) -> Result<Parent, Error> {
33 bind_response(genesis, peer, pending, &response)?;
34 match response.result {
35 Reply::Parent(hex) if hex.len() <= 2 * PARENT_LIMIT => Ok(Parent::Hex(hex)),
36 Reply::Parent(_) => Err(Error::Invalid("peer parent exceeds byte limit".into())),
37 Reply::Busy => Ok(Parent::Busy),
38 Reply::Missing | Reply::TooLarge | Reply::WrongNetwork | Reply::UnsupportedVersion => {
39 Ok(Parent::Absent)
40 }
41 Reply::Record(_) | Reply::Inventory(_) | Reply::Peers(_) | Reply::Refused => Err(
42 Error::Invalid("peer answered a parent request with another kind or a refusal".into()),
43 ),
44 }
45}
46
47pub struct OverlayCommitSource {
52 chain: Chain,
53 peers: Vec<PeerSource>,
54 next: AtomicUsize,
55}
56
57impl OverlayCommitSource {
58 const PEERS_PER_FETCH: usize = ESTABLISHED_PER_DIRECTION;
60 const FETCH_DEADLINE: Duration = Duration::from_secs(20);
61
62 pub(crate) fn new(chain: Chain, peers: Vec<PeerSource>) -> Self {
63 Self {
64 chain,
65 peers,
66 next: AtomicUsize::new(0),
67 }
68 }
69}
70
71impl CommitSource for OverlayCommitSource {
72 fn chain(&self) -> Chain {
73 self.chain
74 }
75
76 fn fetch(&self, txid: Txid, cancelled: &AtomicBool) -> Result<Option<Vec<u8>>, Error> {
82 check_cancelled(cancelled)?;
83 let count = self.peers.len();
84 if count == 0 {
85 return Ok(None);
86 }
87 let deadline = Instant::now() + Self::FETCH_DEADLINE;
88 let window = count.min(Self::PEERS_PER_FETCH);
89 let start = self.next.fetch_add(window, Ordering::Relaxed) % count;
90 let mut answered = false;
91 let mut failure = None;
92 for offset in 0..window {
93 check_cancelled(cancelled)?;
94 if Instant::now() >= deadline {
95 break;
96 }
97 let peer = &self.peers[(start + offset) % count];
98 match peer.parent(txid, deadline) {
99 Ok(Parent::Hex(hex)) => match self.decode(&hex, txid) {
100 Ok(raw) => return Ok(Some(raw)),
101 Err(error) => {
102 peer.disconnect();
103 failure = Some(error);
104 }
105 },
106 Ok(Parent::Absent) => answered = true,
107 Ok(Parent::Busy) | Err(Error::Busy(_)) => (),
108 Err(Error::DialDenied(_local_capacity)) => (),
109 Err(error) => {
110 peer.disconnect();
111 failure = Some(error);
112 }
113 }
114 }
115 match failure {
116 Some(error) if !answered => Err(Error::CommitSource(format!("overlay: {error}"))),
117 _ => Ok(None),
118 }
119 }
120}
121
122impl OverlayCommitSource {
123 fn decode(&self, hex: &str, txid: Txid) -> Result<Vec<u8>, Error> {
124 if !hex
125 .bytes()
126 .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'))
127 {
128 return Err(Error::Invalid("peer parent must be lowercase hex".into()));
129 }
130 let raw = hex::decode(hex)
131 .map_err(|error| Error::Invalid(format!("peer parent hex: {error}")))?;
132 verify_parent(self.chain, txid, &raw)?;
133 Ok(raw)
134 }
135}