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

70.48

/libsigner/src/runloop.rs

// 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/>.

#![allow(dead_code)]

use std::marker::PhantomData;
use std::net::SocketAddr;
use std::sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender};
use std::thread;
use std::thread::JoinHandle;
use std::time::Duration;

use stacks_common::deps_common::ctrlc as termination;
use stacks_common::deps_common::ctrlc::SignalId;

use crate::error::EventError;
use crate::events::{EventReceiver, EventStopSignaler, SignerEvent, SignerEventTrait};

/// Some libcs, like musl, have a very small stack size.
/// Make sure it's big enough.
const THREAD_STACK_SIZE: usize = 128 * 1024 * 1024; // 128 MB

/// stderr fileno
const STDERR: i32 = 2;

/// Trait describing the needful components of a top-level runloop.
/// This is where the signer business logic would go.
/// Implement this, and you get all the multithreaded setup for free.
pub trait SignerRunLoop<R: Send, T: SignerEventTrait> {
    /// Hint to set how long to wait for new events
    fn set_event_timeout(&mut self, timeout: Duration);
    /// Getter for the event poll timeout
    fn get_event_timeout(&self) -> Duration;
    /// Run one pass of the event loop, given new Signer events discovered since the last pass.
    /// Returns Some(R) if this is the final pass -- the runloop evaluated to R
    /// Returns None to keep running.
    fn run_one_pass(&mut self, event: Option<SignerEvent<T>>, res: &Sender<R>) -> Option<R>;

    /// This is the main loop body for the signer. It continuously receives events from
    /// `event_recv`, polling for up to `self.get_event_timeout()` units of time.  Once it has
    /// polled for events, they are fed into `run_one_pass()`.  This continues until either
    /// `run_one_pass()` returns `false`, or the event receiver hangs up.  At this point, this
    /// method calls the `event_stop_signaler.send()` to terminate the receiver.
    ///
    /// This would run in a separate thread from the event receiver.
    fn main_loop<EVST: EventStopSignaler>(
        &mut self,
        event_recv: Receiver<SignerEvent<T>>,
        result_send: Sender<R>,
        mut event_stop_signaler: EVST,
    ) -> Option<R> {
        info!("Signer runloop begin");
        loop {
            let poll_timeout = self.get_event_timeout();
            let next_event_opt = match event_recv.recv_timeout(poll_timeout) {
                Ok(event) => Some(event),
                Err(RecvTimeoutError::Timeout) => None,
                Err(RecvTimeoutError::Disconnected) => {
                    info!("Event receiver disconnected");
                    return None;
                }
            };
            if let Some(final_state) = self.run_one_pass(next_event_opt, &result_send) {
                info!("Runloop exit; signaling event-receiver to stop");
                event_stop_signaler.send();
                return Some(final_state);
            }
        }
    }
}

/// The top-level signer implementation
pub struct Signer<R, SL, EV, T> {
    /// the runloop itself
    signer_loop: Option<SL>,
    /// the event receiver to use
    event_receiver: Option<EV>,
    /// the result sender to use
    result_sender: Option<Sender<R>>,
    /// phantom data for the codec
    phantom_data: PhantomData<T>,
}

/// The running signer implementation
pub struct RunningSigner<EV: EventReceiver<T>, R, T: SignerEventTrait> {
    /// join handle for signer runloop
    signer_join: JoinHandle<Option<R>>,
    /// join handle for event receiver
    event_join: JoinHandle<()>,
    /// kill signal for the event receiver
    stop_signal: EV::ST,
}

impl<EV: EventReceiver<T>, R, T: SignerEventTrait> RunningSigner<EV, R, T> {
    /// Stop the signer, and get the final state
    pub fn stop(mut self) -> Option<R> {
        // kill event receiver
        self.stop_signal.send();

        self.join()
    }

