25404138305-1

Committed 05 May 2026 09:47PM UTC coverage: 85.69% (-0.02%) from 85.712%

Build # 25404138305-1

Build Type

Pull #7169

github

Specific Base 497ffd

Committed by

web-flow

Commit Message

Merge 35db1183d into 53ffba0ab

Pull Request Pull Request #7169: Feat: add defensive memory allocation for miners/signers

Coverage Stats

134 of 139 new or added lines in 11 files covered. (96.4%)

4591 existing lines in 96 files now uncovered.

187733 of 219085 relevant lines covered (85.69%)

18687545.45 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.18

/stacks-node/src/nakamoto_node.rs

// Copyright (C) 2013-2020 Blockstack PBC, a public benefit corporation
// Copyright (C) 2020-2023 Stacks Open Internet Foundation
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.
use std::collections::HashSet;
use std::io::Write;
use std::sync::mpsc::Receiver;
use std::thread::JoinHandle;
use std::{fs, thread};

use stacks::burnchains::{BurnchainSigner, Txid};
use stacks::chainstate::burn::db::sortdb::SortitionDB;
use stacks::chainstate::burn::BlockSnapshot;
use stacks::chainstate::stacks::Error as ChainstateError;
use stacks::libstackerdb::StackerDBChunkAckData;
use stacks::monitoring;
use stacks::monitoring::update_active_miners_count_gauge;
use stacks::net::atlas::AtlasConfig;
use stacks::net::relay::Relayer;
use stacks::net::stackerdb::StackerDBs;
use stacks::net::Error as NetError;
use stacks::util_lib::db::Error as DBError;
use stacks_common::types::chainstate::SortitionId;
use stacks_common::types::StacksEpochId;

use super::{Config, EventDispatcher, Keychain};
use crate::burnchains::Error as BurnchainsError;
use crate::neon_node::{LeaderKeyRegistrationState, StacksNode as NeonNode};
use crate::run_loop::boot_nakamoto::Neon2NakaData;
use crate::run_loop::nakamoto::{Globals, RunLoop};
use crate::run_loop::RegisteredKey;

pub mod miner;
pub mod miner_db;
pub mod peer;
pub mod relayer;
pub mod signer_coordinator;
pub mod stackerdb_listener;

#[cfg(test)]
mod tests;

use self::peer::PeerThread;
use self::relayer::{RelayerDirective, RelayerThread};

pub const RELAYER_MAX_BUFFER: usize = 1;
const VRF_MOCK_MINER_KEY: u64 = 1;

pub const BLOCK_PROCESSOR_STACK_SIZE: usize = 32 * 1024 * 1024; // 32 MB

pub type BlockCommits = HashSet<Txid>;

/// Node implementation for both miners and followers.
/// This struct is used to set up the node proper and launch the p2p thread and relayer thread.
/// It is further used by the main thread to communicate with these two threads.
pub struct StacksNode {
    /// Atlas network configuration
    pub atlas_config: AtlasConfig,
    /// Global inter-thread communication handle
    pub globals: Globals,
    /// True if we're a miner
    is_miner: bool,
    /// handle to the p2p thread
    pub p2p_thread_handle: JoinHandle<()>,
    /// handle to the relayer thread
    pub relayer_thread_handle: JoinHandle<()>,
}

