urma_node/
peer_inventory.rs

1//! Serving local lookups and bounded inventory pages over the overlay, plus the
2//! client-side inventory request. Pages describe only fully scanned blocks.
3use crate::{
4    Error,
5    engine::Engine,
6    model::RecordPackage,
7    peer::{Answer, CALL_TIMEOUT, PeerSource, Pending, bind_response},
8    peer_protocol::{
9        INVENTORY_ENTRIES, INVENTORY_LIMIT, INVENTORY_SPAN, InventoryPage, Kind, PARENT_LIMIT,
10        Placement, Reply, Request, Response,
11    },
12    source::check_cancelled,
13};
14use bitcoin::{BlockHash, Txid};
15use libp2p::PeerId;
16use std::{sync::atomic::AtomicBool, time::Instant};
17
18/// Local retained bytes only: never scans, fetches or asks another peer.
19pub(crate) fn lookup(engine: &Engine, request: &Request) -> Reply {
20    match request.kind {
21        Kind::Record { txid } => match engine.record(txid) {
22            Ok(Some(record)) => Reply::Record(Box::new(record.package)),
23            Ok(None) => Reply::Missing,
24            Err(_) => Reply::Busy,
25        },
26        Kind::Parent { txid } => match engine.raw_transaction(txid) {
27            Ok(Some(raw)) if raw.len() > PARENT_LIMIT => Reply::TooLarge,
28            Ok(Some(raw)) => Reply::Parent(hex::encode(raw)),
29            Ok(None) => Reply::Missing,
30            Err(_) => Reply::Busy,
31        },
32        Kind::Inventory {
33            from_height,
34            to_height,
35            after,
36            limit,
37        } => inventory(engine, (from_height, to_height), after, limit),
38        // Gossip is served by the network thread from memory, never from here.
39        Kind::Peers { .. } => Reply::Refused,
40    }
41}
42
43fn admissible(from: u64, to: u64, limit: u32) -> bool {
44    to >= from && to - from <= INVENTORY_SPAN && (1..=INVENTORY_ENTRIES).contains(&limit)
45}
46
47fn inventory(engine: &Engine, range: (u64, u64), after: Option<Placement>, limit: u32) -> Reply {
48    const ENVELOPE: usize = 1024;
49    let (from, to) = range;
50    if !admissible(from, to, limit) {
51        return Reply::Refused;
52    }
53    let cursor = after.map(|placement| (placement.height, placement.position));
54    let rows = match engine.inventory_page(from, to, cursor, limit) {
55        Ok(rows) => rows,
56        Err(_) => return Reply::Busy,
57    };
58    let entries = rows
59        .entries
60        .into_iter()
61        .map(|(block, position, txid, wtxid)| Placement {
62            height: block.height,
63            block_hash: block.hash,
64            position,
65            txid,
66            wtxid,
67        })
68        .collect();
69    let page = InventoryPage {
70        scanned_through: rows.scanned_through,
71        entries,
72        more: rows.more,
73    };
74    match serde_json::to_vec(&page) {
75        Ok(bytes) if bytes.len() + ENVELOPE <= INVENTORY_LIMIT => Reply::Inventory(page),
76        _ => Reply::TooLarge,
77    }
78}
79
80/// Missing is no answer and Busy is retryable. Refusals, limits and other
81/// kinds are errors.
82pub(crate) fn validate_inventory(
83    genesis: BlockHash,
84    peer: PeerId,
85    pending: &Pending,
86    response: Response,
87) -> Result<Option<InventoryPage>, Error> {
88    bind_response(genesis, peer, pending, &response)?;
89    match response.result {
90        Reply::Inventory(page) => check_page(&response.kind, page).map(Some),
91        Reply::Missing => Ok(None),
92        Reply::Busy => Err(Error::PeerBusy),
93        Reply::Record(_) | Reply::Parent(_) | Reply::Peers(_) => Err(Error::Invalid(
94            "peer answered an inventory request with another kind".into(),
95        )),
96        Reply::Refused | Reply::TooLarge | Reply::WrongNetwork | Reply::UnsupportedVersion => {
97            Err(Error::Invalid("peer refused the inventory request".into()))
98        }
99    }
100}
101
102/// Entries stay inside the requested range and limit, strictly after the
103/// cursor in (height, position) order. Placement truth is the caller's check.
104fn check_page(kind: &Kind, page: InventoryPage) -> Result<InventoryPage, Error> {
105    let Kind::Inventory {
106        from_height,
107        to_height,
108        after,
109        limit,
110    } = *kind
111    else {
112        return Err(Error::Invalid("inventory reply to another request".into()));
113    };
114    let limit = usize::try_from(limit).map_err(|e| Error::Invalid(e.to_string()))?;
115    if page.entries.len() > limit {
116        return Err(Error::Invalid(
117            "peer inventory exceeds the requested limit".into(),
118        ));
119    }
120    let mut previous = after.map(|placement| (placement.height, placement.position));
121    for entry in &page.entries {
122        let at = (entry.height, entry.position);
123        let ordered = previous.is_none_or(|last| at > last);
124        if entry.height < from_height || entry.height > to_height || !ordered {
125            return Err(Error::Invalid(
126                "peer inventory entry out of range or order".into(),
127            ));
128        }
129        previous = Some(at);
130    }
131    Ok(page)
132}
133
134impl PeerSource {
135    /// One bounded inventory page. Missing or Busy, remote or local (including
136    /// deadline and stopped network), is None; refusals and limits are errors.
137    pub fn inventory(
138        &self,
139        from_height: u64,
140        to_height: u64,
141        after: Option<Placement>,
142        limit: u32,
143        cancelled: &AtomicBool,
144    ) -> Result<Option<InventoryPage>, Error> {
145        let range = (from_height, to_height);
146        let deadline = Instant::now() + CALL_TIMEOUT;
147        match self.inventory_until(range, after, limit, cancelled, deadline) {
148            Err(Error::Busy(_) | Error::PeerBusy) => Ok(None),
149            other => other,
150        }
151    }
152
153    /// One record request bounded by CALL_TIMEOUT and the caller's `deadline`.
154    pub(crate) fn fetch_until(
155        &self,
156        txid: Txid,
157        deadline: Instant,
158    ) -> Result<Option<RecordPackage>, Error> {
159        self.request(Kind::Record { txid }, deadline, Answer::Record)
160    }
161
162    /// As `inventory`, also bounded by the caller's own `deadline`; Busy,
163    /// remote or local, stays an error so the scheduler can count it.
164    pub(crate) fn inventory_until(
165        &self,
166        (from_height, to_height): (u64, u64),
167        after: Option<Placement>,
168        limit: u32,
169        cancelled: &AtomicBool,
170        deadline: Instant,
171    ) -> Result<Option<InventoryPage>, Error> {
172        check_cancelled(cancelled)?;
173        let kind = Kind::Inventory {
174            from_height,
175            to_height,
176            after,
177            limit,
178        };
179        self.request(kind, deadline, Answer::Inventory)
180    }
181}
urma_node/peer_inventory.rs

urma_node/
peer_inventory.rs