    /// Wait for the signer to terminate, and get the final state.
    /// WARNING: This will hang forever if the event receiver stop signal was never sent/no error occurs.
    pub fn join(self) -> Option<R> {
        debug!("Try join event loop...");
        // wait for event receiver join
        let _ = self.event_join.join().inspect_err(|thread_panic| {
            error!("Event thread panicked with: '{:?}'", &thread_panic);
        });
        info!("Event receiver thread joined");

        // wait for runloop to join
        debug!("Try join signer loop...");
        let result_opt = self
            .signer_join
            .join()
            .inspect_err(|thread_panic| {
                error!("Event thread panicked with: '{:?}'", &thread_panic);
            })
            .unwrap_or(None);

        info!("Signer thread joined");
        result_opt
    }
}

/// Write to stderr in an async-safe manner.
/// See signal-safety(7)
#[warn(unused)]
fn async_safe_write_stderr(msg: &str) {
    #[cfg(windows)]
    unsafe {
        // write(2) inexplicably has a different ABI only on Windows.
        libc::write(
            STDERR,
            msg.as_ptr() as *const libc::c_void,
            msg.len() as u32,
        );
    }
    #[cfg(not(windows))]
    unsafe {
        libc::write(STDERR, msg.as_ptr() as *const libc::c_void, msg.len());
    }
}

/// This is a termination handler for handling receipt of a terminating UNIX signal, like SIGINT,
/// SIGQUIT, SIGTERM, or SIGBUS.  You'd call this as part of the startup code for the signer daemon.
/// DO NOT call this from within the library!
pub fn set_runloop_signal_handler<ST: EventStopSignaler + Send + 'static>(mut stop_signaler: ST) {
    termination::set_handler(move |sig_id| match sig_id {
        SignalId::Bus => {
            let msg = "Caught SIGBUS; crashing immediately and dumping core\n";
            async_safe_write_stderr(msg);
            unsafe {
                libc::abort();
            }
        }
        _ => {
            let msg = format!("Graceful termination request received (signal `{sig_id}`), will complete the ongoing runloop cycles and terminate\n");
            async_safe_write_stderr(&msg);
            stop_signaler.send();
        }
    }).expect("FATAL: failed to set signal handler");
}

impl<R, SL, EV, T> Signer<R, SL, EV, T> {
    /// Create a new signer with the given runloop and event receiver.
    pub fn new(runloop: SL, event_receiver: EV, result_sender: Sender<R>) -> Signer<R, SL, EV, T> {
        Signer {
            signer_loop: Some(runloop),
            event_receiver: Some(event_receiver),
            result_sender: Some(result_sender),
            phantom_data: PhantomData,
        }
    }
}

