urma_node/
peer_parent.rs

1//! Raw commit parents from configured URMA peers over the overlay.
2use 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
17/// A metadata-bound parent reply. Hex length is checked, bytes are not yet.
18pub(crate) enum Parent {
19    Hex(String),
20    /// Missing, TooLarge, WrongNetwork or UnsupportedVersion: definitive answers.
21    Absent,
22    Busy,
23}
24
25/// Busy is capacity, not an answer. Hex is length-checked before any decode.
26/// TooLarge, WrongNetwork and UnsupportedVersion are definitive for that peer.
27pub(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
47/// Tries one window of configured peers per fetch; successive fetches move to
48/// the next window, so every peer is visited. verify_parent binds bytes to the
49/// TXID; a peer whose bytes fail it only lets the next peer be tried and is
50/// disconnected so it keeps no outbound slot. Engine verification is final.
51pub struct OverlayCommitSource {
52    chain: Chain,
53    peers: Vec<PeerSource>,
54    next: AtomicUsize,
55}
56
57impl OverlayCommitSource {
58    /// No more peers than outbound connections the limits admit at once.
59    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    /// Busy, local or remote, is no answer. Local Busy from request() also
77    /// covers deadline expiry, a dial denied by our connection limits and a
78    /// stopped network, so these are silent (Ok(None), nothing in status);
79    /// FETCH_DEADLINE bounds the whole round. Zero answers plus any other peer
80    /// or transport error is an error; otherwise absence. First valid wins.
81    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}
urma_node/peer_parent.rs

urma_node/
peer_parent.rs
