Send Signer results over a channel

libsigner/src/runloop.rs

@@ -19,7 +19,7 @@
 
 use std::marker::PhantomData;
 use std::net::SocketAddr;
-use std::sync::mpsc::{channel, Receiver, RecvTimeoutError};
+use std::sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender};
 use std::thread;
 use std::thread::JoinHandle;
 use std::time::Duration;
@@ -41,15 +41,20 @@ 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, CMD: Send> {
+pub trait SignerRunLoop<R: Send, CMD: Send> {
     /// 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 StackerDB 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<StackerDBChunksEvent>, cmd: Option<CMD>) -> Option<R>;
+    fn run_one_pass(
+        &mut self,
+        event: Option<StackerDBChunksEvent>,
+        cmd: Option<CMD>,
+        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
@@ -62,6 +67,7 @@ pub trait SignerRunLoop<R, CMD: Send> {
         &mut self,
         event_recv: Receiver<StackerDBChunksEvent>,
         command_recv: Receiver<CMD>,
+        result_send: Sender<R>,
         mut event_stop_signaler: EVST,
     ) -> Option<R> {
         loop {
@@ -82,7 +88,9 @@ pub trait SignerRunLoop<R, CMD: Send> {
                     return None;
                 }
             };
-            if let Some(final_state) = self.run_one_pass(next_event_opt, next_command_opt) {
+            if let Some(final_state) =
+                self.run_one_pass(next_event_opt, next_command_opt, result_send.clone())
+            {
                 info!("Runloop exit; signaling event-receiver to stop");
                 event_stop_signaler.send();
                 return Some(final_state);
@@ -92,13 +100,20 @@ pub trait SignerRunLoop<R, CMD: Send> {
 }
 
 /// The top-level signer implementation
-pub struct Signer<CMD: Send, R, SL: SignerRunLoop<R, CMD> + Send + Sync, EV: EventReceiver + Send> {
+pub struct Signer<
+    CMD: Send,
+    R: Send,
+    SL: SignerRunLoop<R, CMD> + Send + Sync,
+    EV: EventReceiver + Send,
+> {
     /// the runloop itself
     signer_loop: Option<SL>,
     /// the event receiver to use
     event_receiver: Option<EV>,
     /// the command receiver to use
     command_receiver: Option<Receiver<CMD>>,
+    /// the result sender to use
+    result_sender: Option<Sender<R>>,
     /// marker to permit the R type
     _phantom: PhantomData<R>,
 }
@@ -193,11 +208,13 @@ impl<
         runloop: SL,
         event_receiver: EV,
         command_receiver: Receiver<CMD>,
+        result_sender: Sender<R>,
     ) -> Signer<CMD, R, SL, EV> {
         Signer {
             signer_loop: Some(runloop),
             event_receiver: Some(event_receiver),
             command_receiver: Some(command_receiver),
+            result_sender: Some(result_sender),
             _phantom: PhantomData,
         }
     }
@@ -221,6 +238,10 @@ impl<
             .command_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();
@@ -244,7 +265,9 @@ impl<
         let runloop_thread = thread::Builder::new()
             .name("signer_runloop".to_string())
             .stack_size(THREAD_STACK_SIZE)
-            .spawn(move || signer_loop.main_loop(event_recv, command_receiver, stop_signaler))
+            .spawn(move || {
+                signer_loop.main_loop(event_recv, command_receiver, result_sender, stop_signaler)
+            })
             .map_err(|e| {
                 error!("SignerRunLoop failed to start: {:?}", &e);
                 ret_stop_signaler.send();

libsigner/src/tests/mod.rs

@@ -19,7 +19,7 @@ mod http;
 use std::io::Write;
 use std::mem;
 use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
-use std::sync::mpsc::{channel, Receiver};
+use std::sync::mpsc::{channel, Receiver, Sender};
 use std::thread;
 use std::time::Duration;
 
@@ -69,6 +69,7 @@ impl SignerRunLoop<Vec<StackerDBChunksEvent>, Command> for SimpleRunLoop {
         &mut self,
         event: Option<StackerDBChunksEvent>,
         _cmd: Option<Command>,
+        _res: Sender<Vec<StackerDBChunksEvent>>,
     ) -> Option<Vec<StackerDBChunksEvent>> {
         debug!("Got event: {:?}", &event);
         if let Some(event) = event {
@@ -94,7 +95,8 @@ fn test_simple_signer() {
     )
     .unwrap()]);
     let (_cmd_send, cmd_recv) = channel();
-    let mut signer = Signer::new(SimpleRunLoop::new(5), ev, cmd_recv);
+    let (res_send, _res_recv) = channel();
+    let mut signer = Signer::new(SimpleRunLoop::new(5), ev, cmd_recv, res_send);
     let endpoint: SocketAddr = "127.0.0.1:30000".parse().unwrap();
     let thread_endpoint = endpoint.clone();
 

stacks-signer/src/crypto/frost.rs

@@ -627,6 +627,15 @@ impl Coordinatable for Coordinator {
         let message = self.start_signing_round()?;
         Ok(message)
     }
+
+    // Reset internal state
+    fn reset(&mut self) {
+        self.state = State::Idle;
+        self.dkg_public_shares.clear();
+        self.public_nonces.clear();
+        self.signature_shares.clear();
+        self.ids_to_await = (0..self.total_signers).collect();
+    }
 }
 
 #[cfg(test)]
@@ -642,7 +651,7 @@ mod test {
 
     #[test]
     fn test_state_machine() {
-        let mut rng = OsRng::default();
+        let mut rng = OsRng;
         let message_private_key = Scalar::random(&mut rng);
 
         let mut coordinator = Coordinator::new(3, 3, 3, message_private_key);
@@ -694,7 +703,7 @@ mod test {
         let total_signers = 10;
         let total_keys = 40;
         let threshold = 28;
-        let mut rng = OsRng::default();
+        let mut rng = OsRng;
         let message_private_key = Scalar::random(&mut rng);
 
         let coordinator =
@@ -713,18 +722,15 @@ mod test {
         let total_signers = 10;
         let total_keys = 40;
         let threshold = 28;
-        let mut rng = OsRng::default();
+        let mut rng = OsRng;
         let message_private_key = Scalar::random(&mut rng);
         let mut coordinator =
             Coordinator::new(total_signers, total_keys, threshold, message_private_key);
 
         let result = coordinator.start_dkg_round();
 
         assert!(result.is_ok());
-        assert!(match result.unwrap().msg {
-            MessageTypes::DkgBegin(_) => true,
-            _ => false,
-        });
+        assert!(matches!(result.unwrap().msg, MessageTypes::DkgBegin(_)));
         assert_eq!(coordinator.state, State::DkgPublicGather);
         assert_eq!(coordinator.current_dkg_id, 1);
     }
@@ -734,18 +740,15 @@ mod test {
         let total_signers = 10;
         let total_keys = 40;
         let threshold = 28;
-        let mut rng = OsRng::default();
+        let mut rng = OsRng;
         let message_private_key = Scalar::random(&mut rng);
         let mut coordinator =
             Coordinator::new(total_signers, total_keys, threshold, message_private_key);
         coordinator.state = State::DkgPublicDistribute; // Must be in this state before calling start public shares
 
         let result = coordinator.start_public_shares().unwrap();
 
-        assert!(match result.msg {
-            MessageTypes::DkgBegin(_) => true,
-            _ => false,
-        });
+        assert!(matches!(result.msg, MessageTypes::DkgBegin(_)));
         assert_eq!(coordinator.state, State::DkgPublicGather);
         assert_eq!(coordinator.current_dkg_id, 0);
     }
@@ -755,23 +758,20 @@ mod test {
         let total_signers = 10;
         let total_keys = 40;
         let threshold = 28;
-        let mut rng = OsRng::default();
+        let mut rng = OsRng;
         let message_private_key = Scalar::random(&mut rng);
         let mut coordinator =
             Coordinator::new(total_signers, total_keys, threshold, message_private_key);
         coordinator.state = State::DkgPrivateDistribute; // Must be in this state before calling start private shares
 
         let message = coordinator.start_private_shares().unwrap();
-        assert!(match message.msg {
-            MessageTypes::DkgPrivateBegin(_) => true,
-            _ => false,
-        });
+        assert!(matches!(message.msg, MessageTypes::DkgPrivateBegin(_)));
         assert_eq!(coordinator.state, State::DkgEndGather);
         assert_eq!(coordinator.current_dkg_id, 0);
     }
 
     fn setup() -> (Coordinator, Vec<SigningRound>) {
-        let mut rng = OsRng::default();
+        let mut rng = OsRng;
         let total_signers = 5;
         let threshold = total_signers / 10 + 7;
         let keys_per_signer = 3;
@@ -789,11 +789,11 @@ mod test {
         let mut key_ids = Vec::new();
         for (i, (_private_key, public_key)) in key_pairs.iter().enumerate() {
             for _ in 0..keys_per_signer {
-                key_ids_map.insert((key_id + 1) as u32, public_key.clone());
+                key_ids_map.insert(key_id + 1, *public_key);
                 key_ids.push(key_id);
                 key_id += 1;
             }
-            signer_ids_map.insert(i as u32, public_key.clone());
+            signer_ids_map.insert(i as u32, *public_key);
         }
         let public_keys = PublicKeys {
             signers: signer_ids_map,
@@ -839,10 +839,9 @@ mod test {
                 process_inbound_messages(signing_round, feedback_messages.clone()).unwrap();
             inbound_messages.extend(outbound_messages);
         }
-        let outbound_messages = coordinator
+        coordinator
             .process_inbound_messages(inbound_messages)
-            .unwrap();
-        outbound_messages
+            .unwrap()
     }
 
     #[test]

stacks-signer/src/crypto/mod.rs

@@ -35,4 +35,6 @@ pub trait Coordinatable {
     fn start_distributed_key_generation(&mut self) -> Result<Message, Error>;
     /// Trigger a signing round
     fn start_signing_message(&mut self, _message: &[u8]) -> Result<Message, Error>;
+    /// Reset internal state
+    fn reset(&mut self);
 }