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, RecordReceipt},
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_from: rows.scanned_from,
71        scanned_through: rows.scanned_through,
72        entries,
73        more: rows.more,
74    };
75    match serde_json::to_vec(&page) {
76        Ok(bytes) if bytes.len() + ENVELOPE <= INVENTORY_LIMIT => Reply::Inventory(page),
77        _ => Reply::TooLarge,
78    }
79}
80
81/// Missing is no answer and Busy is retryable. Refusals, limits and other
82/// kinds are errors.
83pub(crate) fn validate_inventory(
84    genesis: BlockHash,
85    peer: PeerId,
86    pending: &Pending,
87    response: Response,
88) -> Result<Option<InventoryPage>, Error> {
89    bind_response(genesis, peer, pending, &response)?;
90    match response.result {
91        Reply::Inventory(page) => check_page(&response.kind, page).map(Some),
92        Reply::Missing => Ok(None),
93        Reply::Busy => Err(Error::PeerBusy),
94        Reply::Record(_) | Reply::Parent(_) | Reply::Peers(_) => Err(Error::Invalid(
95            "peer answered an inventory request with another kind".into(),
96        )),
97        Reply::Refused | Reply::TooLarge | Reply::WrongNetwork | Reply::UnsupportedVersion => {
98            Err(Error::Invalid("peer refused the inventory request".into()))
99        }
100    }
101}
102
103/// Entries stay inside the requested range and limit, strictly after the
104/// cursor in (height, position) order. Placement truth is the caller's check.
105fn check_page(kind: &Kind, page: InventoryPage) -> Result<InventoryPage, Error> {
106    let Kind::Inventory {
107        from_height,
108        to_height,
109        after,
110        limit,
111    } = *kind
112    else {
113        return Err(Error::Invalid("inventory reply to another request".into()));
114    };
115    let limit = usize::try_from(limit).map_err(|e| Error::Invalid(e.to_string()))?;
116    if page.entries.len() > limit {
117        return Err(Error::Invalid(
118            "peer inventory exceeds the requested limit".into(),
119        ));
120    }
121    let mut previous = after.map(|placement| (placement.height, placement.position));
122    for entry in &page.entries {
123        let at = (entry.height, entry.position);
124        let ordered = previous.is_none_or(|last| at > last);
125        if entry.height < from_height || entry.height > to_height || !ordered {
126            return Err(Error::Invalid(
127                "peer inventory entry out of range or order".into(),
128            ));
129        }
130        previous = Some(at);
131    }
132    Ok(page)
133}
134
135/// A listing whose block differs from our selected header at that height comes
136/// from another view of the chain (a reorg in flight); it is not a peer error.
137pub(crate) fn selected(engine: &Engine, entry: &Placement) -> Result<bool, Error> {
138    match engine.archive()?.ticket(entry.height) {
139        Ok(ticket) => Ok(ticket.point().hash == entry.block_hash),
140        Err(Error::Invalid(_outside_selected_view)) => Ok(false),
141        Err(error) => Err(error),
142    }
143}
144
145pub(crate) fn placed(receipt: &RecordReceipt, entry: &Placement) -> bool {
146    receipt.block.height == entry.height
147        && receipt.block.hash == entry.block_hash
148        && receipt.position == entry.position
149        && receipt.txid == entry.txid
150        && receipt.wtxid == entry.wtxid
151}
152
153impl PeerSource {
154    /// One bounded inventory page. Missing or Busy, remote or local (including
155    /// deadline and stopped network), is None; refusals and limits are errors.
156    pub fn inventory(
157        &self,
158        from_height: u64,
159        to_height: u64,
160        after: Option<Placement>,
161        limit: u32,
162        cancelled: &AtomicBool,
163    ) -> Result<Option<InventoryPage>, Error> {
164        let range = (from_height, to_height);
165        let deadline = Instant::now() + CALL_TIMEOUT;
166        match self.inventory_until(range, after, limit, cancelled, deadline) {
167            Err(Error::Busy(_) | Error::PeerBusy) => Ok(None),
168            other => other,
169        }
170    }
171
172    /// One record request bounded by CALL_TIMEOUT and the caller's `deadline`.
173    pub(crate) fn fetch_until(
174        &self,
175        txid: Txid,
176        deadline: Instant,
177    ) -> Result<Option<RecordPackage>, Error> {
178        self.request(Kind::Record { txid }, deadline, Answer::Record)
179    }
180
181    /// As `inventory`, also bounded by the caller's own `deadline`; Busy,
182    /// remote or local, stays an error so the scheduler can count it.
183    pub(crate) fn inventory_until(
184        &self,
185        (from_height, to_height): (u64, u64),
186        after: Option<Placement>,
187        limit: u32,
188        cancelled: &AtomicBool,
189        deadline: Instant,
190    ) -> Result<Option<InventoryPage>, Error> {
191        check_cancelled(cancelled)?;
192        let kind = Kind::Inventory {
193            from_height,
194            to_height,
195            after,
196            limit,
197        };
198        self.request(kind, deadline, Answer::Inventory)
199    }
200}
urma_node/peer_inventory.rs

urma_node/
peer_inventory.rs
