feat(universe): add Tari Universe related changes (#31)

src/cli/args.rs

@@ -14,6 +14,10 @@ use crate::cli::util::validate_tribe;
 #[allow(clippy::struct_excessive_bools)]
 #[derive(Clone, Parser, Debug)]
 pub struct StartArgs {
+    /// (Optional) base dir.
+    #[arg(short, long, value_name = "base-dir")]
+    base_dir: Option<PathBuf>,
+
     /// (Optional) gRPC port to use.
     #[arg(short, long, value_name = "grpc-port")]
     pub grpc_port: Option<u16>,
@@ -115,17 +119,21 @@ pub enum Commands {
 pub struct Cli {
     #[command(subcommand)]
     pub command: Commands,
-
-    /// (Optional) base dir.
-    #[arg(short, long, value_name = "base-dir")]
-    base_dir: Option<PathBuf>,
 }
 
 impl Cli {
     pub fn base_dir(&self) -> PathBuf {
-        self.base_dir
-            .clone()
-            .unwrap_or_else(|| dirs::home_dir().unwrap().join(".tari/p2pool"))
+        match &self.command {
+            Commands::Start { args } => args
+                .base_dir
+                .clone()
+                .unwrap_or_else(|| dirs::home_dir().unwrap().join(".tari/p2pool")),
+            Commands::GenerateIdentity => dirs::home_dir().unwrap().join(".tari/p2pool"),
+            Commands::ListTribes { args } => args
+                .base_dir
+                .clone()
+                .unwrap_or_else(|| dirs::home_dir().unwrap().join(".tari/p2pool")),
+        }
     }
 
     /// Handles CLI command.

src/server/grpc/p2pool.rs

@@ -118,7 +118,7 @@ where
         // request new block template with shares as coinbases
         let shares = self.share_chain.generate_shares(reward).await;
 
-        let response = self
+        let mut response = self
             .client
             .lock()
             .await
@@ -136,6 +136,10 @@ where
             .miner_data
             .ok_or_else(|| Status::internal("missing miner data"))?;
         let target_difficulty = miner_data.target_difficulty / SHARE_COUNT;
+        if let Some(mut miner_data) = response.miner_data {
+            miner_data.target_difficulty = target_difficulty;
+            response.miner_data = Some(miner_data);
+        }
 
         Ok(Response::new(GetNewBlockResponse {
             block: Some(response),

src/server/http/stats/handlers.rs

@@ -14,7 +14,7 @@ use tari_core::consensus::ConsensusManager;
 use tari_core::transactions::tari_amount::MicroMinotari;
 use tari_utilities::epoch_time::EpochTime;
 
-use crate::server::http::stats::models::{BlockStats, EstimatedEarnings, Stats};
+use crate::server::http::stats::models::{BlockStats, EstimatedEarnings, Stats, TribeDetails};
 use crate::server::http::stats::server::AppState;
 use crate::server::http::stats::{
     MINER_STAT_ACCEPTED_BLOCKS_COUNT, MINER_STAT_REJECTED_BLOCKS_COUNT, P2POOL_STAT_ACCEPTED_BLOCKS_COUNT,
@@ -144,6 +144,7 @@ pub async fn handle_get_stats(State(state): State<AppState>) -> Result<Json<Stat
         estimated_earnings,
         miner_block_stats: miner_block_stats(state.stats_store.clone()).await,
         p2pool_block_stats: p2pool_block_stats(state.stats_store.clone()).await,
+        tribe: TribeDetails::new(state.tribe.to_string(), state.tribe.formatted()),
     }))
 }
 

src/server/http/stats/models.rs

@@ -73,10 +73,22 @@ impl BlockStats {
     }
 }
 
+#[derive(Serialize, Deserialize)]
+pub struct TribeDetails {
+    pub id: String,
+    pub name: String,
+}
+impl TribeDetails {
+    pub fn new(id: String, name: String) -> Self {
+        Self { id, name }
+    }
+}
+
 #[derive(Serialize, Deserialize)]
 pub struct Stats {
     pub connected: bool,
     pub connected_since: Option<EpochTime>,
+    pub tribe: TribeDetails,
     pub num_of_miners: usize,
     pub last_block_won: Option<StatsBlock>,
     pub share_chain_height: u64,

src/server/http/stats/server.rs

@@ -12,6 +12,7 @@ use tokio::io;
 
 use crate::server::http::stats::handlers;
 use crate::server::p2p::peer_store::PeerStore;
+use crate::server::p2p::Tribe;
 use crate::server::stats_store::StatsStore;
 use crate::sharechain::ShareChain;
 
@@ -46,6 +47,7 @@ where
     peer_store: Arc<PeerStore>,
     stats_store: Arc<StatsStore>,
     port: u16,
+    tribe: Tribe,
     shutdown_signal: ShutdownSignal,
 }
 
@@ -54,6 +56,7 @@ pub struct AppState {
     pub share_chain: Arc<dyn ShareChain>,
     pub peer_store: Arc<PeerStore>,
     pub stats_store: Arc<StatsStore>,
+    pub tribe: Tribe,
 }
 
 impl<S> StatsServer<S>
@@ -65,13 +68,15 @@ where
         peer_store: Arc<PeerStore>,
         stats_store: Arc<StatsStore>,
         port: u16,
+        tribe: Tribe,
         shutdown_signal: ShutdownSignal,
     ) -> Self {
         Self {
             share_chain,
             peer_store,
             stats_store,
             port,
+            tribe,
             shutdown_signal,
         }
     }
@@ -84,6 +89,7 @@ where
                 share_chain: self.share_chain.clone(),
                 peer_store: self.peer_store.clone(),
                 stats_store: self.stats_store.clone(),
+                tribe: self.tribe.clone(),
             })
     }
 

