snarkos_node/bootstrap_client/
network.rs

1// Copyright (c) 2019-2026 Provable Inc.
2// This file is part of the snarkOS library.
3
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License at:
7
8// http://www.apache.org/licenses/LICENSE-2.0
9
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16use crate::{
17    BootstrapClient,
18    bft::{
19        MAX_VALIDATORS_TO_SEND,
20        events::{self, Event},
21    },
22    bootstrap_client::codec::BootstrapClientCodec,
23    network::{ConnectionMode, NodeType, Peer, PeerPoolHandling, Resolver},
24    router::{
25        MAX_PEERS_TO_SEND,
26        messages::{self, Message},
27    },
28    tcp::{ConnectionSide, P2P, Tcp, connections::DisconnectOrigin, protocols::*},
29};
30use snarkvm::prelude::Network;
31
32use indexmap::IndexMap;
33#[cfg(feature = "locktick")]
34use locktick::parking_lot::RwLock;
35#[cfg(not(feature = "locktick"))]
36use parking_lot::RwLock;
37use std::{collections::HashMap, io, net::SocketAddr};
38use tokio::time::sleep;
39use tokio_util::codec::Decoder;
40
41impl<N: Network> P2P for BootstrapClient<N> {
42    fn tcp(&self) -> &Tcp {
43        &self.tcp
44    }
45}
46
47impl<N: Network> PeerPoolHandling<N> for BootstrapClient<N> {
48    const MAXIMUM_POOL_SIZE: usize = 10_000;
49    const OWNER: &'static str = "[Network]";
50    const PEER_SLASHING_COUNT: usize = 200;
51
52    fn is_dev(&self) -> bool {
53        self.dev.is_some()
54    }
55
56    fn trusted_peers_only(&self) -> bool {
57        false
58    }
59
60    fn node_type(&self) -> NodeType {
61        NodeType::BootstrapClient
62    }
63
64    fn peer_pool(&self) -> &RwLock<HashMap<SocketAddr, Peer<N>>> {
65        &self.peer_pool
66    }
67
68    fn resolver(&self) -> &RwLock<Resolver<N>> {
69        &self.resolver
70    }
71}
72
73/// The bootstrap client can handle both validator and non-validator messages.
74#[derive(Debug)]
75pub enum MessageOrEvent<N: Network> {
76    Message(Message<N>),
77    Event(Event<N>),
78}
79
80#[async_trait]
81impl<N: Network> OnConnect for BootstrapClient<N> {
82    async fn on_connect(&self, peer_addr: SocketAddr) {
83        // If the peer is connected in validator (Gateway) mode, save it to the collection
84        // of known validators.
85        if let Some(listener_addr) = self.resolve_to_listener(peer_addr)
86            && let Some(peer) = self.get_connected_peer(listener_addr)
87            && peer.node_type == NodeType::Validator
88        {
89            self.known_validators.write().insert(listener_addr, (peer.aleo_addr, peer.connection_mode));
90        }
91        // The peers should only ask us for the peer list; spawn a task that will
92        // terminate the connection after a while.
93        let tcp = self.tcp().clone();
94        tokio::spawn(async move {
95            sleep(Self::CONNECTION_LIFETIME).await;
96            tcp.disconnect(peer_addr).await;
97        });
98    }
99}
100
101#[async_trait]
102impl<N: Network> Disconnect for BootstrapClient<N> {
103    /// Any extra operations to be performed during a disconnect.
104    async fn handle_disconnect(&self, peer_addr: SocketAddr, origin: DisconnectOrigin) {
105        debug!("Physically disconnecting from {peer_addr}; origin: {origin:?}");
106
107        if let Some(listener_addr) = self.resolve_to_listener(peer_addr) {
108            self.downgrade_peer_to_candidate(listener_addr);
109        }
110    }
111}
112
113#[async_trait]
114impl<N: Network> Reading for BootstrapClient<N> {
115    type Codec = BootstrapClientCodec<N>;
116    type Message = <BootstrapClientCodec<N> as Decoder>::Item;
117
118    /// Creates a [`Decoder`] used to interpret messages from the network.
119    /// The `side` param indicates the connection side **from the node's perspective**.
120    fn codec(&self, _peer_addr: SocketAddr, _side: ConnectionSide) -> Self::Codec {
121        Default::default()
122    }
123
124    /// Processes a message received from the network.
125    async fn process_message(&self, peer_addr: SocketAddr, message: Self::Message) -> io::Result<()> {
126        // Identify the connected peer.
127        let Some(listener_addr) = self.resolve_to_listener(peer_addr) else {
128            // Already disconnecting, ignore.
129            return Ok(());
130        };
131
132        // Handle the right peer request.
133        match message {
134            MessageOrEvent::Message(Message::PeerRequest(_)) => {
135                debug!("Received a PeerRequest from '{listener_addr}'");
136                let mut peers = self.get_candidate_peers();
137
138                // In order to filter out validators properly, we'll need the
139                // peer's node type and the list of validators.
140                let Some(peer) = self.get_connected_peer(listener_addr) else {
141                    return Ok(());
142                };
143                let validators = self.get_validator_addrs().await;
144
145                if peer.node_type == NodeType::Validator {
146                    // Filter out Gateway addresses.
147                    peers.retain(|peer| {
148                        validators
149                            .get(&peer.listener_addr)
150                            .map(|(_, connection_mode)| *connection_mode != ConnectionMode::Gateway)
151                            .unwrap_or(true)
152                    });
153                } else {
154                    // Filter out all validator addresses.
155                    peers.retain(|peer| !validators.contains_key(&peer.listener_addr));
156                }
157                peers.truncate(MAX_PEERS_TO_SEND);
158                let peers = peers.into_iter().map(|peer| (peer.listener_addr, None)).collect::<Vec<_>>();
159
160                debug!("Sending {} peer address(es) to '{listener_addr}'", peers.len());
161                let msg = MessageOrEvent::Message(Message::PeerResponse(messages::PeerResponse { peers }));
162                if let Err(err) = self.unicast(peer_addr, msg)?.await {
163                    warn!("Couldn't deliver a peer list to '{listener_addr}': {err}; disconnecting");
164                } else {
165                    debug!("Disconnecting from '{listener_addr}' - peers provided");
166                }
167
168                self.tcp().disconnect(peer_addr).await;
169            }
170            MessageOrEvent::Event(Event::ValidatorsRequest(_)) => {
171                debug!("Received a ValidatorsRequest from '{listener_addr}'");
172
173                // Procure a list of applicable validator addresses.
174                let validators = self.get_validator_addrs().await;
175                let validators = validators
176                    .into_iter()
177                    .filter_map(|(listener_addr, (aleo_addr, connection_mode))| {
178                        // Only pick addresses connected in Gateway mode.
179                        (connection_mode == ConnectionMode::Gateway).then_some((listener_addr, aleo_addr))
180                    })
181                    .take(MAX_VALIDATORS_TO_SEND)
182                    .collect::<IndexMap<_, _>>();
183
184                debug!("Sending {} validator address(es) to '{listener_addr}'", validators.len());
185                let msg = MessageOrEvent::Event(Event::ValidatorsResponse(events::ValidatorsResponse { validators }));
186                if let Err(err) = self.unicast(peer_addr, msg)?.await {
187                    warn!("Couldn't deliver a peer list to '{listener_addr}': {err}; disconnecting");
188                } else {
189                    debug!("Disconnecting from '{listener_addr}' - peers provided");
190                }
191
192                self.tcp().disconnect(peer_addr).await;
193            }
194            msg => {
195                let name = match msg {
196                    MessageOrEvent::Message(msg) => msg.name(),
197                    MessageOrEvent::Event(msg) => msg.name(),
198                };
199                trace!("Ignoring an unhandled message ({name}) from {listener_addr}");
200            }
201        }
202
203        Ok(())
204    }
205}
206
207#[async_trait]
208impl<N: Network> Writing for BootstrapClient<N> {
209    type Codec = BootstrapClientCodec<N>;
210    type Message = MessageOrEvent<N>;
211
212    /// Creates an [`Encoder`] used to write the outbound messages to the target stream.
213    /// The `side` parameter indicates the connection side **from the node's perspective**.
214    fn codec(&self, _addr: SocketAddr, _side: ConnectionSide) -> Self::Codec {
215        Default::default()
216    }
217}
snarkos_node/bootstrap_client/network.rs

snarkos_node/bootstrap_client/
network.rs
