Merge pull request #26 from sideprotocol/refactor-config-loading

Optimize task processing efficiency

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "shuttler"
-version = "0.2.1"
+version = "0.2.6"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

src/app/config.rs

@@ -28,7 +28,6 @@ pub struct Config {
     pub home: PathBuf,
     pub p2p_keypair: String,
     pub port: u32,
-    pub seed_mode: bool,
     pub bootstrap_nodes: Vec<String>,
     /// logger level
     pub log_level: String,
@@ -50,7 +49,7 @@ pub struct Config {
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
-pub struct Keypair {
+pub struct VaultKeypair {
     pub priv_key: KeyPackage,
     pub pub_key: PublicKeyPackage,
     pub tweak: Option<[u8; 32]>,
@@ -207,7 +206,6 @@ impl Config {
             home,
             p2p_keypair ,
             port: port as u32,
-            seed_mode: false,
             bootstrap_nodes: vec!["/ip4/192.248.180.245/tcp/5158/p2p/12D3KooWMpMtmYQKSn1sZaSRn4CAcsraWZVrZ2zdNjEgsEPSd3Pv".to_string()],
             log_level: "debug".to_string(),
             mnemonic: mnemonic.to_string(),

src/app/signer.rs

@@ -24,7 +24,7 @@ use crate::app::config::Config;
 use crate::helper::bitcoin::get_group_address_by_tweak;
 use crate::helper::cipher::random_bytes;
 use crate::helper::encoding::from_base64;
-use crate::helper::gossip::{subscribe_gossip_topics, SubscribeTopic};
+use crate::helper::gossip::{subscribe_gossip_topics, HeartBeatMessage, SubscribeTopic};
 use crate::helper::mem_store;
 use crate::protocols::sign::{received_sign_message, SignMesage, SignTask};
 use crate::tickers::tss::time_free_tasks_executor;
@@ -39,7 +39,7 @@ use std::io;
 use std::time::Duration;
 use tokio::select;
 use usize as Index;
-use tracing::{debug, error, info, warn};
+use tracing::{error, info, warn};
 
 use ed25519_compact::SecretKey;
 
@@ -168,7 +168,7 @@ impl Signer {
             let address_with_tweak = get_group_address_by_tweak(&pubkey.verifying_key(), tweak.clone(), self.config.bitcoin.network);
 
             addrs.push(address_with_tweak.to_string());
-            self.save_keypair_to_db(address_with_tweak.to_string(), &config::Keypair{
+            self.save_keypair_to_db(address_with_tweak.to_string(), &config::VaultKeypair{
                 priv_key: key.clone(),
                 pub_key: pubkey.clone(),
                 tweak: tweak,
@@ -337,15 +337,14 @@ impl Signer {
         self.db_sign.contains_key(task_id.as_bytes()).map_or(false, |v|v)
     }
 
-    pub fn list_keypairs(&self) -> Vec<String> {
+    pub fn list_keypairs(&self) -> Vec<(String, config::VaultKeypair)> {
         self.db_keypair.iter().map(|v| {
-            let (k, _value) = v.unwrap();
-            // keys.push(String::from_utf8(key.unwrap().0.to_vec()).unwrap());
-            String::from_utf8(k.to_vec()).unwrap()
+            let (k, value) = v.unwrap();
+            (String::from_utf8(k.to_vec()).unwrap(), serde_json::from_slice(&value).unwrap())
         }).collect::<Vec<_>>()
     }
 
-    pub fn get_keypair_from_db(&self, address: &str) -> Option<config::Keypair> {
+    pub fn get_keypair_from_db(&self, address: &str) -> Option<config::VaultKeypair> {
         match self.db_keypair.get(address) {
             Ok(Some(value)) => {
                 Some(serde_json::from_slice(&value).unwrap())
@@ -357,24 +356,35 @@ impl Signer {
         }
     }
 
-    pub fn save_keypair_to_db(&self, address: String, keypair: &config::Keypair) {
+    pub fn save_keypair_to_db(&self, address: String, keypair: &config::VaultKeypair) {
         let value = serde_json::to_vec(keypair).unwrap();
         let _ = self.db_keypair.insert(address, value);
     }
 
+    pub fn reset_db(&self) {
+        self.db_dkg.clear().expect("failed to clean db");
+        self.db_dkg_variables.clear().expect("failed to clean db");
+        self.db_sign.clear().expect("failed to clean db");
+        self.db_sign_variables.clear().expect("failed to clean db");
+    }
+
 }
 
 
-pub async fn run_signer_daemon(conf: Config) {
+pub async fn run_signer_daemon(conf: Config, seed: bool) {
 
     info!("Starting TSS Signer Daemon");
 
     // load config
     conf.load_validator_key();
     let signer = Signer::new(conf.clone());
 
-    for (i, key ) in signer.list_keypairs().iter().enumerate() {
-        info!("address {i}. {key}");
+    for (i, (addr, vkp) ) in signer.list_keypairs().iter().enumerate() {
+        info!("Vault {i}. {addr}");
+        // maintain a permission white list for heartbeat
+        vkp.pub_key.verifying_shares().keys().for_each(|identifier| {
+            mem_store::update_alive_table(HeartBeatMessage { identifier: identifier.clone(), last_seen: 0 });
+        });
     }
 
     let libp2p_keypair = Keypair::from_protobuf_encoding(from_base64(&conf.p2p_keypair).unwrap().as_slice()).unwrap();
@@ -432,7 +442,7 @@ pub async fn run_signer_daemon(conf: Config) {
     // swarm.listen_on(format!("/ip4/0.0.0.0/udp/{}/quic-v1", 5157).parse().expect("address parser error")).expect("failed to listen on all interfaces");
     swarm.listen_on(format!("/ip4/0.0.0.0/tcp/{}", conf.port).parse().expect("Address parse error")).expect("failed to listen on all interfaces");
 
-    if conf.seed_mode {
+    if seed || conf.bootstrap_nodes.len() == 0 {
         swarm.behaviour_mut().kad.set_mode(Some(libp2p::kad::Mode::Server));
     }
 
@@ -455,11 +465,11 @@ pub async fn run_signer_daemon(conf: Config) {
                 },
                 SwarmEvent::ConnectionEstablished { peer_id, num_established, endpoint, ..} => {
                     swarm.behaviour_mut().gossip.add_explicit_peer(&peer_id);
-                    let connected = swarm.connected_peers().map(|p| p.clone()).collect::<Vec<_>>();
-                    if connected.len() > 0 {
-                        swarm.behaviour_mut().identify.push(connected);
-                    }
-                    info!("Connected to {peer_id}, Swarm Connection Established, {num_established} {:?} ", endpoint);                  
+                    // let connected = swarm.connected_peers().map(|p| p.clone()).collect::<Vec<_>>();
+                    // if connected.len() > 0 {
+                    //     swarm.behaviour_mut().identify.push(connected);
+                    // }
+                    info!("Connected to {peer_id}, {num_established} {:?} ", endpoint);                  
                 },
                 SwarmEvent::ConnectionClosed { peer_id, cause, .. } => {
                     info!("Connection {peer_id} closed.{:?}", cause);
@@ -514,31 +524,30 @@ async fn event_handler(event: TSSBehaviourEvent, swarm: &mut Swarm<TSSBehaviour>
                 }
             } else if message.topic == SubscribeTopic::ALIVE.topic().hash() {
                 if let Ok(alive) = serde_json::from_slice(&message.data) {
-                    debug!("Received {:?}", alive);
                     mem_store::update_alive_table( alive );
                 }
             }
         }
-        TSSBehaviourEvent::Identify(identify::Event::Received { peer_id, info, .. }) => {
-            swarm.behaviour_mut().gossip.add_explicit_peer(&peer_id);
-            // info!(" @@(Received) Discovered new peer: {peer_id} with info: {connection_id} {:?}", info);
-            info.listen_addrs.iter().for_each(|addr| {
-                if !addr.to_string().starts_with("/ip4/127.0.0.1") {
-                    debug!("Discovered new address: {addr}/p2p/{peer_id} ");
-                    swarm.behaviour_mut().kad.add_address(&peer_id, addr.clone());
-                }
-            });
-        }
-        TSSBehaviourEvent::Kad(libp2p::kad::Event::RoutablePeer { peer, address }) => {
-            info!("@@@ Kad @@@ discovered a new routable peer {peer} - {:?}", address);
-            swarm.behaviour_mut().kad.add_address(&peer, address);
-        } 
-        TSSBehaviourEvent::Kad(libp2p::kad::Event::RoutingUpdated { peer, is_new_peer, addresses, .. }) => {
-            debug!("KAD Routing updated for {peer} {is_new_peer}: {:?}", addresses);
-            if is_new_peer {
-                swarm.behaviour_mut().gossip.add_explicit_peer(&peer);
-            }
-        }
+        // TSSBehaviourEvent::Identify(identify::Event::Received { peer_id, info, .. }) => {
+        //     swarm.behaviour_mut().gossip.add_explicit_peer(&peer_id);
+        //     // // info!(" @@(Received) Discovered new peer: {peer_id} with info: {connection_id} {:?}", info);
+        //     // info.listen_addrs.iter().for_each(|addr| {
+        //     //     if !addr.to_string().starts_with("/ip4/127.0.0.1") {
+        //     //         debug!("Discovered new address: {addr}/p2p/{peer_id} ");
+        //     //         swarm.behaviour_mut().kad.add_address(&peer_id, addr.clone());
+        //     //     }
+        //     // });
+        // }
+        // TSSBehaviourEvent::Kad(libp2p::kad::Event::RoutablePeer { peer, address }) => {
+        //     info!("@@@ Kad @@@ discovered a new routable peer {peer} - {:?}", address);
+        //     swarm.behaviour_mut().kad.add_address(&peer, address);
+        // } 
+        // TSSBehaviourEvent::Kad(libp2p::kad::Event::RoutingUpdated { peer, is_new_peer, addresses, .. }) => {
+        //     debug!("KAD Routing updated for {peer} {is_new_peer}: {:?}", addresses);
+        //     if is_new_peer {
+        //         swarm.behaviour_mut().gossip.add_explicit_peer(&peer);
+        //     }
+        // }
         TSSBehaviourEvent::Mdns(mdns::Event::Discovered(list)) => {
             for (peer_id, multiaddr) in list {
                 info!("mDNS discovered a new peer: {peer_id}");

src/commands/address.rs

@@ -1,4 +1,4 @@
-use crate::app::config::Config;
+use crate::app::{config::Config, signer::Signer};
 
 use super::Cli;
 
@@ -10,11 +10,10 @@ pub fn execute(cli: &Cli) {
     println!("\n NOTE: Please fund relayer address on sidechain before using it.");
     println!("-------------------------------------------------------------");
 
-    // println!("\nVault addresses: ({})", keypairs.len());
-    // println!("-------------------------------------------------------------");
-    // let keypairs = config::list_keypairs();
-    // for (index, k) in keypairs.iter().enumerate() {
-    //     println!("{}: {}", index, k);
-    // }
-    // println!("\n");
+    let conf = Config::from_file(&cli.home).unwrap();
+    let signer = Signer::new(conf);
+    println!("\nVault Address:");
+    signer.list_keypairs().iter().enumerate().for_each(| (i, (addr, _kp))| {
+        println!("{i}. {addr}");
+    });
 }

src/commands/mod.rs

@@ -27,6 +27,8 @@ pub enum Commands {
         relayer: bool,
         #[clap(long, default_value = "false")]
         signer: bool,
+        #[clap(long, default_value = "false")]
+        seed: bool,
     },
     #[command(about = "Submit a bitcoin header to the sidechain")]
     SubmitHeader {

src/commands/reset.rs

@@ -1,10 +1,11 @@
-// use crate::app::config;
+use crate::app::{config::Config, signer::Signer};
 
 use super::Cli;
 
-pub fn execute(_cli: &Cli) {
-    // config::update_app_home(&cli.home);
-    // dkg::delete_tasks();
-    // sign::delete_tasks();
+pub fn execute(cli: &Cli) {
+
+    let conf = Config::from_file(&cli.home).unwrap();
+    let signer = Signer::new(conf);
+    signer.reset_db();
     println!("Reset all tasks");
 }

src/commands/start.rs

@@ -4,7 +4,7 @@ use tracing_subscriber::{EnvFilter, FmtSubscriber};
 
 use crate::app::{config::Config, relayer, signer};
 
-pub async fn execute(home: &str, relayer: bool, signer: bool) {
+pub async fn execute(home: &str, relayer: bool, signer: bool, seed: bool) {
 
     let conf = Config::from_file(home).unwrap();
 
@@ -21,10 +21,10 @@ pub async fn execute(home: &str, relayer: bool, signer: bool) {
     if relayer && !signer {
         relayer::run_relayer_daemon(conf).await;
     } else if signer && !relayer {
-        signer::run_signer_daemon(conf).await;
+        signer::run_signer_daemon(conf, seed).await;
     } else {
         let conf2 = conf.clone();
-        let sign_handler = tokio::spawn(async move { signer::run_signer_daemon(conf).await });
+        let sign_handler = tokio::spawn(async move { signer::run_signer_daemon(conf, seed).await });
         let relay_handler = tokio::spawn(async move { relayer::run_relayer_daemon(conf2).await });
         // Start both signer and relayer as default
         match join!(sign_handler, relay_handler) {

src/helper/mem_store.rs

@@ -6,7 +6,6 @@
 use std::collections::BTreeMap;
 
 use frost_secp256k1_tr::{keys::dkg, Identifier};
-use tracing::debug;
 use std::sync::Mutex;
 use lazy_static::lazy_static;
 
@@ -27,8 +26,9 @@ lazy_static! {
 }
 
 pub fn update_alive_table(alive: HeartBeatMessage) {
-    let mut x= AliveTable.lock().unwrap();
-    x.insert(alive.identifier, alive.last_seen);
+    let mut table= AliveTable.lock().unwrap();
+    table.insert(alive.identifier, alive.last_seen);
+    table.retain(|_, v| {*v + 1800u64 > now()});
 }
 
 pub fn get_alive_participants(keys: &Vec<&Identifier>) -> usize {
@@ -39,11 +39,11 @@ pub fn get_alive_participants(keys: &Vec<&Identifier>) -> usize {
         // debug!("is alive {:?} {}", key, now() - last_seen);
         now() - last_seen < TASK_ROUND_WINDOW.as_secs() * 2
     }).count() + 1;
-    debug!("alive table: {alive}, {:?}", table);
+    // debug!("alive table: {alive}, {:?}", table);
     alive
 }
 
-pub fn is_alive(identifier: &Identifier) -> bool {
+pub fn is_white_listed_peer(identifier: &Identifier) -> bool {
     let table= AliveTable.lock().unwrap();
     table.contains_key(identifier)
 }

src/main.rs

@@ -11,8 +11,8 @@ async fn main() {
         Commands::Init { port, network } => {
             init::execute(&cli, port.to_owned(), network.to_owned());
         }
-        Commands::Start {relayer, signer} => {
-            start::execute(&cli.home, *relayer, *signer).await;
+        Commands::Start {relayer, signer, seed} => {
+            start::execute(&cli.home, *relayer, *signer, *seed).await;
         }
         Commands::Address => {
             address::execute(&cli);