src/server/p2p/network.rs

@@ -71,6 +71,12 @@ pub struct Tribe {
     inner: String,
 }
 
+impl Tribe {
+    pub fn formatted(&self) -> String {
+        self.inner.to_case(Case::Lower).replace("_", " ").to_case(Case::Title)
+    }
+}
+
 impl ToValue for Tribe {
     fn to_value(&self) -> Value {
         Value::from(self.inner.as_str())
@@ -663,6 +669,9 @@ where
                             self.sync_in_progress.store(false, Ordering::SeqCst);
                         }
                         error!(target: LOG_TARGET, tribe = &self.config.tribe; "REQ-RES outbound failure: {peer:?} -> {error:?}");
+                        // Remove peer from peer store to try to sync from another peer,
+                        // if the peer goes online/accessible again, the peer store will have it again.
+                        self.tribe_peer_store.remove(&peer).await;
                     },
                     request_response::Event::InboundFailure { peer, error, .. } => {
                         if self.sync_in_progress.load(Ordering::SeqCst) {
@@ -689,13 +698,19 @@ where
                     },
                     _ => debug!(target: LOG_TARGET, tribe = &self.config.tribe; "[KADEMLIA] {event:?}"),
                 },
-                ServerNetworkBehaviourEvent::Identify(event) => {
-                    if let identify::Event::Received { peer_id, info } = event {
+                ServerNetworkBehaviourEvent::Identify(event) => match event {
+                    identify::Event::Received { peer_id, info } => {
                         for addr in info.listen_addrs {
                             self.swarm.behaviour_mut().kademlia.add_address(&peer_id, addr);
                         }
                         self.swarm.behaviour_mut().gossipsub.add_explicit_peer(&peer_id);
-                    }
+                    },
+                    identify::Event::Error { peer_id, error } => {
+                        warn!("Failed to identify peer {peer_id:?}: {error:?}");
+                        self.swarm.behaviour_mut().gossipsub.remove_explicit_peer(&peer_id);
+                        self.swarm.behaviour_mut().kademlia.remove_peer(&peer_id);
+                    },
+                    _ => {},
                 },
             },
             _ => {},

src/server/p2p/peer_store.rs

@@ -90,6 +90,13 @@ impl PeerStore {
         self.set_last_connected().await;
     }
 
+    /// Removes a peer from store.
+    pub async fn remove(&self, peer_id: &PeerId) {
+        self.peers.remove(peer_id).await;
+        self.set_tip_of_block_height().await;
+        self.set_last_connected().await;
+    }
+
     /// Collects all current tribes from all PeerInfo collected from broadcasts.
     pub async fn tribes(&self) -> Vec<Tribe> {
         self.peers

src/server/server.rs

@@ -99,6 +99,7 @@ where
                 tribe_peer_store.clone(),
                 stats_store.clone(),
                 config.stats_server.port,
+                config.p2p_service.tribe.clone(),
                 shutdown_signal.clone(),
             )))
         } else {

src/sharechain/in_memory.rs

@@ -3,6 +3,7 @@
 
 use std::ops::{Add, Div};
 use std::slice::Iter;
+use std::str::FromStr;
 use std::{collections::HashMap, sync::Arc};
 
 use async_trait::async_trait;
@@ -329,7 +330,7 @@ impl ShareChain for InMemoryShareChain {
             .with_height(last_block.height() + 1)
             .with_original_block_header(origin_block.header.clone())
             .with_miner_wallet_address(
-                TariAddress::from_hex(request.wallet_payment_address.as_str()).map_err(Error::TariAddress)?,
+                TariAddress::from_str(request.wallet_payment_address.as_str()).map_err(Error::TariAddress)?,
             )
             .build())
     }