urma_node/
peer_parent.rs

1//! Raw commit parents from configured URMA peers over the overlay.
2use crate::{
3    Error,
4    peer::{CALL_TIMEOUT, Pending, SyncPeers, 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,
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 the current overlay peers (configured, then live) per
48/// fetch; successive fetches move to the next window, so every peer is
49/// visited. verify_parent binds bytes to the TXID; a peer whose bytes fail it
50/// only lets the next peer be tried and is disconnected so it keeps no slot.
51/// Engine verification is final.
52pub struct OverlayCommitSource {
53    chain: Chain,
54    peers: SyncPeers,
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: SyncPeers) -> Self {
63        Self { chain, peers }
64    }
65}
66
67impl CommitSource for OverlayCommitSource {
68    fn chain(&self) -> Chain {
69        self.chain
70    }
71
72    /// Busy, local or remote, is no answer. Local Busy from request() also
73    /// covers deadline expiry, a dial denied by our connection limits and a
74    /// stopped network, so these are silent (Ok(None), nothing in status);
75    /// FETCH_DEADLINE bounds the whole round. A configured peer, which may
76    /// need a cold dial, gets a full call bound of what is left; the live
77    /// tail splits the rest evenly, so stalling live peers cannot eat a later
78    /// one's turn. Zero answers plus any other peer or transport error is an
79    /// error; otherwise absence. First valid wins.
80    fn fetch(&self, txid: Txid, cancelled: &AtomicBool) -> Result<Option<Vec<u8>>, Error> {
81        check_cancelled(cancelled)?;
82        let deadline = Instant::now() + Self::FETCH_DEADLINE;
83        let mut answered = false;
84        let mut failure = None;
85        let window = self.peers.window(Self::PEERS_PER_FETCH);
86        let mut live_left = window.iter().filter(|peer| !peer.configured()).count();
87        for peer in &window {
88            check_cancelled(cancelled)?;
89            let left = deadline.saturating_duration_since(Instant::now());
90            let share = match peer.configured() {
91                true => Some(left.min(CALL_TIMEOUT)),
92                false => u32::try_from(live_left)
93                    .ok()
94                    .and_then(|peers| left.checked_div(peers)),
95            };
96            let Some(share) = share.filter(|share| !share.is_zero()) else {
97                break;
98            };
99            live_left = live_left.saturating_sub(usize::from(!peer.configured()));
100            match peer.parent(txid, Instant::now() + share) {
101                Ok(Parent::Hex(hex)) => match self.decode(&hex, txid) {
102                    Ok(raw) => return Ok(Some(raw)),
103                    Err(error) => {
104                        peer.disconnect();
105                        failure = Some(error);
106                    }
107                },
108                Ok(Parent::Absent) => answered = true,
109                Ok(Parent::Busy) | Err(Error::Busy(_)) => (),
110                Err(Error::DialDenied(_local_capacity)) => (),
111                Err(error) => {
112                    peer.disconnect();
113                    failure = Some(error);
114                }
115            }
116        }
117        match failure {
118            Some(error) if !answered => Err(Error::CommitSource(format!("overlay: {error}"))),
119            _ => Ok(None),
120        }
121    }
122}
123
124impl OverlayCommitSource {
125    fn decode(&self, hex: &str, txid: Txid) -> Result<Vec<u8>, Error> {
126        if !hex
127            .bytes()
128            .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'))
129        {
130            return Err(Error::Invalid("peer parent must be lowercase hex".into()));
131        }
132        let raw = hex::decode(hex)
133            .map_err(|error| Error::Invalid(format!("peer parent hex: {error}")))?;
134        verify_parent(self.chain, txid, &raw)?;
135        Ok(raw)
136    }
137}
urma_node/peer_parent.rs

urma_node/
peer_parent.rs
