darkfid2-validator: ProtocolTx created

bin/darkfid2/src/main.rs

@@ -16,10 +16,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use async_std::{
-    stream::StreamExt,
-    sync::{Arc, Mutex},
-};
+use async_std::{stream::StreamExt, sync::Arc};
 use log::info;
 use structopt_toml::{serde::Deserialize, structopt::StructOpt, StructOptToml};
 use url::Url;
@@ -28,10 +25,10 @@ use darkfi::{
     async_daemonize,
     blockchain::BlockInfo,
     cli_desc,
-    net::{settings::SettingsOpt, P2p, P2pPtr},
+    net::{settings::SettingsOpt, P2p, P2pPtr, SESSION_ALL},
     rpc::server::listen_and_serve,
     util::time::TimeKeeper,
-    validator::{Validator, ValidatorConfig, ValidatorPtr},
+    validator::{proto::ProtocolTx, Validator, ValidatorConfig, ValidatorPtr},
     Result,
 };
 use darkfi_contract_test_harness::vks;
@@ -95,7 +92,6 @@ pub struct Darkfid {
     sync_p2p: P2pPtr,
     consensus_p2p: Option<P2pPtr>,
     validator: ValidatorPtr,
-    synced: Mutex<bool>,
 }
 
 impl Darkfid {
@@ -104,7 +100,7 @@ impl Darkfid {
         consensus_p2p: Option<P2pPtr>,
         validator: ValidatorPtr,
     ) -> Self {
-        Self { synced: Mutex::new(false), sync_p2p, consensus_p2p, validator }
+        Self { sync_p2p, consensus_p2p, validator }
     }
 }
 
@@ -116,18 +112,6 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
         info!(target: "darkfid", "Node is configured to run in testing mode!");
     }
 
-    // Initialize syncing P2P network
-    let sync_p2p = P2p::new(args.sync_net.into()).await;
-
-    // Initialize consensus P2P network
-    let consensus_p2p = {
-        if !args.consensus {
-            None
-        } else {
-            Some(P2p::new(args.consensus_net.into()).await)
-        }
-    };
-
     // NOTE: everything is dummy for now
     // Initialize or open sled database
     let sled_db = sled::Config::new().temporary(true).open()?;
@@ -148,6 +132,32 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
     // Initialize validator
     let validator = Validator::new(&sled_db, config).await?;
 
+    // Initialize syncing P2P network
+    let sync_p2p = {
+        info!("Registering sync network P2P protocols...");
+        let p2p = P2p::new(args.sync_net.into()).await;
+        let registry = p2p.protocol_registry();
+
+        let _validator = validator.clone();
+        registry
+            .register(SESSION_ALL, move |channel, p2p| {
+                let validator = _validator.clone();
+                async move { ProtocolTx::init(channel, validator, p2p).await.unwrap() }
+            })
+            .await;
+
+        p2p
+    };
+
+    // Initialize consensus P2P network
+    let consensus_p2p = {
+        if !args.consensus {
+            None
+        } else {
+            Some(P2p::new(args.consensus_net.into()).await)
+        }
+    };
+
     // Initialize node
     let darkfid = Darkfid::new(sync_p2p, consensus_p2p, validator).await;
     let darkfid = Arc::new(darkfid);
@@ -159,7 +169,7 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
     ex.spawn(listen_and_serve(args.rpc_listen, darkfid.clone(), _ex)).detach();
 
     // Simulate that we have synced
-    *darkfid.synced.lock().await = true;
+    darkfid.validator.write().await.synced = true;
 
     // Signal handling for graceful termination.
     let (signals_handler, signals_task) = SignalHandler::new()?;

bin/darkfid2/src/proto/mod.rs

@@ -0,0 +1,21 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero 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 Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+/// Transaction broadcast protocol
+mod protocol_tx;
+pub use protocol_tx::ProtocolTx;

bin/darkfid2/src/proto/protocol_tx.rs

