snarkos_node/validator/
router.rs

1// Copyright (c) 2019-2025 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 super::*;
17use snarkos_node_router::messages::{
18    BlockRequest,
19    BlockResponse,
20    DataBlocks,
21    DisconnectReason,
22    Message,
23    MessageCodec,
24    Ping,
25    Pong,
26    UnconfirmedTransaction,
27};
28use snarkos_node_tcp::{Connection, ConnectionSide, Tcp};
29use snarkvm::{
30    ledger::narwhal::Data,
31    prelude::{Network, block::Transaction, error},
32};
33
34use std::{io, net::SocketAddr};
35
36impl<N: Network, C: ConsensusStorage<N>> P2P for Validator<N, C> {
37    /// Returns a reference to the TCP instance.
38    fn tcp(&self) -> &Tcp {
39        self.router.tcp()
40    }
41}
42
43#[async_trait]
44impl<N: Network, C: ConsensusStorage<N>> Handshake for Validator<N, C> {
45    /// Performs the handshake protocol.
46    async fn perform_handshake(&self, mut connection: Connection) -> io::Result<Connection> {
47        // Perform the handshake.
48        let peer_addr = connection.addr();
49        let conn_side = connection.side();
50        let stream = self.borrow_stream(&mut connection);
51        let genesis_header = self.ledger.get_header(0).map_err(|e| error(format!("{e}")))?;
52        let restrictions_id = self.ledger.vm().restrictions().restrictions_id();
53        self.router.handshake(peer_addr, stream, conn_side, genesis_header, restrictions_id).await?;
54
55        Ok(connection)
56    }
57}
58
59#[async_trait]
60impl<N: Network, C: ConsensusStorage<N>> OnConnect for Validator<N, C>
61where
62    Self: Outbound<N>,
63{
64    async fn on_connect(&self, peer_addr: SocketAddr) {
65        // Resolve the peer address to the listener address.
66        let Some(peer_ip) = self.router.resolve_to_listener(&peer_addr) else { return };
67        // Send the first `Ping` message to the peer.
68        self.ping.on_peer_connected(peer_ip);
69    }
70}
71
72#[async_trait]
73impl<N: Network, C: ConsensusStorage<N>> Disconnect for Validator<N, C> {
74    /// Any extra operations to be performed during a disconnect.
75    async fn handle_disconnect(&self, peer_addr: SocketAddr) {
76        if let Some(peer_ip) = self.router.resolve_to_listener(&peer_addr) {
77            self.sync.remove_peer(&peer_ip);
78            self.router.remove_connected_peer(peer_ip);
79        }
80    }
81}
82
83#[async_trait]
84impl<N: Network, C: ConsensusStorage<N>> Reading for Validator<N, C> {
85    type Codec = MessageCodec<N>;
86    type Message = Message<N>;
87
88    /// Creates a [`Decoder`] used to interpret messages from the network.
89    /// The `side` param indicates the connection side **from the node's perspective**.
90    fn codec(&self, _peer_addr: SocketAddr, _side: ConnectionSide) -> Self::Codec {
91        Default::default()
92    }
93
94    /// Processes a message received from the network.
95    async fn process_message(&self, peer_addr: SocketAddr, message: Self::Message) -> io::Result<()> {
96        let clone = self.clone();
97        if matches!(message, Message::BlockRequest(_) | Message::BlockResponse(_)) {
98            // Handle BlockRequest and BlockResponse messages in a separate task to not block the
99            // inbound queue.
100            tokio::spawn(async move {
101                clone.process_message_inner(peer_addr, message).await;
102            });
103        } else {
104            self.process_message_inner(peer_addr, message).await;
105        }
106        Ok(())
107    }
108}
109
110impl<N: Network, C: ConsensusStorage<N>> Validator<N, C> {
111    async fn process_message_inner(
112        &self,
113        peer_addr: SocketAddr,
114        message: <Validator<N, C> as snarkos_node_tcp::protocols::Reading>::Message,
115    ) {
116        // Process the message. Disconnect if the peer violated the protocol.
117        if let Err(error) = self.inbound(peer_addr, message).await {
118            warn!("Failed to process inbound message from '{peer_addr}' - {error}");
119            if let Some(peer_ip) = self.router().resolve_to_listener(&peer_addr) {
120                warn!("Disconnecting from '{peer_ip}' for protocol violation");
121                self.router().send(peer_ip, Message::Disconnect(DisconnectReason::ProtocolViolation.into()));
122                // Disconnect from this peer.
123                self.router().disconnect(peer_ip);
124            }
125        }
126    }
127}
128
129#[async_trait]
130impl<N: Network, C: ConsensusStorage<N>> Routing<N> for Validator<N, C> {}
131
132impl<N: Network, C: ConsensusStorage<N>> Heartbeat<N> for Validator<N, C> {}
133
134impl<N: Network, C: ConsensusStorage<N>> Outbound<N> for Validator<N, C> {
135    /// Returns a reference to the router.
136    fn router(&self) -> &Router<N> {
137        &self.router
138    }
139
140    /// Returns `true` if the node is synced up to the latest block (within the given tolerance).
141    fn is_block_synced(&self) -> bool {
142        self.sync.is_block_synced()
143    }
144
145    /// Returns the number of blocks this node is behind the greatest peer height,
146    /// or `None` if not connected to peers yet.
147    fn num_blocks_behind(&self) -> Option<u32> {
148        self.sync.num_blocks_behind()
149    }
150}
151
152#[async_trait]
153impl<N: Network, C: ConsensusStorage<N>> Inbound<N> for Validator<N, C> {
154    /// Returns `true` if the message version is valid.
155    fn is_valid_message_version(&self, message_version: u32) -> bool {
156        self.router().is_valid_message_version(message_version)
157    }
158
159    /// Retrieves the blocks within the block request range, and returns the block response to the peer.
160    fn block_request(&self, peer_ip: SocketAddr, message: BlockRequest) -> bool {
161        let BlockRequest { start_height, end_height } = &message;
162
163        // Retrieve the blocks within the requested range.
164        let blocks = match self.ledger.get_blocks(*start_height..*end_height) {
165            Ok(blocks) => Data::Object(DataBlocks(blocks)),
166            Err(error) => {
167                error!("Failed to retrieve blocks {start_height} to {end_height} from the ledger - {error}");
168                return false;
169            }
170        };
171        // Send the `BlockResponse` message to the peer.
172        self.router().send(peer_ip, Message::BlockResponse(BlockResponse { request: message, blocks }));
173        true
174    }
175
176    /// Handles a `BlockResponse` message.
177    fn block_response(&self, peer_ip: SocketAddr, _blocks: Vec<Block<N>>) -> bool {
178        warn!("Received a block response through P2P, not BFT, from {peer_ip}");
179        false
180    }
181
182    /// Processes a ping message from a client (or prover) and sends back a `Pong` message.
183    fn ping(&self, peer_ip: SocketAddr, _message: Ping<N>) -> bool {
184        // In gateway/validator mode, we do not need to process client block locators.
185        // Instead, locators are fetched from other validators in `Gateway` using `PrimaryPing` messages.
186
187        // Send a `Pong` message to the peer.
188        self.router().send(peer_ip, Message::Pong(Pong { is_fork: Some(false) }));
189        true
190    }
191
192    /// Process a Pong message (response to a Ping).
193    fn pong(&self, peer_ip: SocketAddr, _message: Pong) -> bool {
194        self.ping.on_pong_received(peer_ip);
195        true
196    }
197
198    /// Retrieves the latest epoch hash and latest block header, and returns the puzzle response to the peer.
199    fn puzzle_request(&self, peer_ip: SocketAddr) -> bool {
200        // Retrieve the latest epoch hash.
201        let epoch_hash = match self.ledger.latest_epoch_hash() {
202            Ok(epoch_hash) => epoch_hash,
203            Err(error) => {
204                error!("Failed to prepare a puzzle request for '{peer_ip}': {error}");
205                return false;
206            }
207        };
208        // Retrieve the latest block header.
209        let block_header = Data::Object(self.ledger.latest_header());
210        // Send the `PuzzleResponse` message to the peer.
211        self.router().send(peer_ip, Message::PuzzleResponse(PuzzleResponse { epoch_hash, block_header }));
212        true
213    }
214
215    /// Disconnects on receipt of a `PuzzleResponse` message.
216    fn puzzle_response(&self, peer_ip: SocketAddr, _epoch_hash: N::BlockHash, _header: Header<N>) -> bool {
217        debug!("Disconnecting '{peer_ip}' for the following reason - {:?}", DisconnectReason::ProtocolViolation);
218        false
219    }
220
221    /// Propagates the unconfirmed solution to all connected validators.
222    async fn unconfirmed_solution(
223        &self,
224        peer_ip: SocketAddr,
225        serialized: UnconfirmedSolution<N>,
226        solution: Solution<N>,
227    ) -> bool {
228        // Add the unconfirmed solution to the memory pool.
229        if let Err(error) = self.consensus.add_unconfirmed_solution(solution).await {
230            trace!("[UnconfirmedSolution] {error}");
231            return true; // Maintain the connection.
232        }
233        let message = Message::UnconfirmedSolution(serialized);
234        // Propagate the "UnconfirmedSolution" to the connected validators.
235        self.propagate_to_validators(message, &[peer_ip]);
236        true
237    }
238
239    /// Handles an `UnconfirmedTransaction` message.
240    async fn unconfirmed_transaction(
241        &self,
242        peer_ip: SocketAddr,
243        serialized: UnconfirmedTransaction<N>,
244        transaction: Transaction<N>,
245    ) -> bool {
246        // Add the unconfirmed transaction to the memory pool.
247        if let Err(error) = self.consensus.add_unconfirmed_transaction(transaction).await {
248            trace!("[UnconfirmedTransaction] {error}");
249            return true; // Maintain the connection.
250        }
251        let message = Message::UnconfirmedTransaction(serialized);
252        // Propagate the "UnconfirmedTransaction" to the connected validators.
253        self.propagate_to_validators(message, &[peer_ip]);
254        true
255    }
256}
snarkos_node/validator/router.rs

snarkos_node/validator/
router.rs
