snarkos_node/prover/
mod.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
16mod router;
17
18use crate::traits::NodeInterface;
19use snarkos_account::Account;
20use snarkos_node_bft::ledger_service::ProverLedgerService;
21use snarkos_node_router::{
22    Heartbeat,
23    Inbound,
24    Outbound,
25    Router,
26    Routing,
27    messages::{Message, NodeType, UnconfirmedSolution},
28};
29use snarkos_node_sync::{BlockSync, BlockSyncMode};
30use snarkos_node_tcp::{
31    P2P,
32    protocols::{Disconnect, Handshake, OnConnect, Reading, Writing},
33};
34use snarkvm::{
35    ledger::narwhal::Data,
36    prelude::{
37        Network,
38        block::{Block, Header},
39        puzzle::{Puzzle, Solution},
40        store::ConsensusStorage,
41    },
42    synthesizer::VM,
43};
44
45use aleo_std::StorageMode;
46use anyhow::Result;
47use colored::Colorize;
48use core::{marker::PhantomData, time::Duration};
49#[cfg(feature = "locktick")]
50use locktick::parking_lot::{Mutex, RwLock};
51#[cfg(not(feature = "locktick"))]
52use parking_lot::{Mutex, RwLock};
53use rand::{CryptoRng, Rng, rngs::OsRng};
54use snarkos_node_bft::helpers::fmt_id;
55use std::{
56    net::SocketAddr,
57    sync::{
58        Arc,
59        atomic::{AtomicBool, AtomicU8, Ordering},
60    },
61};
62use tokio::task::JoinHandle;
63
64/// A prover is a light node, capable of producing proofs for consensus.
65#[derive(Clone)]
66pub struct Prover<N: Network, C: ConsensusStorage<N>> {
67    /// The router of the node.
68    router: Router<N>,
69    /// The sync module.
70    sync: Arc<BlockSync<N>>,
71    /// The genesis block.
72    genesis: Block<N>,
73    /// The puzzle.
74    puzzle: Puzzle<N>,
75    /// The latest epoch hash.
76    latest_epoch_hash: Arc<RwLock<Option<N::BlockHash>>>,
77    /// The latest block header.
78    latest_block_header: Arc<RwLock<Option<Header<N>>>>,
79    /// The number of puzzle instances.
80    puzzle_instances: Arc<AtomicU8>,
81    /// The maximum number of puzzle instances.
82    max_puzzle_instances: u8,
83    /// The spawned handles.
84    handles: Arc<Mutex<Vec<JoinHandle<()>>>>,
85    /// The shutdown signal.
86    shutdown: Arc<AtomicBool>,
87    /// PhantomData.
88    _phantom: PhantomData<C>,
89}
90
91impl<N: Network, C: ConsensusStorage<N>> Prover<N, C> {
92    /// Initializes a new prover node.
93    pub async fn new(
94        node_ip: SocketAddr,
95        account: Account<N>,
96        trusted_peers: &[SocketAddr],
97        genesis: Block<N>,
98        storage_mode: StorageMode,
99        shutdown: Arc<AtomicBool>,
100    ) -> Result<Self> {
101        // Initialize the signal handler.
102        let signal_node = Self::handle_signals(shutdown.clone());
103
104        // Initialize the ledger service.
105        let ledger_service = Arc::new(ProverLedgerService::new());
106        // Determine if the prover should allow external peers.
107        let allow_external_peers = true;
108        // Determine if the prover should rotate external peers.
109        let rotate_external_peers = false;
110
111        // Initialize the node router.
112        let router = Router::new(
113            node_ip,
114            NodeType::Prover,
115            account,
116            ledger_service.clone(),
117            trusted_peers,
118            Self::MAXIMUM_NUMBER_OF_PEERS as u16,
119            rotate_external_peers,
120            allow_external_peers,
121            matches!(storage_mode, StorageMode::Development(_)),
122        )
123        .await?;
124
125        // Initialize the sync module.
126        let sync = BlockSync::new(BlockSyncMode::Router, ledger_service.clone(), router.tcp().clone());
127
128        // Compute the maximum number of puzzle instances.
129        let max_puzzle_instances = num_cpus::get().saturating_sub(2).clamp(1, 6);
130        // Initialize the node.
131        let node = Self {
132            router,
133            sync: Arc::new(sync),
134            genesis,
135            puzzle: VM::<N, C>::new_puzzle()?,
136            latest_epoch_hash: Default::default(),
137            latest_block_header: Default::default(),
138            puzzle_instances: Default::default(),
139            max_puzzle_instances: u8::try_from(max_puzzle_instances)?,
140            handles: Default::default(),
141            shutdown,
142            _phantom: Default::default(),
143        };
144        // Initialize the routing.
145        node.initialize_routing().await;
146        // Initialize the puzzle.
147        node.initialize_puzzle().await;
148        // Initialize the notification message loop.
149        node.handles.lock().push(crate::start_notification_message_loop());
150        // Pass the node to the signal handler.
151        let _ = signal_node.set(node.clone());
152        // Return the node.
153        Ok(node)
154    }
155}
156
157#[async_trait]
158impl<N: Network, C: ConsensusStorage<N>> NodeInterface<N> for Prover<N, C> {
159    /// Shuts down the node.
160    async fn shut_down(&self) {
161        info!("Shutting down...");
162
163        // Shut down the puzzle.
164        debug!("Shutting down the puzzle...");
165        self.shutdown.store(true, Ordering::Release);
166
167        // Abort the tasks.
168        debug!("Shutting down the prover...");
169        self.handles.lock().iter().for_each(|handle| handle.abort());
170
171        // Shut down the router.
172        self.router.shut_down().await;
173
174        info!("Node has shut down.");
175    }
176}
177
178impl<N: Network, C: ConsensusStorage<N>> Prover<N, C> {
179    /// Initialize a new instance of the puzzle.
180    async fn initialize_puzzle(&self) {
181        for _ in 0..self.max_puzzle_instances {
182            let prover = self.clone();
183            self.handles.lock().push(tokio::spawn(async move {
184                prover.puzzle_loop().await;
185            }));
186        }
187    }
188
189    /// Executes an instance of the puzzle.
190    async fn puzzle_loop(&self) {
191        loop {
192            // If the node is not connected to any peers, then skip this iteration.
193            if self.router.number_of_connected_peers() == 0 {
194                debug!("Skipping an iteration of the puzzle (no connected peers)");
195                tokio::time::sleep(Duration::from_secs(N::ANCHOR_TIME as u64)).await;
196                continue;
197            }
198
199            // If the number of instances of the puzzle exceeds the maximum, then skip this iteration.
200            if self.num_puzzle_instances() > self.max_puzzle_instances {
201                // Sleep for a brief period of time.
202                tokio::time::sleep(Duration::from_millis(500)).await;
203                continue;
204            }
205
206            // Read the latest epoch hash.
207            let latest_epoch_hash = *self.latest_epoch_hash.read();
208            // Read the latest state.
209            let latest_state = self
210                .latest_block_header
211                .read()
212                .as_ref()
213                .map(|header| (header.coinbase_target(), header.proof_target()));
214
215            // If the latest epoch hash and latest state exists, then proceed to generate a solution.
216            if let (Some(epoch_hash), Some((coinbase_target, proof_target))) = (latest_epoch_hash, latest_state) {
217                // Execute the puzzle.
218                let prover = self.clone();
219                let result = tokio::task::spawn_blocking(move || {
220                    prover.puzzle_iteration(epoch_hash, coinbase_target, proof_target, &mut OsRng)
221                })
222                .await;
223
224                // If the prover found a solution, then broadcast it.
225                if let Ok(Some((solution_target, solution))) = result {
226                    info!("Found a Solution '{}' (Proof Target {solution_target})", solution.id());
227                    // Broadcast the solution.
228                    self.broadcast_solution(solution);
229                }
230            } else {
231                // Otherwise, sleep for a brief period of time, to await for puzzle state.
232                tokio::time::sleep(Duration::from_secs(1)).await;
233            }
234
235            // If the Ctrl-C handler registered the signal, stop the prover.
236            if self.shutdown.load(Ordering::Acquire) {
237                debug!("Shutting down the puzzle...");
238                break;
239            }
240        }
241    }
242
243    /// Performs one iteration of the puzzle.
244    fn puzzle_iteration<R: Rng + CryptoRng>(
245        &self,
246        epoch_hash: N::BlockHash,
247        coinbase_target: u64,
248        proof_target: u64,
249        rng: &mut R,
250    ) -> Option<(u64, Solution<N>)> {
251        // Increment the puzzle instances.
252        self.increment_puzzle_instances();
253
254        debug!(
255            "Proving 'Puzzle' for Epoch '{}' {}",
256            fmt_id(epoch_hash),
257            format!("(Coinbase Target {coinbase_target}, Proof Target {proof_target})").dimmed()
258        );
259
260        // Compute the solution.
261        let result =
262            self.puzzle.prove(epoch_hash, self.address(), rng.gen(), Some(proof_target)).ok().and_then(|solution| {
263                self.puzzle.get_proof_target(&solution).ok().map(|solution_target| (solution_target, solution))
264            });
265
266        // Decrement the puzzle instances.
267        self.decrement_puzzle_instances();
268        // Return the result.
269        result
270    }
271
272    /// Broadcasts the solution to the network.
273    fn broadcast_solution(&self, solution: Solution<N>) {
274        // Prepare the unconfirmed solution message.
275        let message = Message::UnconfirmedSolution(UnconfirmedSolution {
276            solution_id: solution.id(),
277            solution: Data::Object(solution),
278        });
279        // Propagate the "UnconfirmedSolution".
280        self.propagate(message, &[]);
281    }
282
283    /// Returns the current number of puzzle instances.
284    fn num_puzzle_instances(&self) -> u8 {
285        self.puzzle_instances.load(Ordering::Relaxed)
286    }
287
288    /// Increments the number of puzzle instances.
289    fn increment_puzzle_instances(&self) {
290        self.puzzle_instances.fetch_add(1, Ordering::Relaxed);
291        #[cfg(debug_assertions)]
292        trace!("Number of Instances - {}", self.num_puzzle_instances());
293    }
294
295    /// Decrements the number of puzzle instances.
296    fn decrement_puzzle_instances(&self) {
297        self.puzzle_instances.fetch_sub(1, Ordering::Relaxed);
298        #[cfg(debug_assertions)]
299        trace!("Number of Instances - {}", self.num_puzzle_instances());
300    }
301}
snarkos_node/prover/mod.rs

snarkos_node/prover/
mod.rs
