urma_node/
peer.rs

1//! Explicitly started URMA peer transport. Noise authenticates the remote
2//! transport identity; packages still need independent Engine admission.
3//! There are no built-in seeds, wallet keys or implicit listeners. Names in
4//! explicitly dialed `/dns*` addresses resolve through the system resolver.
5use crate::{
6    DialLimit, Error,
7    engine::Engine,
8    model::RecordPackage,
9    peer_codec::Codec,
10    peer_gossip::{Dial, Dials, Shared, is_loopback, validate_announce, validate_peers},
11    peer_inventory::{lookup, validate_inventory},
12    peer_limits::{LimitDenied, Limits},
13    peer_live::Link,
14    peer_parent::{Parent, validate_parent},
15    peer_protocol::{GossipPeer, InventoryPage, Kind, PROTOCOL, Reply, Request, Response, VERSION},
16    peer_rate::{Bucket, Count, peer_rate},
17    peerbook::PeerAddress,
18};
19pub use crate::{
20    peer_live::{PeerSource, SyncPeers},
21    peer_parent::OverlayCommitSource,
22};
23use bitcoin::{BlockHash, Txid};
24use libp2p::{
25    Multiaddr, PeerId, StreamProtocol, Swarm, SwarmBuilder,
26    futures::StreamExt,
27    identify,
28    identity::Keypair,
29    multiaddr::Protocol,
30    noise,
31    request_response::{self, Message, OutboundRequestId, ResponseChannel},
32    swarm::{ConnectionId, DialError, NetworkBehaviour, SwarmEvent},
33    tcp, yamux,
34};
35use std::{
36    collections::{HashMap, HashSet},
37    num::{NonZeroU8, NonZeroUsize},
38    sync::{
39        Arc,
40        atomic::{AtomicBool, AtomicU64, Ordering},
41        mpsc,
42    },
43    thread::JoinHandle,
44    time::{Duration, Instant},
45};
46use tokio::sync::mpsc as async_mpsc;
47use urma_chain::observation::Chain;
48
49const REQUEST_TIMEOUT: Duration = Duration::from_secs(8);
50pub(crate) const CALL_TIMEOUT: Duration = Duration::from_secs(10);
51const MAX_PENDING: usize = 8;
52const IDENTIFY_PROTOCOL: &str = "/urma/node/4";
53
54#[derive(NetworkBehaviour)]
55pub(crate) struct Behaviour {
56    // Admission must precede request-response connection side effects.
57    limits: Limits,
58    records: request_response::Behaviour<Codec>,
59    identify: identify::Behaviour,
60}
61
62/// Owns one network thread and one bounded archive-reader thread. Drop requests
63/// shutdown and joins both. An archive operation already running is not forcibly
64/// interrupted. Source handles do not keep the network alive after this owner.
65pub struct PeerNetwork {
66    peer: PeerId,
67    listen: Option<Multiaddr>,
68    pub(crate) commands: async_mpsc::Sender<Command>,
69    pub(crate) stop: Arc<AtomicBool>,
70    connected: Arc<AtomicU64>,
71    pub(crate) shared: Arc<Shared>,
72    worker: Option<JoinHandle<Result<(), Error>>>,
73}
74
75pub(crate) enum Command {
76    Fetch(Fetch),
77    /// Fire and forget: frees the slot held by a peer whose reply was rejected.
78    Disconnect(PeerId),
79    Dial(Dial),
80}
81
82/// A request to one peer, over its existing connection or the given addresses.
83pub(crate) struct Fetch {
84    peer: PeerId,
85    addresses: Vec<Multiaddr>,
86    kind: Kind,
87    deadline: Instant,
88    answer: Answer,
89}
90
91pub(crate) struct Pending {
92    peer: PeerId,
93    pub(crate) kind: Kind,
94    deadline: Instant,
95    answer: Answer,
96}
97
98/// Reply channel typed by request kind; a cross-kind reply is a peer error.
99pub(crate) enum Answer {
100    Record(mpsc::SyncSender<Result<Option<RecordPackage>, Error>>),
101    Parent(mpsc::SyncSender<Result<Parent, Error>>),
102    Inventory(mpsc::SyncSender<Result<Option<InventoryPage>, Error>>),
103    Peers(mpsc::SyncSender<Result<Vec<GossipPeer>, Error>>),
104}
105
106impl Answer {
107    fn fail(&self, error: Error) {
108        match self {
109            Answer::Record(sender) => {
110                let _ = sender.try_send(Err(error));
111            }
112            Answer::Parent(sender) => {
113                let _ = sender.try_send(Err(error));
114            }
115            Answer::Inventory(sender) => {
116                let _ = sender.try_send(Err(error));
117            }
118            Answer::Peers(sender) => {
119                let _ = sender.try_send(Err(error));
120            }
121        }
122    }
123
124    fn deliver(&self, genesis: BlockHash, peer: PeerId, pending: &Pending, response: Response) {
125        match self {
126            Answer::Record(sender) => {
127                let _ = sender.try_send(validate_response(genesis, peer, pending, response));
128            }
129            Answer::Parent(sender) => {
130                let _ = sender.try_send(validate_parent(genesis, peer, pending, response));
131            }
132            Answer::Inventory(sender) => {
133                let _ = sender.try_send(validate_inventory(genesis, peer, pending, response));
134            }
135            Answer::Peers(sender) => {
136                let _ = sender.try_send(validate_peers(genesis, peer, pending, response));
137            }
138        }
139    }
140}
141
142type Startup = mpsc::SyncSender<Result<Option<Multiaddr>, Error>>;
143struct ServeJob {
144    deadline: Instant,
145    request: Request,
146    channel: ResponseChannel<Response>,
147}
148struct Served {
149    response: Response,
150    channel: ResponseChannel<Response>,
151    bytes: usize,
152}
153
154impl PeerNetwork {
155    /// Listen only when an explicit IP/TCP address is supplied. Port zero is
156    /// allowed for an ephemeral listener; returned address includes this PeerId.
157    /// Merely constructing source handles never dials. No peer is a seed by default.
158    /// `announce` are operator-chosen public IP/TCP addresses identify advertises.
159    pub fn start(
160        engine: Arc<Engine>,
161        identity: Keypair,
162        listen: Option<Multiaddr>,
163        announce: Vec<Multiaddr>,
164    ) -> Result<Self, Error> {
165        if let Some(address) = &listen {
166            validate_listener(address)?;
167        }
168        engine.status()?;
169        let peer = identity.public().to_peer_id();
170        let announce = announce
171            .iter()
172            .map(|address| validate_announce(address, peer))
173            .collect::<Result<Vec<_>, _>>()?;
174        let (commands, receiver) = async_mpsc::channel(MAX_PENDING);
175        let (startup, ready) = mpsc::sync_channel(1);
176        let stop = Arc::new(AtomicBool::new(false));
177        let connected = Arc::new(AtomicU64::new(0));
178        let shared = Arc::new(Shared::new(peer));
179        let control = Control {
180            commands: receiver,
181            stop: stop.clone(),
182            connected: connected.clone(),
183            startup,
184            shared: shared.clone(),
185            announce,
186        };
187        let worker = std::thread::Builder::new()
188            .name("urma-peer".into())
189            .spawn(move || {
190                let runtime = tokio::runtime::Builder::new_current_thread()
191                    .enable_all()
192                    .build()?;
193                runtime.block_on(run(engine, identity, listen, control))
194            })?;
195        let actual = match ready.recv_timeout(Duration::from_secs(5)) {
196            Ok(Ok(address)) => address,
197            result => {
198                stop.store(true, Ordering::Release);
199                let joined = worker.join();
200                return Err(match result {
201                    Ok(Err(error)) => error,
202                    _ => match joined {
203                        Ok(Err(error)) => error,
204                        _ => Error::Busy("peer listener startup failed or expired".into()),
205                    },
206                });
207            }
208        };
209        Ok(Self {
210            peer,
211            listen: actual,
212            commands,
213            stop,
214            connected,
215            shared,
216            worker: Some(worker),
217        })
218    }
219
220    pub fn peer_id(&self) -> PeerId {
221        self.peer
222    }
223    /// Distinct peers with at least one established connection right now.
224    pub fn connected_peers(&self) -> u64 {
225        self.connected.load(Ordering::Relaxed)
226    }
227    pub fn listen_address(&self) -> Option<Multiaddr> {
228        self.listen.clone()
229    }
230
231    /// Every address the listener is bound to now, with `/p2p`.
232    pub fn listen_addresses(&self) -> Vec<Multiaddr> {
233        self.shared.listeners()
234    }
235
236    fn link(&self) -> Link {
237        Link {
238            commands: self.commands.clone(),
239            stop: self.stop.clone(),
240            shared: Arc::clone(&self.shared),
241        }
242    }
243
244    pub fn source(&self, address: PeerAddress) -> PeerSource {
245        self.link()
246            .source(address.peer_id(), vec![address.address().clone()])
247    }
248
249    /// `manual` peers first, then whoever is connected and on our chain.
250    pub fn sync_peers(&self, manual: Vec<PeerAddress>) -> SyncPeers {
251        SyncPeers::new(self.link(), manual)
252    }
253
254    pub fn commit_source(&self, chain: Chain, manual: Vec<PeerAddress>) -> OverlayCommitSource {
255        OverlayCommitSource::new(chain, self.sync_peers(manual))
256    }
257
258    pub fn shutdown(mut self) -> Result<(), Error> {
259        self.stop.store(true, Ordering::Release);
260        match self.worker.take().expect("network owner has worker").join() {
261            Ok(result) => result,
262            Err(_) => Err(Error::Invalid("peer worker panicked".into())),
263        }
264    }
265}
266
267impl Drop for PeerNetwork {
268    fn drop(&mut self) {
269        self.stop.store(true, Ordering::Release);
270        if let Some(worker) = self.worker.take() {
271            let _ = worker.join();
272        }
273    }
274}
275
276/// One bounded request: CALL_TIMEOUT and the caller's own deadline, if earlier.
277/// Without addresses the peer must already be connected or known to the swarm.
278pub(crate) fn request<T>(
279    commands: &async_mpsc::Sender<Command>,
280    stop: &AtomicBool,
281    (peer, addresses): (PeerId, Vec<Multiaddr>),
282    kind: Kind,
283    limit: Instant,
284    answer: fn(mpsc::SyncSender<Result<T, Error>>) -> Answer,
285) -> Result<T, Error> {
286    if stop.load(Ordering::Acquire) {
287        return Err(Error::Busy("peer network stopped".into()));
288    }
289    let deadline = limit.min(Instant::now() + CALL_TIMEOUT);
290    let (sender, receiver) = mpsc::sync_channel(1);
291    commands
292        .try_send(Command::Fetch(Fetch {
293            peer,
294            addresses,
295            kind,
296            deadline,
297            answer: answer(sender),
298        }))
299        .map_err(|_| Error::Busy("peer command queue unavailable".into()))?;
300    receiver
301        .recv_timeout(deadline.saturating_duration_since(Instant::now()))
302        .map_err(|_| Error::Busy("peer acquisition deadline or shutdown".into()))?
303}
304
305impl PeerSource {
306    pub(crate) fn parent(&self, txid: Txid, deadline: Instant) -> Result<Parent, Error> {
307        self.request(Kind::Parent { txid }, deadline, Answer::Parent)
308    }
309}
310
311fn validate_listener(address: &Multiaddr) -> Result<(), Error> {
312    let mut parts = address.iter();
313    let ip = matches!(
314        parts.next(),
315        Some(Protocol::Ip4(_)) | Some(Protocol::Ip6(_))
316    );
317    if !ip || !matches!(parts.next(), Some(Protocol::Tcp(_))) || parts.next().is_some() {
318        return Err(Error::Invalid(
319            "peer listener requires only IP and TCP".into(),
320        ));
321    }
322    Ok(())
323}
324
325fn swarm(identity: Keypair) -> Result<Swarm<Behaviour>, Error> {
326    let records = request_response::Behaviour::with_codec(
327        Codec,
328        [(
329            StreamProtocol::new(PROTOCOL),
330            request_response::ProtocolSupport::Full,
331        )],
332        request_response::Config::default()
333            .with_request_timeout(REQUEST_TIMEOUT)
334            .with_max_concurrent_streams(2),
335    );
336    let identify = identify::Behaviour::new(
337        identify::Config::new(IDENTIFY_PROTOCOL.into(), identity.public())
338            .with_agent_version(format!("urma-node/{}", env!("CARGO_PKG_VERSION"))),
339    );
340    let limits = Limits::default();
341    Ok(SwarmBuilder::with_existing_identity(identity)
342        .with_tokio()
343        .with_tcp(
344            tcp::Config::default().nodelay(true),
345            noise::Config::new,
346            || {
347                let mut config = yamux::Config::default();
348                // These setters select libp2p-yamux's bounded 0.12 backend.
349                config.set_max_num_streams(4);
350                config.set_max_buffer_size(256 * 1024);
351                config.set_receive_window_size(256 * 1024);
352                config
353            },
354        )
355        .map_err(|e| Error::Invalid(format!("peer transport: {e}")))?
356        .with_dns()
357        .map_err(|e| Error::Invalid(format!("peer DNS resolver: {e}")))?
358        .with_behaviour(|_| Behaviour {
359            limits,
360            records,
361            identify,
362        })
363        .map_err(|e| Error::Invalid(format!("peer behaviour: {e}")))?
364        .with_swarm_config(|config| {
365            config
366                .with_notify_handler_buffer_size(NonZeroUsize::new(2).unwrap())
367                .with_per_connection_event_buffer_size(2)
368                .with_max_negotiating_inbound_streams(2)
369                .with_dial_concurrency_factor(NonZeroU8::new(1).unwrap())
370                .with_idle_connection_timeout(Duration::from_secs(20))
371        })
372        .with_connection_timeout(Duration::from_secs(5))
373        .build())
374}
375
376struct Control {
377    commands: async_mpsc::Receiver<Command>,
378    stop: Arc<AtomicBool>,
379    connected: Arc<AtomicU64>,
380    startup: Startup,
381    shared: Arc<Shared>,
382    announce: Vec<Multiaddr>,
383}
384
385/// Remote address per live connection, so gossip can tell loopback requesters.
386#[derive(Default)]
387struct Remotes(HashMap<ConnectionId, (Multiaddr, bool)>);
388
389impl Remotes {
390    fn loopback(&self, connection: ConnectionId) -> bool {
391        self.0
392            .get(&connection)
393            .is_some_and(|(address, _)| is_loopback(address))
394    }
395
396    fn outbound(&self, connection: ConnectionId) -> bool {
397        self.0
398            .get(&connection)
399            .is_some_and(|(_, outbound)| *outbound)
400    }
401}
402
403async fn run(
404    engine: Arc<Engine>,
405    identity: Keypair,
406    listen: Option<Multiaddr>,
407    control: Control,
408) -> Result<(), Error> {
409    let Control {
410        mut commands,
411        stop,
412        connected,
413        startup,
414        shared,
415        announce,
416    } = control;
417    let genesis = engine.status()?.chain_genesis;
418    let mut swarm = swarm(identity)?;
419    for address in announce {
420        swarm.add_external_address(address);
421    }
422    let mut ready = Some(startup);
423    let listening = listen.is_some();
424    if let Some(address) = listen {
425        swarm
426            .listen_on(address)
427            .map_err(|e| Error::Invalid(format!("peer listen: {e}")))?;
428    }
429    let (jobs, input) = mpsc::sync_channel::<ServeJob>(2);
430    let (served, mut responses) = async_mpsc::channel::<Served>(2);
431    let serving_stop = stop.clone();
432    let serving = std::thread::Builder::new()
433        .name("urma-peer-read".into())
434        .spawn(move || serve(&engine, genesis, &serving_stop, input, served))?;
435    if !listening {
436        let _ = ready.take().unwrap().send(Ok(None));
437    }
438    let mut pending = HashMap::<OutboundRequestId, Pending>::new();
439    let mut inbound = HashSet::new();
440    let mut remotes = Remotes::default();
441    let mut dials = Dials::default();
442    let mut rates = HashMap::<PeerId, Bucket>::new();
443    let mut requests = Bucket::new(16, 16);
444    let mut bytes = Bucket::new(64 * 1024 * 1024, 4 * 1024 * 1024);
445    let mut tick = tokio::time::interval(Duration::from_millis(50));
446    let started = Instant::now();
447    let result = loop {
448        if stop.load(Ordering::Acquire) {
449            break Ok(());
450        }
451        if ready.is_some() && started.elapsed() > Duration::from_secs(4) {
452            break Err(Error::Busy("peer listener startup deadline".into()));
453        }
454        tokio::select! {
455            _ = tick.tick() => {
456                // Keep the request tracked until libp2p reports completion. This
457                // bounds its internal pending queue even if a caller times out.
458                for value in pending.values_mut() {
459                    if Instant::now() >= value.deadline {
460                        value.answer.fail(Error::Busy("peer acquisition deadline".into()));
461                        let _ = swarm.disconnect_peer_id(value.peer);
462                    }
463                }
464                dials.expire();
465            }
466            command = commands.recv() => {
467                let command = match command {
468                    Some(Command::Fetch(fetch)) => fetch,
469                    // A Fetch queued for this peer since the rejection keeps its connection.
470                    Some(Command::Disconnect(peer)) => {
471                        if !pending.values().any(|p| p.peer == peer) { let _ = swarm.disconnect_peer_id(peer); }
472                        continue;
473                    }
474                    Some(Command::Dial(dial)) => { dials.start(&mut swarm, dial); continue; }
475                    None => break Ok(()),
476                };
477                if Instant::now() >= command.deadline || pending.len() >= MAX_PENDING
478                    || pending.values().any(|p| p.peer == command.peer) {
479                    command.answer.fail(Error::Busy("peer acquisition capacity or deadline".into()));
480                    continue;
481                }
482                let request = Request { version: VERSION, chain_genesis: genesis, kind: command.kind };
483                let id = swarm.behaviour_mut().records.send_request_with_addresses(
484                    &command.peer, request, command.addresses);
485                pending.insert(id, Pending { peer: command.peer, kind: command.kind, deadline: command.deadline, answer: command.answer });
486            }
487            response = responses.recv() => {
488                let Some(mut response) = response else { break Err(Error::Invalid("peer archive reader stopped".into())); };
489                if response.channel.is_open() {
490                    if !bytes.take(u64::try_from(response.bytes.saturating_sub(256)).unwrap_or(u64::MAX)) { response.response.result = Reply::Busy; }
491                    let _ = swarm.behaviour_mut().records.send_response(response.channel, response.response);
492                }
493            }
494            event = swarm.select_next_some() => match event {
495                SwarmEvent::NewListenAddr { mut address, .. } => {
496                    address.push(Protocol::P2p(*swarm.local_peer_id()));
497                    shared.listening(address.clone(), true);
498                    if let Some(ready) = ready.take() { let _ = ready.send(Ok(Some(address))); }
499                }
500                SwarmEvent::ExpiredListenAddr { address, .. } => shared.listening(address.with(Protocol::P2p(*swarm.local_peer_id())), false),
501                SwarmEvent::ListenerClosed { addresses, .. } => for address in addresses { shared.listening(address.with(Protocol::P2p(*swarm.local_peer_id())), false) },
502                SwarmEvent::ListenerError { error, .. } if ready.is_some() => break Err(Error::Io(error)),
503                SwarmEvent::ConnectionEstablished { peer_id, connection_id, endpoint, num_established, .. } => {
504                    remotes.0.insert(connection_id, (endpoint.get_remote_address().clone(), endpoint.is_dialer()));
505                    dials.established(connection_id, peer_id);
506                    shared.opened(peer_id, endpoint.is_dialer());
507                    if num_established.get() == 1 {
508                        connected.fetch_add(1, Ordering::Relaxed);
509                    }
510                }
511                SwarmEvent::ConnectionClosed { peer_id, connection_id, endpoint, num_established, .. } => {
512                    remotes.0.remove(&connection_id);
513                    shared.closed(peer_id, endpoint.is_dialer());
514                    if num_established == 0 {
515                        connected.fetch_sub(1, Ordering::Relaxed);
516                    }
517                }
518                SwarmEvent::OutgoingConnectionError { connection_id, peer_id, error } => {
519                    // Queued before request-response reports the same DialFailure.
520                    if let (Some(peer), DialError::Denied { cause }) = (peer_id, &error) {
521                        if let Some(LimitDenied(limit)) = cause.downcast_ref::<LimitDenied>() { release_denied(&mut pending, peer, *limit); }
522                    }
523                    dials.failed(connection_id, &error);
524                }
525                SwarmEvent::Behaviour(BehaviourEvent::Identify(identify::Event::Received { connection_id, peer_id, info })) => {
526                    // Inbound peers could steer our address; only dialed peers witness it.
527                    if remotes.outbound(connection_id) {
528                        if let Some(confirmed) = shared.observe(info.observed_addr, peer_id) { swarm.add_external_address(confirmed); }
529                    }
530                    shared.identified(peer_id, info.listen_addrs);
531                }
532                SwarmEvent::Behaviour(BehaviourEvent::Records(event)) => match event {
533                    request_response::Event::Message { peer, connection_id, message: Message::Request { request_id, request, channel } } => {
534                        // Every request, including refusals, consumes rate and
535                        // response allowance. Over-rate requests get an explicit
536                        // Busy; only an exhausted byte allowance drops the channel.
537                        if request.version == VERSION && request.chain_genesis == genesis { shared.on_chain(peer); }
538                        let over_rate = !requests.take(1) || !peer_rate(&mut rates, peer);
539                        if !bytes.take(256) {
540                            continue;
541                        }
542                        let refusal = if over_rate { Some(Reply::Busy) }
543                            else if request.version != VERSION { Some(Reply::UnsupportedVersion) }
544                            else if request.chain_genesis != genesis { Some(Reply::WrongNetwork) }
545                            else if let Kind::Peers { limit } = request.kind {
546                                // Gossip is answered from memory, never by the archive reader.
547                                Some(shared.serve(peer, remotes.loopback(connection_id), limit))
548                            }
549                            else if inbound.len() >= 2 || inbound.iter().any(|(p, _)| *p == peer)
550                                { Some(Reply::Busy) }
551                            else { None };
552                        if let Some(result) = refusal {
553                            let _ = swarm.behaviour_mut().records.send_response(channel, Response { version: VERSION, chain_genesis: genesis, kind: request.kind, result });
554                        } else {
555                            match jobs.try_send(ServeJob { deadline: Instant::now() + REQUEST_TIMEOUT, request, channel }) {
556                                Ok(()) => { inbound.insert((peer, request_id)); }
557                                Err(_) => { /* Dropped response channel refuses the request. */ }
558                            }
559                        }
560                    }
561                    request_response::Event::Message { peer, message: Message::Response { request_id, response }, .. } => {
562                        if response.version == VERSION && response.chain_genesis == genesis { shared.on_chain(peer); }
563                        if let Some(value) = pending.remove(&request_id) {
564                            value.answer.deliver(genesis, peer, &value, response);
565                        }
566                    }
567                    request_response::Event::OutboundFailure { request_id, error, .. } => {
568                        if let Some(value) = pending.remove(&request_id) {
569                            value.answer.fail(Error::Transport(error.to_string()));
570                        }
571                    }
572                    request_response::Event::InboundFailure { peer, request_id, .. }
573                    | request_response::Event::ResponseSent { peer, request_id, .. } => { inbound.remove(&(peer, request_id)); }
574                },
575                _ => (),
576            }
577        }
578    };
579    stop.store(true, Ordering::Release);
580    for (_, value) in pending {
581        value
582            .answer
583            .fail(Error::Busy("peer network stopped".into()));
584    }
585    dials.close();
586    // Drop channels before joining: the archive reader must not wait for a
587    // consumer which is already shutting down. No socket belongs to that reader.
588    drop(swarm);
589    drop(jobs);
590    drop(responses);
591    if serving.join().is_err() {
592        return Err(Error::Invalid("peer archive reader panicked".into()));
593    }
594    result
595}
596
597/// The bounded archive reader: one local lookup per job, sized before sending.
598fn serve(
599    engine: &Engine,
600    genesis: BlockHash,
601    stop: &AtomicBool,
602    input: mpsc::Receiver<ServeJob>,
603    served: async_mpsc::Sender<Served>,
604) {
605    while let Ok(job) = input.recv() {
606        if stop.load(Ordering::Acquire) {
607            break;
608        }
609        if !job.channel.is_open() || Instant::now() >= job.deadline {
610            continue;
611        }
612        let result = lookup(engine, &job.request);
613        if stop.load(Ordering::Acquire) || !job.channel.is_open() || Instant::now() >= job.deadline
614        {
615            continue;
616        }
617        let mut response = Response {
618            version: VERSION,
619            chain_genesis: genesis,
620            kind: job.request.kind,
621            result,
622        };
623        let mut count = Count(0);
624        if serde_json::to_writer(&mut count, &response).is_err() {
625            response.result = Reply::TooLarge;
626            count.0 = 256;
627        }
628        let reply = Served {
629            channel: job.channel,
630            response,
631            bytes: count.0,
632        };
633        if served.blocking_send(reply).is_err() {
634            break;
635        }
636    }
637}
638
639/// A dial refused by our own limits is local capacity: the waiting requests
640/// fail with which limit refused it, never blaming the peer.
641fn release_denied(
642    pending: &mut HashMap<OutboundRequestId, Pending>,
643    peer: PeerId,
644    limit: DialLimit,
645) {
646    pending.retain(|_, value| {
647        if value.peer != peer {
648            return true;
649        }
650        value.answer.fail(Error::DialDenied(limit));
651        false
652    });
653}
654
655pub(crate) fn bind_response(
656    genesis: BlockHash,
657    peer: PeerId,
658    pending: &Pending,
659    response: &Response,
660) -> Result<(), Error> {
661    if Instant::now() >= pending.deadline {
662        return Err(Error::Busy("peer acquisition deadline".into()));
663    }
664    if peer != pending.peer
665        || response.version != VERSION
666        || response.chain_genesis != genesis
667        || response.kind != pending.kind
668    {
669        return Err(Error::Invalid(
670            "peer response identity, network or version mismatch".into(),
671        ));
672    }
673    Ok(())
674}
675
676fn validate_response(
677    genesis: BlockHash,
678    peer: PeerId,
679    pending: &Pending,
680    response: Response,
681) -> Result<Option<RecordPackage>, Error> {
682    bind_response(genesis, peer, pending, &response)?;
683    match response.result {
684        Reply::Record(package) if package.chain_genesis == genesis => Ok(Some(*package)),
685        Reply::Missing => Ok(None),
686        Reply::Busy => Err(Error::PeerBusy),
687        _ => Err(Error::Invalid(
688            "peer refused record or returned wrong network".into(),
689        )),
690    }
691}
692
693#[cfg(test)]
694mod tests {
695    use super::*;
696    use bitcoin::hashes::Hash;
697
698    #[test]
699    fn public_response_metadata_is_bound_but_is_not_record_verification() {
700        let genesis = BlockHash::all_zeros();
701        let txid = Txid::all_zeros();
702        let peer = Keypair::generate_ed25519().public().to_peer_id();
703        let other = Keypair::generate_ed25519().public().to_peer_id();
704        let (answer, _receiver) = mpsc::sync_channel(1);
705        let pending = Pending {
706            peer,
707            kind: Kind::Record { txid },
708            deadline: Instant::now() + CALL_TIMEOUT,
709            answer: Answer::Record(answer),
710        };
711        let response = Response {
712            version: VERSION,
713            chain_genesis: genesis,
714            kind: Kind::Record { txid },
715            result: Reply::Missing,
716        };
717        assert!(
718            validate_response(genesis, peer, &pending, response.clone())
719                .unwrap()
720                .is_none()
721        );
722        assert!(validate_response(genesis, other, &pending, response.clone()).is_err());
723        let mut bad = response.clone();
724        bad.version += 1;
725        assert!(validate_response(genesis, peer, &pending, bad).is_err());
726        let mut bad = response.clone();
727        bad.chain_genesis = BlockHash::from_byte_array([1; 32]);
728        assert!(validate_response(genesis, peer, &pending, bad).is_err());
729        let mut bad = response.clone();
730        bad.kind = Kind::Record {
731            txid: Txid::from_byte_array([1; 32]),
732        };
733        assert!(validate_response(genesis, peer, &pending, bad).is_err());
734        let expired = Pending {
735            deadline: Instant::now(),
736            ..pending
737        };
738        assert!(matches!(
739            validate_response(genesis, peer, &expired, response),
740            Err(Error::Busy(_))
741        ));
742    }
743
744    #[test]
745    fn listener_accepts_only_explicit_ip_tcp_without_implicit_resolution() {
746        for value in [
747            "/ip4/127.0.0.1/tcp/0",
748            "/ip6/::1/tcp/0",
749            "/ip4/0.0.0.0/tcp/28332",
750        ] {
751            validate_listener(&value.parse().unwrap()).unwrap();
752        }
753        for value in [
754            "/dns4/localhost/tcp/1",
755            "/ip4/127.0.0.1/udp/1",
756            "/ip4/127.0.0.1/tcp/1/ws",
757        ] {
758            assert!(validate_listener(&value.parse().unwrap()).is_err());
759        }
760    }
761}
urma_node/peer.rs

urma_node/
peer.rs