/// Types of errors that can arise during Nakamoto StacksNode operation
#[derive(thiserror::Error, Debug)]
pub enum Error {
    /// Can't find the block sortition snapshot for the chain tip
    #[error("Can't find the block sortition snapshot for the chain tip")]
    SnapshotNotFoundForChainTip,
    /// The burnchain tip changed while this operation was in progress
    #[error("The burnchain tip changed while this operation was in progress")]
    BurnchainTipChanged,
    /// The Stacks tip changed while this operation was in progress
    #[error("The Stacks tip changed while this operation was in progress")]
    StacksTipChanged,
    /// Signers rejected a block
    #[error("Signers rejected a block")]
    SignersRejected {
        /// Transaction IDs to exclude from the next block build (e.g., due to contextual rejections)
        temporarily_excluded_txids: HashSet<Txid>,
        /// Transaction IDs to permanently ban from the mempool
        permanently_excluded_txids: HashSet<Txid>,
    },
    /// Error while spawning a subordinate thread
    #[error("Error while spawning a subordinate thread: {0}")]
    SpawnError(std::io::Error),
    /// Injected testing errors
    #[error("Injected testing errors")]
    FaultInjection,
    /// This miner was elected, but another sortition occurred before mining started
    #[error("This miner was elected, but another sortition occurred before mining started")]
    MissedMiningOpportunity,
    /// Attempted to mine while there was no active VRF key
    #[error("Attempted to mine while there was no active VRF key")]
    NoVRFKeyActive,
    /// The parent block or tenure could not be found
    #[error("The parent block or tenure could not be found")]
    ParentNotFound,
    /// Something unexpected happened (e.g., hash mismatches)
    #[error("Something unexpected happened (e.g., hash mismatches)")]
    UnexpectedChainState,
    /// A burnchain operation failed when submitting it to the burnchain
    #[error("A burnchain operation failed when submitting it to the burnchain: {0}")]
    BurnchainSubmissionFailed(BurnchainsError),
    /// A new parent has been discovered since mining started
    #[error("A new parent has been discovered since mining started")]
    NewParentDiscovered,
    /// A failure occurred while constructing a VRF Proof
    #[error("A failure occurred while constructing a VRF Proof")]
    BadVrfConstruction,
    #[error("A failure occurred while mining: {0}")]
    MiningFailure(#[from] ChainstateError),
    /// The miner didn't accept their own block
    #[error("The miner didn't accept their own block: {0}")]
    AcceptFailure(ChainstateError),
    #[error("A failure occurred while signing a miner's block: {0}")]
    MinerSignatureError(&'static str),
    #[error("A failure occurred while signing a signer's block: {0}")]
    SignerSignatureError(String),
    /// A failure occurred while configuring the miner thread
    #[error("A failure occurred while configuring the miner thread: {0}")]
    MinerConfigurationFailed(&'static str),
    /// An error occurred while operating as the signing coordinator
    #[error("An error occurred while operating as the signing coordinator: {0}")]
    SigningCoordinatorFailure(String),
    /// An error occurred on StackerDB post
    #[error("An error occurred while uploading data to StackerDB: {0}")]
    StackerDBUploadError(StackerDBChunkAckData),
    // The thread that we tried to send to has closed
    #[error("The thread that we tried to send to has closed")]
    ChannelClosed,
    /// DBError wrapper
    #[error("DBError: {0}")]
    DBError(#[from] DBError),
    /// NetError wrapper
    #[error("NetError: {0}")]
    NetError(#[from] NetError),
    #[error("Timed out waiting for signatures")]
    SignatureTimeout,
}

impl StacksNode {
    /// This function sets the global var `GLOBAL_BURNCHAIN_SIGNER`.
    ///
    /// This variable is used for prometheus monitoring (which only
    /// runs when the feature flag `monitoring_prom` is activated).
    /// The address is set using the single-signature BTC address
    /// associated with `keychain`'s public key. This address always
    /// assumes Epoch-2.1 rules for the miner address: if the
    /// node is configured for segwit, then the miner address generated
    /// is a segwit address, otherwise it is a p2pkh.
    ///
    fn set_monitoring_miner_address(keychain: &Keychain, relayer_thread: &RelayerThread) {
        let public_key = keychain.get_pub_key();
        let miner_addr = relayer_thread
            .bitcoin_controller
            .get_miner_address(StacksEpochId::Epoch21, &public_key);
        let miner_addr_str = miner_addr.to_string();
        let _ = monitoring::set_burnchain_signer(BurnchainSigner(miner_addr_str)).map_err(|e| {
            warn!("Failed to set global burnchain signer: {e:?}");
            e
        });
    }

    pub fn spawn(
        runloop: &RunLoop,
        globals: Globals,
        // relay receiver endpoint for the p2p thread, so the relayer can feed it data to push
        relay_recv: Receiver<RelayerDirective>,
        data_from_neon: Option<Neon2NakaData>,
    ) -> StacksNode {
        let config = runloop.config().clone();
        let is_miner = runloop.is_miner();
        let burnchain = runloop.get_burnchain();
        let atlas_config = config.atlas.clone();
        let mut keychain = Keychain::default(config.node.seed.clone());
        if let Some(mining_key) = config.miner.mining_key.clone() {
            keychain.set_nakamoto_sk(mining_key);
        }

        let _ = config
            .connect_mempool_db()
            .expect("FATAL: database failure opening mempool");

        let data_from_neon = data_from_neon.unwrap_or_default();

        let mut p2p_net = data_from_neon
            .peer_network
            .unwrap_or_else(|| NeonNode::setup_peer_network(&config, &atlas_config, burnchain));

        let stackerdbs = StackerDBs::connect(&config.get_stacker_db_file_path(), true)
            .expect("FATAL: failed to connect to stacker DB");

        let relayer = Relayer::from_p2p(&mut p2p_net, stackerdbs);

        let local_peer = p2p_net.local_peer.clone();

        // setup initial key registration
        let leader_key_registration_state = if config.get_node_config(false).mock_mining {
            // mock mining, pretend to have a registered key
            let (vrf_public_key, _) = keychain.make_vrf_keypair(VRF_MOCK_MINER_KEY);
            LeaderKeyRegistrationState::Active(RegisteredKey {
                target_block_height: VRF_MOCK_MINER_KEY,
                block_height: 1,
                op_vtxindex: 1,
                vrf_public_key,
                memo: keychain.get_nakamoto_pkh().as_bytes().to_vec(),
            })
        } else {
            match &data_from_neon.leader_key_registration_state {
                LeaderKeyRegistrationState::Active(registered_key) => {
                    let pubkey_hash = keychain.get_nakamoto_pkh();
                    if pubkey_hash.as_ref() == registered_key.memo {
                        data_from_neon.leader_key_registration_state
                    } else {
                        LeaderKeyRegistrationState::Inactive
                    }
                }
                _ => LeaderKeyRegistrationState::Inactive,
            }
        };

        globals.set_initial_leader_key_registration_state(leader_key_registration_state);

        let relayer_thread =
            RelayerThread::new(runloop, local_peer.clone(), relayer, keychain.clone());

        StacksNode::set_monitoring_miner_address(&keychain, &relayer_thread);

        let relayer_thread_name = format!("relayer:{}", local_peer.port);
        let relayer_thread_handle = thread::Builder::new()
            .name(relayer_thread_name)
            .stack_size(BLOCK_PROCESSOR_STACK_SIZE)
            .spawn(move || {
                relayer_thread.main(relay_recv);
            })
            .expect("FATAL: failed to start relayer thread");

        let p2p_port = config
            .node
            .p2p_bind_addr()
            .unwrap_or_else(|| panic!("Failed to parse socket: {}", &config.node.p2p_bind))
            .port();
        let rpc_port = config
            .node
            .rpc_bind_addr()
            .unwrap_or_else(|| panic!("Failed to parse socket: {}", &config.node.rpc_bind))
            .port();

        let p2p_event_dispatcher = runloop.get_event_dispatcher();
        let p2p_thread = PeerThread::new(runloop, p2p_net);
        let p2p_thread_handle = thread::Builder::new()
            .stack_size(BLOCK_PROCESSOR_STACK_SIZE)
            .name(format!("p2p:({p2p_port},{rpc_port})"))
            .spawn(move || {
                p2p_thread.main(p2p_event_dispatcher);
            })
            .expect("FATAL: failed to start p2p thread");

        info!("Start HTTP server on: {}", &config.node.rpc_bind);
        info!("Start P2P server on: {}", &config.node.p2p_bind);

        StacksNode {
            atlas_config,
            globals,
            is_miner,
            p2p_thread_handle,
            relayer_thread_handle,
        }
    }

    /// Notify the relayer that a new burn block has been processed by the sortition db,
    ///  telling it to process the block and begin mining if this miner won.
    /// returns _false_ if the relayer hung up the channel.
    /// Called from the main thread.
    fn relayer_burnchain_notify(&self, snapshot: BlockSnapshot) -> Result<(), Error> {
        if !self.is_miner {
            // node is a follower, don't need to notify the relayer of these events.
            return Ok(());
        }

        info!(
            "Tenure: Notify burn block!";
            "consensus_hash" => %snapshot.consensus_hash,
            "burn_block_hash" => %snapshot.burn_header_hash,
            "winning_stacks_block_hash" => %snapshot.winning_stacks_block_hash,
            "burn_block_height" => &snapshot.block_height,
            "sortition_id" => %snapshot.sortition_id
        );

        // unlike in neon_node, the nakamoto node should *always* notify the relayer of
        //  a new burnchain block

        self.globals
            .relay_send
            .send(RelayerDirective::ProcessedBurnBlock(
                snapshot.consensus_hash,
                snapshot.parent_burn_header_hash,
                snapshot.winning_stacks_block_hash,
            ))
            .map_err(|_| Error::ChannelClosed)?;

        Ok(())
    }

    /// Process a state coming from the burnchain, by extracting the validated KeyRegisterOp
    /// and inspecting if a sortition was won.
    /// `ibd`: boolean indicating whether or not we are in the initial block download
    /// Called from the main thread.
    pub fn process_burnchain_state(
        &mut self,
        config: &Config,
        sortdb: &SortitionDB,
        sort_id: &SortitionId,
        ibd: bool,
    ) -> Result<(), Error> {
        let ic = sortdb.index_conn();

        let block_snapshot = SortitionDB::get_block_snapshot(&ic, sort_id)
            .expect("Failed to obtain block snapshot for processed burn block.")
            .expect("Failed to obtain block snapshot for processed burn block.");
        let block_height = block_snapshot.block_height;

        let block_commits =
            SortitionDB::get_block_commits_by_block(&ic, &block_snapshot.sortition_id)
                .expect("Unexpected SortitionDB error fetching block commits");

        let num_block_commits = block_commits.len();

        update_active_miners_count_gauge(block_commits.len() as i64);

        for op in block_commits.into_iter() {
            if op.txid == block_snapshot.winning_block_txid {
                info!(
                    "Received burnchain block #{block_height} including block_commit_op (winning) - {} ({})",
                    op.apparent_sender, &op.block_header_hash
                );
            } else if self.is_miner {
                info!(
                    "Received burnchain block #{block_height} including block_commit_op - {} ({})",
                    op.apparent_sender, &op.block_header_hash
                );
            }
        }

        let key_registers =
            SortitionDB::get_leader_keys_by_block(&ic, &block_snapshot.sortition_id)
                .expect("Unexpected SortitionDB error fetching key registers");

        let num_key_registers = key_registers.len();

        let activated_key_opt = self
            .globals
            .try_activate_leader_key_registration(block_height, key_registers);

        // save the registered VRF key
        if let (Some(activated_key), Some(path)) = (
            activated_key_opt,
            config.miner.activated_vrf_key_path.as_ref(),
        ) {
            save_activated_vrf_key(path, &activated_key);
        }

        debug!(
            "Processed burnchain state";
            "burn_height" => block_height,
            "leader_keys_count" => num_key_registers,
            "block_commits_count" => num_block_commits,
            "in_initial_block_download?" => ibd,
        );

        self.globals.set_last_sortition(block_snapshot.clone());

        // notify the relayer thread of the new sortition state
        self.relayer_burnchain_notify(block_snapshot)
    }

    /// Join all inner threads
    pub fn join(self) {
        self.relayer_thread_handle.join().unwrap();
        self.p2p_thread_handle.join().unwrap();
    }
}

pub(crate) fn save_activated_vrf_key(path: &str, activated_key: &RegisteredKey) {
    info!("Activated VRF key; saving to {path}");

    let Ok(key_json) = serde_json::to_string(&activated_key) else {
        warn!("Failed to serialize VRF key");
        return;
    };

    let mut f = match fs::File::create(path) {
        Ok(f) => f,
        Err(e) => {
            warn!("Failed to create {path}: {e:?}");
            return;
        }
    };

    if let Err(e) = f.write_all(key_json.as_bytes()) {
        warn!("Failed to write activated VRF key to {path}: {e:?}");
        return;
    }

    info!("Saved activated VRF key to {path}");
}

1	// Copyright (C) 2013-2020 Blockstack PBC, a public benefit corporation
2	// Copyright (C) 2020-2023 Stacks Open Internet Foundation
3	//
4	// This program is free software: you can redistribute it and/or modify
5	// it under the terms of the GNU General Public License as published by
6	// the Free Software Foundation, either version 3 of the License, or
7	// (at your option) any later version.
8	//
9	// This program is distributed in the hope that it will be useful,
10	// but WITHOUT ANY WARRANTY; without even the implied warranty of
11	// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12	// GNU General Public License for more details.
13	//
14	// You should have received a copy of the GNU General Public License
15	// along with this program. If not, see <http://www.gnu.org/licenses/>.
16	use std::collections::HashSet;
17	use std::io::Write;
18	use std::sync::mpsc::Receiver;
19	use std::thread::JoinHandle;
20	use std::{fs, thread};
21
22	use stacks::burnchains::{BurnchainSigner, Txid};
23	use stacks::chainstate::burn::db::sortdb::SortitionDB;
24	use stacks::chainstate::burn::BlockSnapshot;
25	use stacks::chainstate::stacks::Error as ChainstateError;
26	use stacks::libstackerdb::StackerDBChunkAckData;
27	use stacks::monitoring;
28	use stacks::monitoring::update_active_miners_count_gauge;
29	use stacks::net::atlas::AtlasConfig;
30	use stacks::net::relay::Relayer;
31	use stacks::net::stackerdb::StackerDBs;
32	use stacks::net::Error as NetError;
33	use stacks::util_lib::db::Error as DBError;
34	use stacks_common::types::chainstate::SortitionId;
35	use stacks_common::types::StacksEpochId;
36
37	use super::{Config, EventDispatcher, Keychain};
38	use crate::burnchains::Error as BurnchainsError;
39	use crate::neon_node::{LeaderKeyRegistrationState, StacksNode as NeonNode};
40	use crate::run_loop::boot_nakamoto::Neon2NakaData;
41	use crate::run_loop::nakamoto::{Globals, RunLoop};
42	use crate::run_loop::RegisteredKey;
43
44	pub mod miner;
45	pub mod miner_db;
46	pub mod peer;
47	pub mod relayer;
48	pub mod signer_coordinator;
49	pub mod stackerdb_listener;
50
51	#[cfg(test)]
52	mod tests;
53
54	use self::peer::PeerThread;
55	use self::relayer::{RelayerDirective, RelayerThread};
56
57	pub const RELAYER_MAX_BUFFER: usize = 1;
58	const VRF_MOCK_MINER_KEY: u64 = 1;
59
60	pub const BLOCK_PROCESSOR_STACK_SIZE: usize = 32 * 1024 * 1024; // 32 MB
61
62	pub type BlockCommits = HashSet<Txid>;
63
64	/// Node implementation for both miners and followers.
65	/// This struct is used to set up the node proper and launch the p2p thread and relayer thread.
66	/// It is further used by the main thread to communicate with these two threads.
67	pub struct StacksNode {
68	/// Atlas network configuration
69	pub atlas_config: AtlasConfig,
70	/// Global inter-thread communication handle
71	pub globals: Globals,
72	/// True if we're a miner
73	is_miner: bool,
74	/// handle to the p2p thread
75	pub p2p_thread_handle: JoinHandle<()>,
76	/// handle to the relayer thread
77	pub relayer_thread_handle: JoinHandle<()>,
78	}
79
80	/// Types of errors that can arise during Nakamoto StacksNode operation
81	#[derive(thiserror::Error, Debug)]
82	pub enum Error {
83	/// Can't find the block sortition snapshot for the chain tip
84	#[error("Can't find the block sortition snapshot for the chain tip")]
85	SnapshotNotFoundForChainTip,
86	/// The burnchain tip changed while this operation was in progress
87	#[error("The burnchain tip changed while this operation was in progress")]
88	BurnchainTipChanged,
89	/// The Stacks tip changed while this operation was in progress
90	#[error("The Stacks tip changed while this operation was in progress")]
91	StacksTipChanged,
92	/// Signers rejected a block
93	#[error("Signers rejected a block")]
94	SignersRejected {
95	/// Transaction IDs to exclude from the next block build (e.g., due to contextual rejections)
96	temporarily_excluded_txids: HashSet<Txid>,
97	/// Transaction IDs to permanently ban from the mempool
98	permanently_excluded_txids: HashSet<Txid>,
99	},
100	/// Error while spawning a subordinate thread
101	#[error("Error while spawning a subordinate thread: {0}")]
102	SpawnError(std::io::Error),
103	/// Injected testing errors
104	#[error("Injected testing errors")]
105	FaultInjection,
106	/// This miner was elected, but another sortition occurred before mining started
107	#[error("This miner was elected, but another sortition occurred before mining started")]
108	MissedMiningOpportunity,
109	/// Attempted to mine while there was no active VRF key
110	#[error("Attempted to mine while there was no active VRF key")]
111	NoVRFKeyActive,
112	/// The parent block or tenure could not be found
113	#[error("The parent block or tenure could not be found")]
114	ParentNotFound,
115	/// Something unexpected happened (e.g., hash mismatches)
116	#[error("Something unexpected happened (e.g., hash mismatches)")]
117	UnexpectedChainState,
118	/// A burnchain operation failed when submitting it to the burnchain
119	#[error("A burnchain operation failed when submitting it to the burnchain: {0}")]
120	BurnchainSubmissionFailed(BurnchainsError),
121	/// A new parent has been discovered since mining started
122	#[error("A new parent has been discovered since mining started")]
123	NewParentDiscovered,
124	/// A failure occurred while constructing a VRF Proof
125	#[error("A failure occurred while constructing a VRF Proof")]
126	BadVrfConstruction,
127	#[error("A failure occurred while mining: {0}")]
128	MiningFailure(#[from] ChainstateError),
129	/// The miner didn't accept their own block
130	#[error("The miner didn't accept their own block: {0}")]
131	AcceptFailure(ChainstateError),
132	#[error("A failure occurred while signing a miner's block: {0}")]
133	MinerSignatureError(&'static str),
134	#[error("A failure occurred while signing a signer's block: {0}")]
135	SignerSignatureError(String),
136	/// A failure occurred while configuring the miner thread
137	#[error("A failure occurred while configuring the miner thread: {0}")]
138	MinerConfigurationFailed(&'static str),
139	/// An error occurred while operating as the signing coordinator
140	#[error("An error occurred while operating as the signing coordinator: {0}")]
141	SigningCoordinatorFailure(String),
142	/// An error occurred on StackerDB post
143	#[error("An error occurred while uploading data to StackerDB: {0}")]
144	StackerDBUploadError(StackerDBChunkAckData),
145	// The thread that we tried to send to has closed
146	#[error("The thread that we tried to send to has closed")]
147	ChannelClosed,
148	/// DBError wrapper
149	#[error("DBError: {0}")]
150	DBError(#[from] DBError),
151	/// NetError wrapper
152	#[error("NetError: {0}")]
153	NetError(#[from] NetError),
154	#[error("Timed out waiting for signatures")]
155	SignatureTimeout,
156	}
157
158	impl StacksNode {
159	/// This function sets the global var `GLOBAL_BURNCHAIN_SIGNER`.
160	///
161	/// This variable is used for prometheus monitoring (which only
162	/// runs when the feature flag `monitoring_prom` is activated).
163	/// The address is set using the single-signature BTC address
164	/// associated with `keychain`'s public key. This address always
165	/// assumes Epoch-2.1 rules for the miner address: if the
166	/// node is configured for segwit, then the miner address generated
167	/// is a segwit address, otherwise it is a p2pkh.
168	///
169	fn set_monitoring_miner_address(keychain: &Keychain, relayer_thread: &RelayerThread) {	241✔
170	let public_key = keychain.get_pub_key();	241✔
171	let miner_addr = relayer_thread	241✔
172	.bitcoin_controller	241✔
173	.get_miner_address(StacksEpochId::Epoch21, &public_key);	241✔
174	let miner_addr_str = miner_addr.to_string();	241✔
175	let _ = monitoring::set_burnchain_signer(BurnchainSigner(miner_addr_str)).map_err(\|e\| {	241✔
176	warn!("Failed to set global burnchain signer: {e:?}");	241✔
177	e	241✔
178	});	241✔
179	}	241✔
180
181	pub fn spawn(	241✔
182	runloop: &RunLoop,	241✔
183	globals: Globals,	241✔
184	// relay receiver endpoint for the p2p thread, so the relayer can feed it data to push	241✔
185	relay_recv: Receiver<RelayerDirective>,	241✔
186	data_from_neon: Option<Neon2NakaData>,	241✔
187	) -> StacksNode {	241✔
188	let config = runloop.config().clone();	241✔
189	let is_miner = runloop.is_miner();	241✔
190	let burnchain = runloop.get_burnchain();	241✔
191	let atlas_config = config.atlas.clone();	241✔
192	let mut keychain = Keychain::default(config.node.seed.clone());	241✔
193	if let Some(mining_key) = config.miner.mining_key.clone() {	241✔
194	keychain.set_nakamoto_sk(mining_key);	241✔
195	}	241✔
196
197	let _ = config	241✔
198	.connect_mempool_db()	241✔
199	.expect("FATAL: database failure opening mempool");	241✔
200
201	let data_from_neon = data_from_neon.unwrap_or_default();	241✔
202
203	let mut p2p_net = data_from_neon	241✔
204	.peer_network	241✔
205	.unwrap_or_else(\|\| NeonNode::setup_peer_network(&config, &atlas_config, burnchain));	241✔
206
207	let stackerdbs = StackerDBs::connect(&config.get_stacker_db_file_path(), true)	241✔
208	.expect("FATAL: failed to connect to stacker DB");	241✔
209
210	let relayer = Relayer::from_p2p(&mut p2p_net, stackerdbs);	241✔
211
212	let local_peer = p2p_net.local_peer.clone();	241✔
213
214	// setup initial key registration
215	let leader_key_registration_state = if config.get_node_config(false).mock_mining {	241✔
216	// mock mining, pretend to have a registered key
217	let (vrf_public_key, _) = keychain.make_vrf_keypair(VRF_MOCK_MINER_KEY);	5✔
218	LeaderKeyRegistrationState::Active(RegisteredKey {	5✔
219	target_block_height: VRF_MOCK_MINER_KEY,	5✔
220	block_height: 1,	5✔
221	op_vtxindex: 1,	5✔
222	vrf_public_key,	5✔
223	memo: keychain.get_nakamoto_pkh().as_bytes().to_vec(),	5✔
224	})	5✔
225	} else {
226	match &data_from_neon.leader_key_registration_state {	236✔
227	LeaderKeyRegistrationState::Active(registered_key) => {	229✔
228	let pubkey_hash = keychain.get_nakamoto_pkh();	229✔
229	if pubkey_hash.as_ref() == registered_key.memo {	229✔
230	data_from_neon.leader_key_registration_state	195✔
231	} else {
232	LeaderKeyRegistrationState::Inactive	34✔
233	}
234	}
235	_ => LeaderKeyRegistrationState::Inactive,	7✔
236	}
237	};
238
239	globals.set_initial_leader_key_registration_state(leader_key_registration_state);	241✔
240
241	let relayer_thread =	241✔
242	RelayerThread::new(runloop, local_peer.clone(), relayer, keychain.clone());	241✔
243
244	StacksNode::set_monitoring_miner_address(&keychain, &relayer_thread);	241✔
245
246	let relayer_thread_name = format!("relayer:{}", local_peer.port);	241✔
247	let relayer_thread_handle = thread::Builder::new()	241✔
248	.name(relayer_thread_name)	241✔
249	.stack_size(BLOCK_PROCESSOR_STACK_SIZE)	241✔
250	.spawn(move \|\| {	241✔
251	relayer_thread.main(relay_recv);	241✔
252	})	241✔
253	.expect("FATAL: failed to start relayer thread");	241✔
254
255	let p2p_port = config	241✔
256	.node	241✔
257	.p2p_bind_addr()	241✔
258	.unwrap_or_else(\|\| panic!("Failed to parse socket: {}", &config.node.p2p_bind))	241✔
259	.port();	241✔
260	let rpc_port = config	241✔
261	.node	241✔
262	.rpc_bind_addr()	241✔
263	.unwrap_or_else(\|\| panic!("Failed to parse socket: {}", &config.node.rpc_bind))	241✔
264	.port();	241✔
265
266	let p2p_event_dispatcher = runloop.get_event_dispatcher();	241✔
267	let p2p_thread = PeerThread::new(runloop, p2p_net);	241✔
268	let p2p_thread_handle = thread::Builder::new()	241✔
269	.stack_size(BLOCK_PROCESSOR_STACK_SIZE)	241✔
270	.name(format!("p2p:({p2p_port},{rpc_port})"))	241✔
271	.spawn(move \|\| {	241✔
272	p2p_thread.main(p2p_event_dispatcher);	241✔
273	})	241✔
274	.expect("FATAL: failed to start p2p thread");	241✔
275
276	info!("Start HTTP server on: {}", &config.node.rpc_bind);	241✔
277	info!("Start P2P server on: {}", &config.node.p2p_bind);	241✔
278
279	StacksNode {	241✔
280	atlas_config,	241✔
281	globals,	241✔
282	is_miner,	241✔
283	p2p_thread_handle,	241✔
284	relayer_thread_handle,	241✔
285	}	241✔
286	}	241✔
287
288	/// Notify the relayer that a new burn block has been processed by the sortition db,
289	/// telling it to process the block and begin mining if this miner won.
290	/// returns _false_ if the relayer hung up the channel.
291	/// Called from the main thread.
292	fn relayer_burnchain_notify(&self, snapshot: BlockSnapshot) -> Result<(), Error> {	4,407✔
293	if !self.is_miner {	4,407✔
294	// node is a follower, don't need to notify the relayer of these events.
295	return Ok(());	95✔
296	}	4,312✔
297
298	info!(	4,312✔
299	"Tenure: Notify burn block!";
300	"consensus_hash" => %snapshot.consensus_hash,
301	"burn_block_hash" => %snapshot.burn_header_hash,
302	"winning_stacks_block_hash" => %snapshot.winning_stacks_block_hash,
303	"burn_block_height" => &snapshot.block_height,	4,312✔
304	"sortition_id" => %snapshot.sortition_id
305	);
306
307	// unlike in neon_node, the nakamoto node should always notify the relayer of
308	// a new burnchain block
309
310	self.globals	4,312✔
311	.relay_send	4,312✔
312	.send(RelayerDirective::ProcessedBurnBlock(	4,312✔
313	snapshot.consensus_hash,	4,312✔
314	snapshot.parent_burn_header_hash,	4,312✔
315	snapshot.winning_stacks_block_hash,	4,312✔
316	))	4,312✔
317	.map_err(\|_\| Error::ChannelClosed)?;	4,312✔
318
319	Ok(())	4,312✔
320	}	4,407✔
321
322	/// Process a state coming from the burnchain, by extracting the validated KeyRegisterOp
323	/// and inspecting if a sortition was won.
324	/// `ibd`: boolean indicating whether or not we are in the initial block download
325	/// Called from the main thread.
326	pub fn process_burnchain_state(	4,407✔
327	&mut self,	4,407✔
328	config: &Config,	4,407✔
329	sortdb: &SortitionDB,	4,407✔
330	sort_id: &SortitionId,	4,407✔
331	ibd: bool,	4,407✔
332	) -> Result<(), Error> {	4,407✔
333	let ic = sortdb.index_conn();	4,407✔
334
335	let block_snapshot = SortitionDB::get_block_snapshot(&ic, sort_id)	4,407✔
336	.expect("Failed to obtain block snapshot for processed burn block.")	4,407✔
337	.expect("Failed to obtain block snapshot for processed burn block.");	4,407✔
338	let block_height = block_snapshot.block_height;	4,407✔
339
340	let block_commits =	4,407✔
341	SortitionDB::get_block_commits_by_block(&ic, &block_snapshot.sortition_id)	4,407✔
342	.expect("Unexpected SortitionDB error fetching block commits");	4,407✔
343
344	let num_block_commits = block_commits.len();	4,407✔
345
346	update_active_miners_count_gauge(block_commits.len() as i64);	4,407✔
347
348	for op in block_commits.into_iter() {	4,478✔
349	if op.txid == block_snapshot.winning_block_txid {	4,330✔
350	info!(	4,253✔
351	"Received burnchain block #{block_height} including block_commit_op (winning) - {} ({})",
352	op.apparent_sender, &op.block_header_hash	4,253✔
353	);
354	} else if self.is_miner {	77✔
355	info!(	77✔
356	"Received burnchain block #{block_height} including block_commit_op - {} ({})",
357	op.apparent_sender, &op.block_header_hash	77✔
358	);
UNCOV 359	}	×
360	}
361
362	let key_registers =	4,407✔
363	SortitionDB::get_leader_keys_by_block(&ic, &block_snapshot.sortition_id)	4,407✔
364	.expect("Unexpected SortitionDB error fetching key registers");	4,407✔
365
366	let num_key_registers = key_registers.len();	4,407✔
367
368	let activated_key_opt = self	4,407✔
369	.globals	4,407✔
370	.try_activate_leader_key_registration(block_height, key_registers);	4,407✔
371
372	// save the registered VRF key
373	if let (Some(activated_key), Some(path)) = (	36✔
374	activated_key_opt,	4,407✔
375	config.miner.activated_vrf_key_path.as_ref(),	4,407✔
376	) {	36✔
377	save_activated_vrf_key(path, &activated_key);	36✔
378	}	4,371✔
379
380	debug!(	4,407✔
381	"Processed burnchain state";
UNCOV 382	"burn_height" => block_height,	×
UNCOV 383	"leader_keys_count" => num_key_registers,	×
UNCOV 384	"block_commits_count" => num_block_commits,	×
UNCOV 385	"in_initial_block_download?" => ibd,	×
386	);
387
388	self.globals.set_last_sortition(block_snapshot.clone());	4,407✔
389
390	// notify the relayer thread of the new sortition state
391	self.relayer_burnchain_notify(block_snapshot)	4,407✔
392	}	4,407✔
393
394	/// Join all inner threads
395	pub fn join(self) {	221✔
396	self.relayer_thread_handle.join().unwrap();	221✔
397	self.p2p_thread_handle.join().unwrap();	221✔
398	}	221✔
399	}
400
401	pub(crate) fn save_activated_vrf_key(path: &str, activated_key: &RegisteredKey) {	38✔
402	info!("Activated VRF key; saving to {path}");	38✔
403
404	let Ok(key_json) = serde_json::to_string(&activated_key) else {	38✔
UNCOV 405	warn!("Failed to serialize VRF key");	×
406	return;	×
407	};
408
409	let mut f = match fs::File::create(path) {	38✔
410	Ok(f) => f,	38✔
UNCOV 411	Err(e) => {	×
UNCOV 412	warn!("Failed to create {path}: {e:?}");	×
413	return;	×
414	}
415	};
416
417	if let Err(e) = f.write_all(key_json.as_bytes()) {	38✔
UNCOV 418	warn!("Failed to write activated VRF key to {path}: {e:?}");	×
UNCOV 419	return;	×
420	}	38✔
421
422	info!("Saved activated VRF key to {path}");	38✔
423	}	38✔

stacks-network / stacks-core / 25404138305-1

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous