24264100700

// Copyright (C) 2013-2020 Blockstack PBC, a public benefit corporation
// Copyright (C) 2020-2024 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::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;

use stacks::burnchains::{Burnchain, Error as burnchain_error};
use stacks_common::util::{get_epoch_time_secs, sleep_ms};

use crate::burnchains::BurnchainTip;
use crate::Config;

#[derive(Clone)]
pub struct PoxSyncWatchdogComms {
    /// how many passes in the p2p state machine have taken place since startup?
    p2p_state_passes: Arc<AtomicU64>,
    /// how many times have we done an inv sync?
    inv_sync_passes: Arc<AtomicU64>,
    /// how many times have we done a download pass?
    download_passes: Arc<AtomicU64>,
    /// What's our last IBD status?
    last_ibd: Arc<AtomicBool>,
    /// Should keep running?
    should_keep_running: Arc<AtomicBool>,
}

impl PoxSyncWatchdogComms {
    pub fn new(should_keep_running: Arc<AtomicBool>) -> PoxSyncWatchdogComms {
        PoxSyncWatchdogComms {
            p2p_state_passes: Arc::new(AtomicU64::new(0)),
            inv_sync_passes: Arc::new(AtomicU64::new(0)),
            download_passes: Arc::new(AtomicU64::new(0)),
            last_ibd: Arc::new(AtomicBool::new(true)),
            should_keep_running,
        }
    }

    pub fn get_p2p_state_passes(&self) -> u64 {
        self.p2p_state_passes.load(Ordering::SeqCst)
    }

    pub fn get_inv_sync_passes(&self) -> u64 {
        self.inv_sync_passes.load(Ordering::SeqCst)
    }

    pub fn get_download_passes(&self) -> u64 {
        self.download_passes.load(Ordering::SeqCst)
    }

    pub fn get_ibd(&self) -> bool {
        self.last_ibd.load(Ordering::SeqCst)
    }

    fn interruptable_sleep(&self, secs: u64) -> Result<(), burnchain_error> {
        let deadline = secs + get_epoch_time_secs();
        while get_epoch_time_secs() < deadline {
            sleep_ms(1000);
            if !self.should_keep_running() {
                return Err(burnchain_error::CoordinatorClosed);
            }
        }
        Ok(())
    }

    pub fn should_keep_running(&self) -> bool {
        self.should_keep_running.load(Ordering::SeqCst)
    }

    pub fn notify_p2p_state_pass(&mut self) {
        self.p2p_state_passes.fetch_add(1, Ordering::SeqCst);
    }

    pub fn notify_inv_sync_pass(&mut self) {
        self.inv_sync_passes.fetch_add(1, Ordering::SeqCst);
    }

    pub fn notify_download_pass(&mut self) {
        self.download_passes.fetch_add(1, Ordering::SeqCst);
    }

    pub fn set_ibd(&mut self, value: bool) {
        self.last_ibd.store(value, Ordering::SeqCst);
    }
}

/// Monitor the state of the Stacks blockchain as the peer network and relay threads download and
/// proces Stacks blocks.  Don't allow the node to process the next PoX reward cycle's sortitions
/// unless it's reasonably sure that it has processed all Stacks blocks for this reward cycle.
/// This struct monitors the Stacks chainstate to make this determination.
pub struct PoxSyncWatchdog {
    /// time between burnchain syncs in steady state
    steady_state_burnchain_sync_interval: u64,
    /// handle to relayer thread that informs the watchdog when the P2P state-machine does stuff
    relayer_comms: PoxSyncWatchdogComms,
    /// should this sync watchdog always download? used in integration tests.
    unconditionally_download: bool,
}

impl PoxSyncWatchdog {
    pub fn new(
        config: &Config,
        watchdog_comms: PoxSyncWatchdogComms,
    ) -> Result<PoxSyncWatchdog, String> {
        let burnchain_poll_time = config.burnchain.poll_time_secs;
        let unconditionally_download = config.node.pox_sync_sample_secs == 0;

        Ok(PoxSyncWatchdog {
            unconditionally_download,
            steady_state_burnchain_sync_interval: burnchain_poll_time,
            relayer_comms: watchdog_comms,
        })
    }

    pub fn make_comms_handle(&self) -> PoxSyncWatchdogComms {
        self.relayer_comms.clone()
    }

    /// Are we in the initial burnchain block download? i.e. is the burn tip snapshot far enough away
    /// from the burnchain height that we should be eagerly downloading snapshots?
    fn infer_initial_burnchain_block_download(
        burnchain: &Burnchain,
        last_processed_height: u64,
        burnchain_height: u64,
    ) -> bool {
        let ibd =
            last_processed_height + (burnchain.stable_confirmations as u64) < burnchain_height;
        if ibd {
            debug!(
                "PoX watchdog: {last_processed_height} + {} < {burnchain_height}, so initial block download",
                burnchain.stable_confirmations
            );
        } else {
            debug!(
                "PoX watchdog: {last_processed_height} + {} >= {burnchain_height}, so steady-state",
                burnchain.stable_confirmations
            );
        }
        ibd
    }

    /// Wait until the next PoX anchor block arrives.
    /// We know for a fact that they all exist for Epochs 2.5 and earlier, in both mainnet and
    /// testnet.
    /// Return (still-in-ibd?, maximum-burnchain-sync-height) on success.
    pub fn pox_sync_wait(
        &mut self,
        burnchain: &Burnchain,
        burnchain_tip: &BurnchainTip, // this is the highest burnchain snapshot we've sync'ed to
        burnchain_height: u64,        // this is the absolute burnchain block height
    ) -> Result<(bool, u64), burnchain_error> {
        let burnchain_rc = burnchain
            .block_height_to_reward_cycle(burnchain_height)
            .expect("FATAL: burnchain height is before system start");

        let sortition_rc = burnchain
            .block_height_to_reward_cycle(burnchain_tip.block_snapshot.block_height)
            .expect("FATAL: sortition height is before system start");

        let ibbd = PoxSyncWatchdog::infer_initial_burnchain_block_download(
            burnchain,
            burnchain_tip.block_snapshot.block_height,
            burnchain_height,
        );

        let max_sync_height = if sortition_rc < burnchain_rc {
            burnchain
                .reward_cycle_to_block_height(sortition_rc + 1)
                .min(burnchain_height)
        } else {
            burnchain_tip
                .block_snapshot
                .block_height
                .max(burnchain_height)
        };

        self.relayer_comms.set_ibd(ibbd);
        if !self.unconditionally_download {
            self.relayer_comms
                .interruptable_sleep(self.steady_state_burnchain_sync_interval)?;
        }

        Ok((ibbd, max_sync_height))
    }
}

1	// Copyright (C) 2013-2020 Blockstack PBC, a public benefit corporation
2	// Copyright (C) 2020-2024 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
17	use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
18	use std::sync::Arc;
19
20	use stacks::burnchains::{Burnchain, Error as burnchain_error};
21	use stacks_common::util::{get_epoch_time_secs, sleep_ms};
22
23	use crate::burnchains::BurnchainTip;
24	use crate::Config;
25
26	#[derive(Clone)]
27	pub struct PoxSyncWatchdogComms {
28	/// how many passes in the p2p state machine have taken place since startup?
29	p2p_state_passes: Arc<AtomicU64>,
30	/// how many times have we done an inv sync?
31	inv_sync_passes: Arc<AtomicU64>,
32	/// how many times have we done a download pass?
33	download_passes: Arc<AtomicU64>,
34	/// What's our last IBD status?
35	last_ibd: Arc<AtomicBool>,
36	/// Should keep running?
37	should_keep_running: Arc<AtomicBool>,
38	}
39
40	impl PoxSyncWatchdogComms {
41	pub fn new(should_keep_running: Arc<AtomicBool>) -> PoxSyncWatchdogComms {	492✔
42	PoxSyncWatchdogComms {	492✔
43	p2p_state_passes: Arc::new(AtomicU64::new(0)),	492✔
44	inv_sync_passes: Arc::new(AtomicU64::new(0)),	492✔
45	download_passes: Arc::new(AtomicU64::new(0)),	492✔
46	last_ibd: Arc::new(AtomicBool::new(true)),	492✔
47	should_keep_running,	492✔
48	}	492✔
49	}	492✔
50
51	pub fn get_p2p_state_passes(&self) -> u64 {	×
52	self.p2p_state_passes.load(Ordering::SeqCst)	×
53	}	×
54
55	pub fn get_inv_sync_passes(&self) -> u64 {	×
56	self.inv_sync_passes.load(Ordering::SeqCst)	×
57	}	×
58
59	pub fn get_download_passes(&self) -> u64 {	4✔
60	self.download_passes.load(Ordering::SeqCst)	4✔
61	}	4✔
62
63	pub fn get_ibd(&self) -> bool {	1,751,711✔
64	self.last_ibd.load(Ordering::SeqCst)	1,751,711✔
65	}	1,751,711✔
66
67	fn interruptable_sleep(&self, secs: u64) -> Result<(), burnchain_error> {	169,023✔
68	let deadline = secs + get_epoch_time_secs();	169,023✔
69	while get_epoch_time_secs() < deadline {	169,023✔
70	sleep_ms(1000);	×
71	if !self.should_keep_running() {	×
72	return Err(burnchain_error::CoordinatorClosed);	×
73	}	×
74	}
75	Ok(())	169,023✔
76	}	169,023✔
77
78	pub fn should_keep_running(&self) -> bool {	×
79	self.should_keep_running.load(Ordering::SeqCst)	×
80	}	×
81
82	pub fn notify_p2p_state_pass(&mut self) {	781,945✔
83	self.p2p_state_passes.fetch_add(1, Ordering::SeqCst);	781,945✔
84	}	781,945✔
85
86	pub fn notify_inv_sync_pass(&mut self) {	486,848✔
87	self.inv_sync_passes.fetch_add(1, Ordering::SeqCst);	486,848✔
88	}	486,848✔
89
90	pub fn notify_download_pass(&mut self) {	113,750✔
91	self.download_passes.fetch_add(1, Ordering::SeqCst);	113,750✔
92	}	113,750✔
93
94	pub fn set_ibd(&mut self, value: bool) {	1,233,128✔
95	self.last_ibd.store(value, Ordering::SeqCst);	1,233,128✔
96	}	1,233,128✔
97	}
98
99	/// Monitor the state of the Stacks blockchain as the peer network and relay threads download and
100	/// proces Stacks blocks. Don't allow the node to process the next PoX reward cycle's sortitions
101	/// unless it's reasonably sure that it has processed all Stacks blocks for this reward cycle.
102	/// This struct monitors the Stacks chainstate to make this determination.
103	pub struct PoxSyncWatchdog {
104	/// time between burnchain syncs in steady state
105	steady_state_burnchain_sync_interval: u64,
106	/// handle to relayer thread that informs the watchdog when the P2P state-machine does stuff
107	relayer_comms: PoxSyncWatchdogComms,
108	/// should this sync watchdog always download? used in integration tests.
109	unconditionally_download: bool,
110	}
111
112	impl PoxSyncWatchdog {
113	pub fn new(	262✔
114	config: &Config,	262✔
115	watchdog_comms: PoxSyncWatchdogComms,	262✔
116	) -> Result<PoxSyncWatchdog, String> {	262✔
117	let burnchain_poll_time = config.burnchain.poll_time_secs;	262✔
118	let unconditionally_download = config.node.pox_sync_sample_secs == 0;	262✔
119
120	Ok(PoxSyncWatchdog {	262✔
121	unconditionally_download,	262✔
122	steady_state_burnchain_sync_interval: burnchain_poll_time,	262✔
123	relayer_comms: watchdog_comms,	262✔
124	})	262✔
125	}	262✔
126
127	pub fn make_comms_handle(&self) -> PoxSyncWatchdogComms {	×
128	self.relayer_comms.clone()	×
129	}	×
130
131	/// Are we in the initial burnchain block download? i.e. is the burn tip snapshot far enough away
132	/// from the burnchain height that we should be eagerly downloading snapshots?
133	fn infer_initial_burnchain_block_download(	523,669✔
134	burnchain: &Burnchain,	523,669✔
135	last_processed_height: u64,	523,669✔
136	burnchain_height: u64,	523,669✔
137	) -> bool {	523,669✔
138	let ibd =	523,669✔
139	last_processed_height + (burnchain.stable_confirmations as u64) < burnchain_height;	523,669✔
140	if ibd {	523,669✔
141	debug!(	3,370✔
142	"PoX watchdog: {last_processed_height} + {} < {burnchain_height}, so initial block download",
143	burnchain.stable_confirmations
144	);
145	} else {
146	debug!(	520,299✔
147	"PoX watchdog: {last_processed_height} + {} >= {burnchain_height}, so steady-state",
148	burnchain.stable_confirmations
149	);
150	}
151	ibd	523,669✔
152	}	523,669✔
153
154	/// Wait until the next PoX anchor block arrives.
155	/// We know for a fact that they all exist for Epochs 2.5 and earlier, in both mainnet and
156	/// testnet.
157	/// Return (still-in-ibd?, maximum-burnchain-sync-height) on success.
158	pub fn pox_sync_wait(	523,669✔
159	&mut self,	523,669✔
160	burnchain: &Burnchain,	523,669✔
161	burnchain_tip: &BurnchainTip, // this is the highest burnchain snapshot we've sync'ed to	523,669✔
162	burnchain_height: u64, // this is the absolute burnchain block height	523,669✔
163	) -> Result<(bool, u64), burnchain_error> {	523,669✔
164	let burnchain_rc = burnchain	523,669✔
165	.block_height_to_reward_cycle(burnchain_height)	523,669✔
166	.expect("FATAL: burnchain height is before system start");	523,669✔
167
168	let sortition_rc = burnchain	523,669✔
169	.block_height_to_reward_cycle(burnchain_tip.block_snapshot.block_height)	523,669✔
170	.expect("FATAL: sortition height is before system start");	523,669✔
171
172	let ibbd = PoxSyncWatchdog::infer_initial_burnchain_block_download(	523,669✔
173	burnchain,	523,669✔
174	burnchain_tip.block_snapshot.block_height,	523,669✔
175	burnchain_height,	523,669✔
176	);
177
178	let max_sync_height = if sortition_rc < burnchain_rc {	523,669✔
179	burnchain	3,596✔
180	.reward_cycle_to_block_height(sortition_rc + 1)	3,596✔
181	.min(burnchain_height)	3,596✔
182	} else {
183	burnchain_tip	520,073✔
184	.block_snapshot	520,073✔
185	.block_height	520,073✔
186	.max(burnchain_height)	520,073✔
187	};
188
189	self.relayer_comms.set_ibd(ibbd);	523,669✔
190	if !self.unconditionally_download {	523,669✔
191	self.relayer_comms	169,023✔
192	.interruptable_sleep(self.steady_state_burnchain_sync_interval)?;	169,023✔
193	}	354,646✔
194
195	Ok((ibbd, max_sync_height))	523,669✔
196	}	523,669✔
197	}

stacks-network / stacks-core / 24264100700

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