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_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
134/// A listing whose block differs from our selected header at that height comes
135/// from another view of the chain (a reorg in flight); it is not a peer error.
136pub(crate) fn selected(engine: &Engine, entry: &Placement) -> Result<bool, Error> {
137    match engine.archive()?.ticket(entry.height) {
138        Ok(ticket) => Ok(ticket.point().hash == entry.block_hash),
139        Err(Error::Invalid(_outside_selected_view)) => Ok(false),
140        Err(error) => Err(error),
141    }
142}
143
144pub(crate) fn placed(receipt: &RecordReceipt, entry: &Placement) -> bool {
145    receipt.block.height == entry.height
146        && receipt.block.hash == entry.block_hash
147        && receipt.position == entry.position
148        && receipt.txid == entry.txid
149        && receipt.wtxid == entry.wtxid
150}
151
152impl PeerSource {
153    /// One bounded inventory page. Missing or Busy, remote or local (including
154    /// deadline and stopped network), is None; refusals and limits are errors.
155    pub fn inventory(
156        &self,
157        from_height: u64,
158        to_height: u64,
159        after: Option<Placement>,
160        limit: u32,
161        cancelled: &AtomicBool,
162    ) -> Result<Option<InventoryPage>, Error> {
163        let range = (from_height, to_height);
164        let deadline = Instant::now() + CALL_TIMEOUT;
165        match self.inventory_until(range, after, limit, cancelled, deadline) {
166            Err(Error::Busy(_) | Error::PeerBusy) => Ok(None),
167            other => other,
168        }
169    }
170
171    /// One record request bounded by CALL_TIMEOUT and the caller's `deadline`.
172    pub(crate) fn fetch_until(
173        &self,
174        txid: Txid,
175        deadline: Instant,
176    ) -> Result<Option<RecordPackage>, Error> {
177        self.request(Kind::Record { txid }, deadline, Answer::Record)
178    }
179
180    /// As `inventory`, also bounded by the caller's own `deadline`; Busy,
181    /// remote or local, stays an error so the scheduler can count it.
182    pub(crate) fn inventory_until(
183        &self,
184        (from_height, to_height): (u64, u64),
185        after: Option<Placement>,
186        limit: u32,
187        cancelled: &AtomicBool,
188        deadline: Instant,
189    ) -> Result<Option<InventoryPage>, Error> {
190        check_cancelled(cancelled)?;
191        let kind = Kind::Inventory {
192            from_height,
193            to_height,
194            after,
195            limit,
196        };
197        self.request(kind, deadline, Answer::Inventory)
198    }
199}
urma_node/peer_inventory.rs

urma_node/
peer_inventory.rs
