23943169302

Committed 03 Apr 2026 10:28AM UTC coverage: 77.573% (-8.1%) from 85.712%

Build # 23943169302

Build Type

Pull #7076

github

Specific Base 7f2377

Committed by

web-flow

Commit Message

Merge bb87ecec2 into c529ad924

Pull Request Pull Request #7076: feat: sortition side-table copy and validation

Coverage Stats

3743 of 4318 new or added lines in 19 files covered. (86.68%)

19304 existing lines in 182 files now uncovered.

172097 of 221852 relevant lines covered (77.57%)

7722182.76 hits per line

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

5.11

/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,
    /// 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	/// Error while spawning a subordinate thread
96	#[error("Error while spawning a subordinate thread: {0}")]
97	SpawnError(std::io::Error),
98	/// Injected testing errors
99	#[error("Injected testing errors")]
100	FaultInjection,
101	/// This miner was elected, but another sortition occurred before mining started
102	#[error("This miner was elected, but another sortition occurred before mining started")]
103	MissedMiningOpportunity,
104	/// Attempted to mine while there was no active VRF key
105	#[error("Attempted to mine while there was no active VRF key")]
106	NoVRFKeyActive,
107	/// The parent block or tenure could not be found
108	#[error("The parent block or tenure could not be found")]
109	ParentNotFound,
110	/// Something unexpected happened (e.g., hash mismatches)
111	#[error("Something unexpected happened (e.g., hash mismatches)")]
112	UnexpectedChainState,
113	/// A burnchain operation failed when submitting it to the burnchain
114	#[error("A burnchain operation failed when submitting it to the burnchain: {0}")]
115	BurnchainSubmissionFailed(BurnchainsError),
116	/// A new parent has been discovered since mining started
117	#[error("A new parent has been discovered since mining started")]
118	NewParentDiscovered,
119	/// A failure occurred while constructing a VRF Proof
120	#[error("A failure occurred while constructing a VRF Proof")]
121	BadVrfConstruction,
122	#[error("A failure occurred while mining: {0}")]
123	MiningFailure(#[from] ChainstateError),
124	/// The miner didn't accept their own block
125	#[error("The miner didn't accept their own block: {0}")]
126	AcceptFailure(ChainstateError),
127	#[error("A failure occurred while signing a miner's block: {0}")]
128	MinerSignatureError(&'static str),
129	#[error("A failure occurred while signing a signer's block: {0}")]
130	SignerSignatureError(String),
131	/// A failure occurred while configuring the miner thread
132	#[error("A failure occurred while configuring the miner thread: {0}")]
133	MinerConfigurationFailed(&'static str),
134	/// An error occurred while operating as the signing coordinator
135	#[error("An error occurred while operating as the signing coordinator: {0}")]
136	SigningCoordinatorFailure(String),
137	/// An error occurred on StackerDB post
138	#[error("An error occurred while uploading data to StackerDB: {0}")]
139	StackerDBUploadError(StackerDBChunkAckData),
140	// The thread that we tried to send to has closed
141	#[error("The thread that we tried to send to has closed")]
142	ChannelClosed,
143	/// DBError wrapper
144	#[error("DBError: {0}")]
145	DBError(#[from] DBError),
146	/// NetError wrapper
147	#[error("NetError: {0}")]
148	NetError(#[from] NetError),
149	#[error("Timed out waiting for signatures")]
150	SignatureTimeout,
151	}
152
153	impl StacksNode {
154	/// This function sets the global var `GLOBAL_BURNCHAIN_SIGNER`.
155	///
156	/// This variable is used for prometheus monitoring (which only
157	/// runs when the feature flag `monitoring_prom` is activated).
158	/// The address is set using the single-signature BTC address
159	/// associated with `keychain`'s public key. This address always
160	/// assumes Epoch-2.1 rules for the miner address: if the
161	/// node is configured for segwit, then the miner address generated
162	/// is a segwit address, otherwise it is a p2pkh.
163	///
UNCOV 164	fn set_monitoring_miner_address(keychain: &Keychain, relayer_thread: &RelayerThread) {	×
UNCOV 165	let public_key = keychain.get_pub_key();	×
UNCOV 166	let miner_addr = relayer_thread	×
UNCOV 167	.bitcoin_controller	×
UNCOV 168	.get_miner_address(StacksEpochId::Epoch21, &public_key);	×
UNCOV 169	let miner_addr_str = miner_addr.to_string();	×
UNCOV 170	let _ = monitoring::set_burnchain_signer(BurnchainSigner(miner_addr_str)).map_err(\|e\| {	×
UNCOV 171	warn!("Failed to set global burnchain signer: {e:?}");	×
UNCOV 172	e	×
UNCOV 173	});	×
UNCOV 174	}	×
175
UNCOV 176	pub fn spawn(	×
UNCOV 177	runloop: &RunLoop,	×
UNCOV 178	globals: Globals,	×
UNCOV 179	// relay receiver endpoint for the p2p thread, so the relayer can feed it data to push	×
UNCOV 180	relay_recv: Receiver<RelayerDirective>,	×
UNCOV 181	data_from_neon: Option<Neon2NakaData>,	×
UNCOV 182	) -> StacksNode {	×
UNCOV 183	let config = runloop.config().clone();	×
UNCOV 184	let is_miner = runloop.is_miner();	×
UNCOV 185	let burnchain = runloop.get_burnchain();	×
UNCOV 186	let atlas_config = config.atlas.clone();	×
UNCOV 187	let mut keychain = Keychain::default(config.node.seed.clone());	×
UNCOV 188	if let Some(mining_key) = config.miner.mining_key.clone() {	×
UNCOV 189	keychain.set_nakamoto_sk(mining_key);	×
UNCOV 190	}	×
191
UNCOV 192	let _ = config	×
UNCOV 193	.connect_mempool_db()	×
UNCOV 194	.expect("FATAL: database failure opening mempool");	×
195
UNCOV 196	let data_from_neon = data_from_neon.unwrap_or_default();	×
197
UNCOV 198	let mut p2p_net = data_from_neon	×
UNCOV 199	.peer_network	×
UNCOV 200	.unwrap_or_else(\|\| NeonNode::setup_peer_network(&config, &atlas_config, burnchain));	×
201
UNCOV 202	let stackerdbs = StackerDBs::connect(&config.get_stacker_db_file_path(), true)	×
UNCOV 203	.expect("FATAL: failed to connect to stacker DB");	×
204
UNCOV 205	let relayer = Relayer::from_p2p(&mut p2p_net, stackerdbs);	×
206
UNCOV 207	let local_peer = p2p_net.local_peer.clone();	×
208
209	// setup initial key registration
UNCOV 210	let leader_key_registration_state = if config.get_node_config(false).mock_mining {	×
211	// mock mining, pretend to have a registered key
UNCOV 212	let (vrf_public_key, _) = keychain.make_vrf_keypair(VRF_MOCK_MINER_KEY);	×
UNCOV 213	LeaderKeyRegistrationState::Active(RegisteredKey {	×
UNCOV 214	target_block_height: VRF_MOCK_MINER_KEY,	×
UNCOV 215	block_height: 1,	×
UNCOV 216	op_vtxindex: 1,	×
UNCOV 217	vrf_public_key,	×
UNCOV 218	memo: keychain.get_nakamoto_pkh().as_bytes().to_vec(),	×
UNCOV 219	})	×
220	} else {
UNCOV 221	match &data_from_neon.leader_key_registration_state {	×
UNCOV 222	LeaderKeyRegistrationState::Active(registered_key) => {	×
UNCOV 223	let pubkey_hash = keychain.get_nakamoto_pkh();	×
UNCOV 224	if pubkey_hash.as_ref() == registered_key.memo {	×
UNCOV 225	data_from_neon.leader_key_registration_state	×
226	} else {
UNCOV 227	LeaderKeyRegistrationState::Inactive	×
228	}
229	}
UNCOV 230	_ => LeaderKeyRegistrationState::Inactive,	×
231	}
232	};
233
UNCOV 234	globals.set_initial_leader_key_registration_state(leader_key_registration_state);	×
235
UNCOV 236	let relayer_thread =	×
UNCOV 237	RelayerThread::new(runloop, local_peer.clone(), relayer, keychain.clone());	×
238
UNCOV 239	StacksNode::set_monitoring_miner_address(&keychain, &relayer_thread);	×
240
UNCOV 241	let relayer_thread_name = format!("relayer:{}", local_peer.port);	×
UNCOV 242	let relayer_thread_handle = thread::Builder::new()	×
UNCOV 243	.name(relayer_thread_name)	×
UNCOV 244	.stack_size(BLOCK_PROCESSOR_STACK_SIZE)	×
UNCOV 245	.spawn(move \|\| {	×
UNCOV 246	relayer_thread.main(relay_recv);	×
UNCOV 247	})	×
UNCOV 248	.expect("FATAL: failed to start relayer thread");	×
249
UNCOV 250	let p2p_port = config	×
UNCOV 251	.node	×
UNCOV 252	.p2p_bind_addr()	×
UNCOV 253	.unwrap_or_else(\|\| panic!("Failed to parse socket: {}", &config.node.p2p_bind))	×
UNCOV 254	.port();	×
UNCOV 255	let rpc_port = config	×
UNCOV 256	.node	×
UNCOV 257	.rpc_bind_addr()	×
UNCOV 258	.unwrap_or_else(\|\| panic!("Failed to parse socket: {}", &config.node.rpc_bind))	×
UNCOV 259	.port();	×
260
UNCOV 261	let p2p_event_dispatcher = runloop.get_event_dispatcher();	×
UNCOV 262	let p2p_thread = PeerThread::new(runloop, p2p_net);	×
UNCOV 263	let p2p_thread_handle = thread::Builder::new()	×
UNCOV 264	.stack_size(BLOCK_PROCESSOR_STACK_SIZE)	×
UNCOV 265	.name(format!("p2p:({p2p_port},{rpc_port})"))	×
UNCOV 266	.spawn(move \|\| {	×
UNCOV 267	p2p_thread.main(p2p_event_dispatcher);	×
UNCOV 268	})	×
UNCOV 269	.expect("FATAL: failed to start p2p thread");	×
270
UNCOV 271	info!("Start HTTP server on: {}", &config.node.rpc_bind);	×
UNCOV 272	info!("Start P2P server on: {}", &config.node.p2p_bind);	×
273
UNCOV 274	StacksNode {	×
UNCOV 275	atlas_config,	×
UNCOV 276	globals,	×
UNCOV 277	is_miner,	×
UNCOV 278	p2p_thread_handle,	×
UNCOV 279	relayer_thread_handle,	×
UNCOV 280	}	×
UNCOV 281	}	×
282
283	/// Notify the relayer that a new burn block has been processed by the sortition db,
284	/// telling it to process the block and begin mining if this miner won.
285	/// returns _false_ if the relayer hung up the channel.
286	/// Called from the main thread.
UNCOV 287	fn relayer_burnchain_notify(&self, snapshot: BlockSnapshot) -> Result<(), Error> {	×
UNCOV 288	if !self.is_miner {	×
289	// node is a follower, don't need to notify the relayer of these events.
UNCOV 290	return Ok(());	×
UNCOV 291	}	×
292
UNCOV 293	info!(	×
294	"Tenure: Notify burn block!";
295	"consensus_hash" => %snapshot.consensus_hash,
296	"burn_block_hash" => %snapshot.burn_header_hash,
297	"winning_stacks_block_hash" => %snapshot.winning_stacks_block_hash,
UNCOV 298	"burn_block_height" => &snapshot.block_height,	×
299	"sortition_id" => %snapshot.sortition_id
300	);
301
302	// unlike in neon_node, the nakamoto node should always notify the relayer of
303	// a new burnchain block
304
UNCOV 305	self.globals	×
UNCOV 306	.relay_send	×
UNCOV 307	.send(RelayerDirective::ProcessedBurnBlock(	×
UNCOV 308	snapshot.consensus_hash,	×
UNCOV 309	snapshot.parent_burn_header_hash,	×
UNCOV 310	snapshot.winning_stacks_block_hash,	×
UNCOV 311	))	×
UNCOV 312	.map_err(\|_\| Error::ChannelClosed)?;	×
313
UNCOV 314	Ok(())	×
UNCOV 315	}	×
316
317	/// Process a state coming from the burnchain, by extracting the validated KeyRegisterOp
318	/// and inspecting if a sortition was won.
319	/// `ibd`: boolean indicating whether or not we are in the initial block download
320	/// Called from the main thread.
UNCOV 321	pub fn process_burnchain_state(	×
UNCOV 322	&mut self,	×
UNCOV 323	config: &Config,	×
UNCOV 324	sortdb: &SortitionDB,	×
UNCOV 325	sort_id: &SortitionId,	×
UNCOV 326	ibd: bool,	×
UNCOV 327	) -> Result<(), Error> {	×
UNCOV 328	let ic = sortdb.index_conn();	×
329
UNCOV 330	let block_snapshot = SortitionDB::get_block_snapshot(&ic, sort_id)	×
UNCOV 331	.expect("Failed to obtain block snapshot for processed burn block.")	×
UNCOV 332	.expect("Failed to obtain block snapshot for processed burn block.");	×
UNCOV 333	let block_height = block_snapshot.block_height;	×
334
UNCOV 335	let block_commits =	×
UNCOV 336	SortitionDB::get_block_commits_by_block(&ic, &block_snapshot.sortition_id)	×
UNCOV 337	.expect("Unexpected SortitionDB error fetching block commits");	×
338
UNCOV 339	let num_block_commits = block_commits.len();	×
340
UNCOV 341	update_active_miners_count_gauge(block_commits.len() as i64);	×
342
UNCOV 343	for op in block_commits.into_iter() {	×
UNCOV 344	if op.txid == block_snapshot.winning_block_txid {	×
UNCOV 345	info!(	×
346	"Received burnchain block #{block_height} including block_commit_op (winning) - {} ({})",
UNCOV 347	op.apparent_sender, &op.block_header_hash	×
348	);
UNCOV 349	} else if self.is_miner {	×
UNCOV 350	info!(	×
351	"Received burnchain block #{block_height} including block_commit_op - {} ({})",
UNCOV 352	op.apparent_sender, &op.block_header_hash	×
353	);
354	}	×
355	}
356
UNCOV 357	let key_registers =	×
UNCOV 358	SortitionDB::get_leader_keys_by_block(&ic, &block_snapshot.sortition_id)	×
UNCOV 359	.expect("Unexpected SortitionDB error fetching key registers");	×
360
UNCOV 361	let num_key_registers = key_registers.len();	×
362
UNCOV 363	let activated_key_opt = self	×
UNCOV 364	.globals	×
UNCOV 365	.try_activate_leader_key_registration(block_height, key_registers);	×
366
367	// save the registered VRF key
UNCOV 368	if let (Some(activated_key), Some(path)) = (	×
UNCOV 369	activated_key_opt,	×
UNCOV 370	config.miner.activated_vrf_key_path.as_ref(),	×
UNCOV 371	) {	×
UNCOV 372	save_activated_vrf_key(path, &activated_key);	×
UNCOV 373	}	×
374
UNCOV 375	debug!(	×
376	"Processed burnchain state";
377	"burn_height" => block_height,	×
378	"leader_keys_count" => num_key_registers,	×
379	"block_commits_count" => num_block_commits,	×
380	"in_initial_block_download?" => ibd,	×
381	);
382
UNCOV 383	self.globals.set_last_sortition(block_snapshot.clone());	×
384
385	// notify the relayer thread of the new sortition state
UNCOV 386	self.relayer_burnchain_notify(block_snapshot)	×
UNCOV 387	}	×
388
389	/// Join all inner threads
UNCOV 390	pub fn join(self) {	×
UNCOV 391	self.relayer_thread_handle.join().unwrap();	×
UNCOV 392	self.p2p_thread_handle.join().unwrap();	×
UNCOV 393	}	×
394	}
395
396	pub(crate) fn save_activated_vrf_key(path: &str, activated_key: &RegisteredKey) {	2✔
397	info!("Activated VRF key; saving to {path}");	2✔
398
399	let Ok(key_json) = serde_json::to_string(&activated_key) else {	2✔
400	warn!("Failed to serialize VRF key");	×
401	return;	×
402	};
403
404	let mut f = match fs::File::create(path) {	2✔
405	Ok(f) => f,	2✔
406	Err(e) => {	×
407	warn!("Failed to create {path}: {e:?}");	×
408	return;	×
409	}
410	};
411
412	if let Err(e) = f.write_all(key_json.as_bytes()) {	2✔
413	warn!("Failed to write activated VRF key to {path}: {e:?}");	×
414	return;	×
415	}	2✔
416
417	info!("Saved activated VRF key to {path}");	2✔
418	}	2✔

stacks-network / stacks-core / 23943169302

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