impl<
        R: Send + 'static,
        T: SignerEventTrait + 'static,
        SL: SignerRunLoop<R, T> + Send + 'static,
        EV: EventReceiver<T> + Send + 'static,
    > Signer<R, SL, EV, T>
{
    /// This is a helper function to spawn both the runloop and event receiver in their own
    /// threads.  Advanced signers may not need this method, and instead opt to run the receiver
    /// and runloop directly.  However, this method is present to help signer developers to get
    /// their implementations off the ground.
    ///
    /// The given `bind_addr` is the server address this event receiver needs to listen on, so the
    /// stacks node can POST events to it.
    ///
    /// On success, this method consumes the Signer and returns a RunningSigner with the relevant
    /// inter-thread communication primitives for the caller to shut down the system.
    pub fn spawn(&mut self, bind_addr: SocketAddr) -> Result<RunningSigner<EV, R, T>, EventError> {
        let mut event_receiver = self
            .event_receiver
            .take()
            .ok_or(EventError::AlreadyRunning)?;
        let result_sender = self
            .result_sender
            .take()
            .ok_or(EventError::AlreadyRunning)?;
        let mut signer_loop = self.signer_loop.take().ok_or(EventError::AlreadyRunning)?;

        let (event_send, event_recv) = channel();
        event_receiver.add_consumer(event_send);

        let bind_port = bind_addr.port();
        event_receiver.bind(bind_addr)?;
        let stop_signaler = event_receiver.get_stop_signaler()?;
        let mut ret_stop_signaler = event_receiver.get_stop_signaler()?;

        // start a thread for the event receiver
        let event_thread = thread::Builder::new()
            .name(format!("event_receiver:{bind_port}"))
            .stack_size(THREAD_STACK_SIZE)
            .spawn(move || event_receiver.main_loop())
            .map_err(|e| {
                error!("EventReceiver failed to start: {:?}", &e);
                EventError::FailedToStart
            })?;

        // start receiving events and doing stuff with them
        let runloop_thread = thread::Builder::new()
            .name(format!("signer_runloop:{bind_port}"))
            .stack_size(THREAD_STACK_SIZE)
            .spawn(move || signer_loop.main_loop(event_recv, result_sender, stop_signaler))
            .map_err(|e| {
                error!("SignerRunLoop failed to start: {:?}", &e);
                ret_stop_signaler.send();
                EventError::FailedToStart
            })?;

        let running_signer = RunningSigner {
            signer_join: runloop_thread,
            event_join: event_thread,
            stop_signal: ret_stop_signaler,
        };

        Ok(running_signer)
    }
}

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	#![allow(dead_code)]
18
19	use std::marker::PhantomData;
20	use std::net::SocketAddr;
21	use std::sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender};
22	use std::thread;
23	use std::thread::JoinHandle;
24	use std::time::Duration;
25
26	use stacks_common::deps_common::ctrlc as termination;
27	use stacks_common::deps_common::ctrlc::SignalId;
28
29	use crate::error::EventError;
30	use crate::events::{EventReceiver, EventStopSignaler, SignerEvent, SignerEventTrait};
31
32	/// Some libcs, like musl, have a very small stack size.
33	/// Make sure it's big enough.
34	const THREAD_STACK_SIZE: usize = 128 * 1024 * 1024; // 128 MB
35
36	/// stderr fileno
37	const STDERR: i32 = 2;
38
39	/// Trait describing the needful components of a top-level runloop.
40	/// This is where the signer business logic would go.
41	/// Implement this, and you get all the multithreaded setup for free.
42	pub trait SignerRunLoop<R: Send, T: SignerEventTrait> {
43	/// Hint to set how long to wait for new events
44	fn set_event_timeout(&mut self, timeout: Duration);
45	/// Getter for the event poll timeout
46	fn get_event_timeout(&self) -> Duration;
47	/// Run one pass of the event loop, given new Signer events discovered since the last pass.
48	/// Returns Some(R) if this is the final pass -- the runloop evaluated to R
49	/// Returns None to keep running.
50	fn run_one_pass(&mut self, event: Option<SignerEvent<T>>, res: &Sender<R>) -> Option<R>;
51
52	/// This is the main loop body for the signer. It continuously receives events from
53	/// `event_recv`, polling for up to `self.get_event_timeout()` units of time. Once it has
54	/// polled for events, they are fed into `run_one_pass()`. This continues until either
55	/// `run_one_pass()` returns `false`, or the event receiver hangs up. At this point, this
56	/// method calls the `event_stop_signaler.send()` to terminate the receiver.
57	///
58	/// This would run in a separate thread from the event receiver.
59	fn main_loop<EVST: EventStopSignaler>(	2✔
60	&mut self,	2✔
61	event_recv: Receiver<SignerEvent<T>>,	2✔
62	result_send: Sender<R>,	2✔
63	mut event_stop_signaler: EVST,	2✔
64	) -> Option<R> {	2✔
65	info!("Signer runloop begin");	2✔
66	loop {
67	let poll_timeout = self.get_event_timeout();	6✔
68	let next_event_opt = match event_recv.recv_timeout(poll_timeout) {	6✔
69	Ok(event) => Some(event),	6✔
UNCOV 70	Err(RecvTimeoutError::Timeout) => None,	×
71	Err(RecvTimeoutError::Disconnected) => {
UNCOV 72	info!("Event receiver disconnected");	×
UNCOV 73	return None;	×
74	}
75	};
76	if let Some(final_state) = self.run_one_pass(next_event_opt, &result_send) {	6✔
77	info!("Runloop exit; signaling event-receiver to stop");	2✔
78	event_stop_signaler.send();	2✔
79	return Some(final_state);	2✔
80	}	4✔
81	}
82	}	2✔
83	}
84
85	/// The top-level signer implementation
86	pub struct Signer<R, SL, EV, T> {
87	/// the runloop itself
88	signer_loop: Option<SL>,
89	/// the event receiver to use
90	event_receiver: Option<EV>,
91	/// the result sender to use
92	result_sender: Option<Sender<R>>,
93	/// phantom data for the codec
94	phantom_data: PhantomData<T>,
95	}
96
97	/// The running signer implementation
98	pub struct RunningSigner<EV: EventReceiver<T>, R, T: SignerEventTrait> {
99	/// join handle for signer runloop
100	signer_join: JoinHandle<Option<R>>,
101	/// join handle for event receiver
102	event_join: JoinHandle<()>,
103	/// kill signal for the event receiver
104	stop_signal: EV::ST,
105	}
106
107	impl<EV: EventReceiver<T>, R, T: SignerEventTrait> RunningSigner<EV, R, T> {
108	/// Stop the signer, and get the final state
109	pub fn stop(mut self) -> Option<R> {	2✔
110	// kill event receiver
111	self.stop_signal.send();	2✔
112
113	self.join()	2✔
114	}	2✔
115
116	/// Wait for the signer to terminate, and get the final state.
117	/// WARNING: This will hang forever if the event receiver stop signal was never sent/no error occurs.
118	pub fn join(self) -> Option<R> {	2✔
119	debug!("Try join event loop...");	2✔
120	// wait for event receiver join
121	let _ = self.event_join.join().inspect_err(\|thread_panic\| {	2✔
122	error!("Event thread panicked with: '{:?}'", &thread_panic);	×
123	});	×
124	info!("Event receiver thread joined");	2✔
125
126	// wait for runloop to join
127	debug!("Try join signer loop...");	2✔
128	let result_opt = self	2✔
129	.signer_join	2✔
130	.join()	2✔
131	.inspect_err(\|thread_panic\| {	2✔
132	error!("Event thread panicked with: '{:?}'", &thread_panic);	×
133	})	×
134	.unwrap_or(None);	2✔
135
136	info!("Signer thread joined");	2✔
137	result_opt	2✔
138	}	2✔
139	}
140
141	/// Write to stderr in an async-safe manner.
142	/// See signal-safety(7)
143	#[warn(unused)]
144	fn async_safe_write_stderr(msg: &str) {	×
145	#[cfg(windows)]
146	unsafe {
147	// write(2) inexplicably has a different ABI only on Windows.
148	libc::write(
149	STDERR,
150	msg.as_ptr() as *const libc::c_void,
151	msg.len() as u32,
152	);
153	}
154	#[cfg(not(windows))]
155	unsafe {	×
156	libc::write(STDERR, msg.as_ptr() as *const libc::c_void, msg.len());	×
157	}	×
158	}	×
159
160	/// This is a termination handler for handling receipt of a terminating UNIX signal, like SIGINT,
161	/// SIGQUIT, SIGTERM, or SIGBUS. You'd call this as part of the startup code for the signer daemon.
162	/// DO NOT call this from within the library!
163	pub fn set_runloop_signal_handler<ST: EventStopSignaler + Send + 'static>(mut stop_signaler: ST) {	×
164	termination::set_handler(move \|sig_id\| match sig_id {	×
165	SignalId::Bus => {
166	let msg = "Caught SIGBUS; crashing immediately and dumping core\n";	×
167	async_safe_write_stderr(msg);	×
168	unsafe {
169	libc::abort();	×
170	}
171	}
172	_ => {	×
173	let msg = format!("Graceful termination request received (signal `{sig_id}`), will complete the ongoing runloop cycles and terminate\n");	×
174	async_safe_write_stderr(&msg);	×
175	stop_signaler.send();	×
176	}	×
177	}).expect("FATAL: failed to set signal handler");	×
178	}	×
179
180	impl<R, SL, EV, T> Signer<R, SL, EV, T> {
181	/// Create a new signer with the given runloop and event receiver.
182	pub fn new(runloop: SL, event_receiver: EV, result_sender: Sender<R>) -> Signer<R, SL, EV, T> {	2✔
183	Signer {	2✔
184	signer_loop: Some(runloop),	2✔
185	event_receiver: Some(event_receiver),	2✔
186	result_sender: Some(result_sender),	2✔
187	phantom_data: PhantomData,	2✔
188	}	2✔
189	}	2✔
190	}
191
192	impl<
193	R: Send + 'static,
194	T: SignerEventTrait + 'static,
195	SL: SignerRunLoop<R, T> + Send + 'static,
196	EV: EventReceiver<T> + Send + 'static,
197	> Signer<R, SL, EV, T>
198	{
199	/// This is a helper function to spawn both the runloop and event receiver in their own
200	/// threads. Advanced signers may not need this method, and instead opt to run the receiver
201	/// and runloop directly. However, this method is present to help signer developers to get
202	/// their implementations off the ground.
203	///
204	/// The given `bind_addr` is the server address this event receiver needs to listen on, so the
205	/// stacks node can POST events to it.
206	///
207	/// On success, this method consumes the Signer and returns a RunningSigner with the relevant
208	/// inter-thread communication primitives for the caller to shut down the system.
209	pub fn spawn(&mut self, bind_addr: SocketAddr) -> Result<RunningSigner<EV, R, T>, EventError> {	2✔
210	let mut event_receiver = self	2✔
211	.event_receiver	2✔
212	.take()	2✔
213	.ok_or(EventError::AlreadyRunning)?;	2✔
214	let result_sender = self	2✔
215	.result_sender	2✔
216	.take()	2✔
217	.ok_or(EventError::AlreadyRunning)?;	2✔
218	let mut signer_loop = self.signer_loop.take().ok_or(EventError::AlreadyRunning)?;	2✔
219
220	let (event_send, event_recv) = channel();	2✔
221	event_receiver.add_consumer(event_send);	2✔
222
223	let bind_port = bind_addr.port();	2✔
224	event_receiver.bind(bind_addr)?;	2✔
225	let stop_signaler = event_receiver.get_stop_signaler()?;	2✔
226	let mut ret_stop_signaler = event_receiver.get_stop_signaler()?;	2✔
227
228	// start a thread for the event receiver
229	let event_thread = thread::Builder::new()	2✔
230	.name(format!("event_receiver:{bind_port}"))	2✔
231	.stack_size(THREAD_STACK_SIZE)	2✔
232	.spawn(move \|\| event_receiver.main_loop())	2✔
233	.map_err(\|e\| {	2✔
234	error!("EventReceiver failed to start: {:?}", &e);	×
235	EventError::FailedToStart	×
236	})?;	×
237
238	// start receiving events and doing stuff with them
239	let runloop_thread = thread::Builder::new()	2✔
240	.name(format!("signer_runloop:{bind_port}"))	2✔
241	.stack_size(THREAD_STACK_SIZE)	2✔
242	.spawn(move \|\| signer_loop.main_loop(event_recv, result_sender, stop_signaler))	2✔
243	.map_err(\|e\| {	2✔
244	error!("SignerRunLoop failed to start: {:?}", &e);	×
245	ret_stop_signaler.send();	×
246	EventError::FailedToStart	×
247	})?;	×
248
249	let running_signer = RunningSigner {	2✔
250	signer_join: runloop_thread,	2✔
251	event_join: event_thread,	2✔
252	stop_signal: ret_stop_signaler,	2✔
253	};	2✔
254
255	Ok(running_signer)	2✔
256	}	2✔
257	}

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