@@ -0,0 +1,121 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero 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 Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use async_std::sync::Arc;
+use async_trait::async_trait;
+use log::debug;
+use smol::Executor;
+use url::Url;
+
+use darkfi::{
+    consensus::ValidatorStatePtr,
+    impl_p2p_message,
+    net::{
+        ChannelPtr, Message, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
+        ProtocolJobsManager, ProtocolJobsManagerPtr,
+    },
+    tx::Transaction,
+    Result,
+};
+
+pub struct ProtocolTx {
+    tx_sub: MessageSubscription<Transaction>,
+    jobsman: ProtocolJobsManagerPtr,
+    state: ValidatorStatePtr,
+    p2p: P2pPtr,
+    channel_address: Url,
+}
+
+impl_p2p_message!(Transaction, "tx");
+
+impl ProtocolTx {
+    pub async fn init(
+        channel: ChannelPtr,
+        state: ValidatorStatePtr,
+        p2p: P2pPtr,
+    ) -> Result<ProtocolBasePtr> {
+        debug!(
+            target: "darkfid::protocol_tx::init",
+            "Adding ProtocolTx to the protocol registry"
+        );
+        let msg_subsystem = channel.message_subsystem();
+        msg_subsystem.add_dispatch::<Transaction>().await;
+
+        let tx_sub = channel.subscribe_msg::<Transaction>().await?;
+
+        Ok(Arc::new(Self {
+            tx_sub,
+            jobsman: ProtocolJobsManager::new("TxProtocol", channel.clone()),
+            state,
+            p2p,
+            channel_address: channel.address().clone(),
+        }))
+    }
+
+    async fn handle_receive_tx(self: Arc<Self>) -> Result<()> {
+        debug!(
+            target: "darkfid::protocol_tx::handle_receive_tx",
+            "START"
+        );
+        let exclude_list = vec![self.channel_address.clone()];
+        loop {
+            let tx = match self.tx_sub.receive().await {
+                Ok(v) => v,
+                Err(e) => {
+                    debug!(
+                        target: "darkfid::protocol_tx::handle_receive_tx",
+                        "recv fail: {}",
+                        e
+                    );
+                    continue
+                }
+            };
+
+            // Check if node has finished syncing its blockchain
+            if !self.state.read().await.synced {
+                debug!(
+                    target: "darkfid::protocol_tx::handle_receive_tx",
+                    "Node still syncing blockchain, skipping..."
+                );
+                continue
+            }
+
+            let tx_copy = (*tx).clone();
+
+            // Nodes use unconfirmed_txs vector as seen_txs pool.
+            if self.state.write().await.append_tx(tx_copy.clone()).await {
+                self.p2p.broadcast_with_exclude(&tx_copy, &exclude_list).await;
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl ProtocolBase for ProtocolTx {
+    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!(target: "darkfid::protocol_tx::start", "START");
+        self.jobsman.clone().start(executor.clone());
+        self.jobsman.clone().spawn(self.clone().handle_receive_tx(), executor.clone()).await;
+        debug!(target: "darkfid::protocol_tx::start", "END");
+        Ok(())
+    }
+
+    fn name(&self) -> &'static str {
+        "ProtocolTx"
+    }
+}

bin/darkfid2/src/rpc_blockchain.rs

@@ -53,13 +53,9 @@ impl Darkfid {
         }
 
         let slot = params[0].as_u64().unwrap();
-        let validator = self.validator.read().await;
 
-        let blocks = match validator.blockchain.get_blocks_by_slot(&[slot]) {
-            Ok(v) => {
-                drop(validator);
-                v
-            }
+        let blocks = match self.validator.read().await.blockchain.get_blocks_by_slot(&[slot]) {
+            Ok(v) => v,
             Err(e) => {
                 error!(target: "darkfid::rpc::blockchain_get_slot", "Failed fetching block by slot: {}", e);
                 return JsonError::new(InternalError, None, id).into()
@@ -103,13 +99,8 @@ impl Darkfid {
             return JsonError::new(ParseError, None, id).into()
         };
 
-        let validator = self.validator.read().await;
-
-        let txs = match validator.blockchain.transactions.get(&[tx_hash], true) {
-            Ok(txs) => {
-                drop(validator);
-                txs
-            }
+        let txs = match self.validator.read().await.blockchain.transactions.get(&[tx_hash], true) {
+            Ok(txs) => txs,
             Err(e) => {
                 error!(target: "darkfid::rpc::blockchain_get_tx", "Failed fetching tx by hash: {}", e);
                 return JsonError::new(InternalError, None, id).into()

bin/darkfid2/src/rpc_tx.rs

@@ -44,7 +44,7 @@ impl Darkfid {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        if !(*self.synced.lock().await) {
+        if !self.validator.read().await.synced {
             error!(target: "darkfid::rpc::tx_simulate", "Blockchain is not synced");
             return server_error(RpcError::NotSynced, id, None)
         }
@@ -94,7 +94,7 @@ impl Darkfid {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        if !(*self.synced.lock().await) {
+        if !self.validator.read().await.synced {
             error!(target: "darkfid::rpc::tx_broadcast", "Blockchain is not synced");
             return server_error(RpcError::NotSynced, id, None)
         }
@@ -117,8 +117,9 @@ impl Darkfid {
         };
 
         if self.consensus_p2p.is_some() {
-            // Consider we're participating in consensus here?
-            // The append_tx function performs a state transition check.
+            // Consensus participants can directly perform
+            // the state transition check and append to their
+            // pending transactions store.
             if self.validator.write().await.append_tx(tx.clone()).await.is_err() {
                 error!(target: "darkfid::rpc::tx_broadcast", "Failed to append transaction to mempool");
                 return server_error(RpcError::TxSimulationFail, id, None)
@@ -158,17 +159,13 @@ impl Darkfid {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        if !(*self.synced.lock().await) {
+        if !self.validator.read().await.synced {
             error!(target: "darkfid::rpc::tx_pending", "Blockchain is not synced");
             return server_error(RpcError::NotSynced, id, None)
         }
 
-        let validator = self.validator.read().await;
-        let pending_txs = match validator.blockchain.get_pending_txs() {
-            Ok(v) => {
-                drop(validator);
-                v
-            }
+        let pending_txs = match self.validator.read().await.blockchain.get_pending_txs() {
+            Ok(v) => v,
             Err(e) => {
                 error!(target: "darkfid::rpc::tx_pending", "Failed fetching pending txs: {}", e);
                 return JsonError::new(InternalError, None, id).into()
@@ -189,13 +186,12 @@ impl Darkfid {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        if !(*self.synced.lock().await) {
+        if !self.validator.read().await.synced {
             error!(target: "darkfid::rpc::tx_clean_pending", "Blockchain is not synced");
             return server_error(RpcError::NotSynced, id, None)
         }
 
-        let validator = self.validator.read().await;
-        let pending_txs = match validator.blockchain.get_pending_txs() {
+        let pending_txs = match self.validator.read().await.blockchain.get_pending_txs() {
             Ok(v) => v,
             Err(e) => {
                 error!(target: "darkfid::rpc::tx_clean_pending", "Failed fetching pending txs: {}", e);
@@ -203,12 +199,9 @@ impl Darkfid {
             }
         };
 
-        match validator.blockchain.remove_pending_txs(&pending_txs) {
-            Ok(()) => drop(validator),
-            Err(e) => {
-                error!(target: "darkfid::rpc::tx_clean_pending", "Failed fetching pending txs: {}", e);
-                return JsonError::new(InternalError, None, id).into()
-            }
+        if let Err(e) = self.validator.read().await.blockchain.remove_pending_txs(&pending_txs) {
+            error!(target: "darkfid::rpc::tx_clean_pending", "Failed fetching pending txs: {}", e);
+            return JsonError::new(InternalError, None, id).into()
         };
 
         let pending_txs: Vec<String> = pending_txs.iter().map(|x| x.hash().to_string()).collect();

src/validator/mod.rs

@@ -37,6 +37,9 @@ use consensus::{next_block_reward, Consensus};
 pub mod verification;
 use verification::{verify_block, verify_genesis_block, verify_transactions};
 
+/// P2P net protocols
+pub mod proto;
+
 /// Helper utilities
 pub mod utils;
 use utils::deploy_native_contracts;
@@ -77,6 +80,8 @@ pub struct Validator {
     pub blockchain: Blockchain,
     /// Hot/Live data used by the consensus algorithm
     pub consensus: Consensus,
+    /// Flag signalling node has finished initial sync
+    pub synced: bool,
     /// Flag to enable testing mode
     pub testing_mode: bool,
 }
@@ -114,7 +119,8 @@ impl Validator {
         let consensus = Consensus::new(blockchain.clone(), config.time_keeper);
 
         // Create the actual state
-        let state = Arc::new(RwLock::new(Self { blockchain, consensus, testing_mode }));
+        let state =
+            Arc::new(RwLock::new(Self { blockchain, consensus, synced: false, testing_mode }));
         info!(target: "validator", "Finished initializing validator");
 
         Ok(state)

src/validator/proto/mod.rs

@@ -0,0 +1,21 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero 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 Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+/// Transaction broadcast protocol
+mod protocol_tx;
+pub use protocol_tx::ProtocolTx;

src/validator/proto/protocol_tx.rs

@@ -0,0 +1,125 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero 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 Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use async_std::sync::Arc;
+use async_trait::async_trait;
+use log::debug;
+use smol::Executor;
+use url::Url;
+
+use crate::{
+    net::{
+        ChannelPtr, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
+        ProtocolJobsManager, ProtocolJobsManagerPtr,
+    },
+    tx::Transaction,
+    validator::ValidatorPtr,
+    Result,
+};
+
+pub struct ProtocolTx {
+    tx_sub: MessageSubscription<Transaction>,
+    jobsman: ProtocolJobsManagerPtr,
+    validator: ValidatorPtr,
+    p2p: P2pPtr,
+    channel_address: Url,
+}
+
+impl ProtocolTx {
+    pub async fn init(
+        channel: ChannelPtr,
+        validator: ValidatorPtr,
+        p2p: P2pPtr,
+    ) -> Result<ProtocolBasePtr> {
+        debug!(
+            target: "validator::protocol_tx::init",
+            "Adding ProtocolTx to the protocol registry"
+        );
+        let msg_subsystem = channel.message_subsystem();
+        msg_subsystem.add_dispatch::<Transaction>().await;
+
+        let tx_sub = channel.subscribe_msg::<Transaction>().await?;
+
+        Ok(Arc::new(Self {
+            tx_sub,
+            jobsman: ProtocolJobsManager::new("TxProtocol", channel.clone()),
+            validator,
+            p2p,
+            channel_address: channel.address().clone(),
+        }))
+    }
+
+    async fn handle_receive_tx(self: Arc<Self>) -> Result<()> {
+        debug!(
+            target: "validator::protocol_tx::handle_receive_tx",
+            "START"
+        );
+        let exclude_list = vec![self.channel_address.clone()];
+        loop {
+            let tx = match self.tx_sub.receive().await {
+                Ok(v) => v,
+                Err(e) => {
+                    debug!(
+                        target: "validator::protocol_tx::handle_receive_tx",
+                        "recv fail: {}",
+                        e
+                    );
+                    continue
+                }
+            };
+
+            // Check if node has finished syncing its blockchain
+            if !self.validator.read().await.synced {
+                debug!(
+                    target: "validator::protocol_tx::handle_receive_tx",
+                    "Node still syncing blockchain, skipping..."
+                );
+                continue
+            }
+
+            let tx_copy = (*tx).clone();
+
+            // Nodes use unconfirmed_txs vector as seen_txs pool.
+            match self.validator.write().await.append_tx(tx_copy.clone()).await {
+                Ok(()) => self.p2p.broadcast_with_exclude(&tx_copy, &exclude_list).await,
+                Err(e) => {
+                    debug!(
+                        target: "validator::protocol_tx::handle_receive_tx",
+                        "append_tc fail: {}",
+                        e
+                    );
+                }
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl ProtocolBase for ProtocolTx {
+    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!(target: "validator::protocol_tx::start", "START");
+        self.jobsman.clone().start(executor.clone());
+        self.jobsman.clone().spawn(self.clone().handle_receive_tx(), executor.clone()).await;
+        debug!(target: "validator::protocol_tx::start", "END");
+        Ok(())
+    }
+
+    fn name(&self) -> &'static str {
+        "ProtocolTx"
+    }
+}