darkfid2: node blocks sync implemented

bin/darkfid2/src/main.rs

@@ -25,13 +25,10 @@ use darkfi::{
     async_daemonize,
     blockchain::BlockInfo,
     cli_desc,
-    net::{settings::SettingsOpt, P2p, P2pPtr, SESSION_ALL},
+    net::{settings::SettingsOpt, P2p, P2pPtr},
     rpc::server::listen_and_serve,
     util::time::TimeKeeper,
-    validator::{
-        proto::{ProtocolBlock, ProtocolTx},
-        Validator, ValidatorConfig, ValidatorPtr,
-    },
+    validator::{Validator, ValidatorConfig, ValidatorPtr},
     Result,
 };
 use darkfi_contract_test_harness::vks;
@@ -47,9 +44,13 @@ mod rpc;
 mod rpc_blockchain;
 mod rpc_tx;
 
+/// Validator async tasks
+mod task;
+use task::sync::sync_task;
+
 /// Utility functions
 mod utils;
-use utils::genesis_txs_total;
+use utils::{genesis_txs_total, spawn_sync_p2p};
 
 const CONFIG_FILE: &str = "darkfid_config.toml";
 const CONFIG_FILE_CONTENTS: &str = include_str!("../darkfid_config.toml");
@@ -139,29 +140,7 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
     let validator = Validator::new(&sled_db, config).await?;
 
     // Initialize syncing P2P network
-    let sync_p2p = {
-        info!(target: "darkfid", "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 { ProtocolBlock::init(channel, validator, p2p).await.unwrap() }
-            })
-            .await;
-
-        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
-    };
+    let sync_p2p = spawn_sync_p2p(&args.sync_net.into(), &validator).await;
 
     // Initialize consensus P2P network
     let consensus_p2p = {
@@ -173,7 +152,7 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
     };
 
     // Initialize node
-    let darkfid = Darkfid::new(sync_p2p.clone(), consensus_p2p.clone(), validator).await;
+    let darkfid = Darkfid::new(sync_p2p.clone(), consensus_p2p.clone(), validator.clone()).await;
     let darkfid = Arc::new(darkfid);
     info!(target: "darkfid", "Node initialized successfully!");
 
@@ -209,8 +188,13 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
         info!("Not starting consensus P2P network");
     }
 
-    // Simulate that we have synced
-    darkfid.validator.write().await.synced = true;
+    // Sync blockchain
+    info!("Waiting for sync P2P outbound connections");
+    // TODO: we have to wait here because sync task can start
+    // before P2P, so it will seem as we are not connected
+    // to other nodes, therefore not sync.
+    //sync_p2p.wait_for_outbound(ex).await?;
+    sync_task(&darkfid).await?;
 
     // Signal handling for graceful termination.
     let (signals_handler, signals_task) = SignalHandler::new()?;
@@ -218,11 +202,11 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
     info!(target: "darkfid", "Caught termination signal, cleaning up and exiting...");
 
     info!(target: "darkfid", "Stopping syncing P2P network...");
-    darkfid.sync_p2p.stop().await;
+    sync_p2p.stop().await;
 
     if args.consensus {
         info!(target: "darkfid", "Stopping consensus P2P network...");
-        darkfid.consensus_p2p.clone().unwrap().stop().await;
+        consensus_p2p.unwrap().stop().await;
     }
 
     info!(target: "darkfid", "Flushing sled database...");

bin/darkfid2/src/task/mod.rs

@@ -0,0 +1,22 @@
+/* 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/>.
+ */
+
+// TODO: Handle ? with matches in these files. They should be robust.
+
+pub mod sync;
+pub use sync::sync_task;

bin/darkfid2/src/task/sync.rs

@@ -0,0 +1,71 @@
+/* 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 darkfi::{
+    validator::proto::{SyncRequest, SyncResponse},
+    Result,
+};
+use log::{debug, info, warn};
+
+use crate::Darkfid;
+
+/// async task used for block syncing
+pub async fn sync_task(node: &Darkfid) -> Result<()> {
+    info!(target: "darkfid::task::sync_task", "Starting blockchain sync...");
+    // Getting a random connected channel to ask from peers
+    match node.sync_p2p.random_channel().await {
+        Some(channel) => {
+            // Communication setup
+            let msg_subsystem = channel.message_subsystem();
+            msg_subsystem.add_dispatch::<SyncResponse>().await;
+            let block_response_sub = channel.subscribe_msg::<SyncResponse>().await?;
+
+            // Node sends the last known block hash of the canonical blockchain
+            // and loops until the response is the same block (used to utilize
+            // batch requests).
+            let mut last = node.validator.read().await.blockchain.last()?;
+            info!(target: "darkfid::task::sync_task", "Last known block: {:?} - {:?}", last.0, last.1);
+            loop {
+                // Node creates a `SyncRequest` and sends it
+                let request = SyncRequest { slot: last.0, block: last.1 };
+                channel.send(&request).await?;
+
+                // Node stores response data
+                let response = block_response_sub.receive().await?;
+
+                // Verify and store retrieved blocks
+                debug!(target: "darkfid::task::sync_task", "block_sync_task(): Processing received blocks");
+                node.validator.write().await.add_blocks(&response.blocks).await?;
+
+                let last_received = node.validator.read().await.blockchain.last()?;
+                info!(target: "darkfid::task::sync_task", "Last received block: {:?} - {:?}", last_received.0, last_received.1);
+
+                if last == last_received {
+                    break
+                }
+
+                last = last_received;
+            }
+        }
+        None => warn!(target: "darkfid::task::sync_task", "Node is not connected to other nodes"),
+    };
+
+    node.validator.write().await.synced = true;
+    info!(target: "darkfid::task::sync_task", "Blockchain synced!");
+    Ok(())
+}

bin/darkfid2/src/tests/harness.rs

@@ -19,12 +19,11 @@
 use async_std::sync::Arc;
 use darkfi::{
     blockchain::{BlockInfo, Header},
-    net::{P2p, P2pPtr, Settings, SESSION_ALL},
+    net::Settings,
     util::time::TimeKeeper,
     validator::{
         consensus::{next_block_reward, pid::slot_pid_output},
-        proto::{ProtocolBlock, ProtocolTx},
-        Validator, ValidatorConfig, ValidatorPtr,
+        Validator, ValidatorConfig,
     },
     Result,
 };
@@ -36,7 +35,11 @@ use darkfi_sdk::{
 use log::error;
 use url::Url;
 
-use crate::{utils::genesis_txs_total, Darkfid};
+use crate::{
+    task::sync::sync_task,
+    utils::{genesis_txs_total, spawn_sync_p2p},
+    Darkfid,
+};
 
 pub struct HarnessConfig {
     pub testing_node: bool,
@@ -46,12 +49,14 @@ pub struct HarnessConfig {
 
 pub struct Harness {
     pub config: HarnessConfig,
+    pub vks: Vec<(Vec<u8>, String, Vec<u8>)>,
+    pub validator_config: ValidatorConfig,
     pub alice: Darkfid,
     pub bob: Darkfid,
 }
 
 impl Harness {
-    pub async fn new(config: HarnessConfig, ex: Arc<smol::Executor<'_>>) -> Result<Self> {
+    pub async fn new(config: HarnessConfig, ex: &Arc<smol::Executor<'_>>) -> Result<Self> {
         // Use test harness to generate genesis transactions
         let mut th = TestHarness::new(&["money".to_string(), "consensus".to_string()]).await?;
         let (genesis_stake_tx, _) = th.genesis_stake(&Holder::Alice, config.alice_initial)?;
@@ -70,7 +75,7 @@ impl Harness {
         // NOTE: we are not using consensus constants here so we
         // don't get circular dependencies.
         let time_keeper = TimeKeeper::new(genesis_block.header.timestamp, 10, 90, 0);
-        let val_config = ValidatorConfig::new(
+        let validator_config = ValidatorConfig::new(
             time_keeper,
             genesis_block,
             genesis_txs_total,
@@ -86,18 +91,18 @@ impl Harness {
         // Alice
         let alice_url = Url::parse("tcp+tls://127.0.0.1:18340")?;
         settings.inbound_addrs = vec![alice_url.clone()];
-        let alice = generate_node(&vks, &val_config, &settings, &ex).await?;
+        let alice = generate_node(&vks, &validator_config, &settings, ex).await?;
 
         // Bob
         let bob_url = Url::parse("tcp+tls://127.0.0.1:18341")?;
         settings.inbound_addrs = vec![bob_url];
         settings.peers = vec![alice_url];
-        let bob = generate_node(&vks, &val_config, &settings, &ex).await?;
+        let bob = generate_node(&vks, &validator_config, &settings, ex).await?;
 
-        Ok(Self { config, alice, bob })
+        Ok(Self { config, vks, validator_config, alice, bob })
     }
 
-    pub async fn validate_chains(&self, total_blocks: usize) -> Result<()> {
+    pub async fn validate_chains(&self, total_blocks: usize, total_slots: usize) -> Result<()> {
         let genesis_txs_total = self.config.alice_initial + self.config.bob_initial;
         let alice = &self.alice.validator.read().await;
         let bob = &self.bob.validator.read().await;
@@ -109,6 +114,10 @@ impl Harness {
         assert_eq!(alice_blockchain_len, bob.blockchain.len());
         assert_eq!(alice_blockchain_len, total_blocks);
 
+        let alice_slots_len = alice.blockchain.slots.len();
+        assert_eq!(alice_slots_len, bob.blockchain.slots.len());
+        assert_eq!(alice_slots_len, total_slots);
+
         Ok(())
     }
 
@@ -175,7 +184,7 @@ impl Harness {
     }
 }
 
-async fn generate_node(
+pub async fn generate_node(
     vks: &Vec<(Vec<u8>, String, Vec<u8>)>,
     config: &ValidatorConfig,
     settings: &Settings,
@@ -194,30 +203,9 @@ async fn generate_node(
         }
     })
     .detach();
-    node.validator.write().await.synced = true;
+    // TODO: enable this when ready
+    //sync_p2p.wait_for_outbound(ex).await?;
+    sync_task(&node).await?;
 
     Ok(node)
 }
-
-async fn spawn_sync_p2p(settings: &Settings, validator: &ValidatorPtr) -> P2pPtr {
-    let p2p = P2p::new(settings.clone()).await;
-    let registry = p2p.protocol_registry();
-
-    let _validator = validator.clone();
-    registry
-        .register(SESSION_ALL, move |channel, p2p| {
-            let validator = _validator.clone();
-            async move { ProtocolBlock::init(channel, validator, p2p).await.unwrap() }
-        })
-        .await;
-
-    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
-}

bin/darkfid2/src/tests/mod.rs

@@ -17,19 +17,20 @@
  */
 
 use async_std::sync::Arc;
-use darkfi::Result;
+use darkfi::{net::Settings, Result};
 use darkfi_contract_test_harness::init_logger;
 use smol::Executor;
+use url::Url;
 
 mod harness;
-use harness::{Harness, HarnessConfig};
+use harness::{generate_node, Harness, HarnessConfig};
 
 async fn sync_blocks_real(ex: Arc<Executor<'_>>) -> Result<()> {
     init_logger();
 
     // Initialize harness in testing mode
     let config = HarnessConfig { testing_node: true, alice_initial: 1000, bob_initial: 500 };
-    let th = Harness::new(config, ex).await?;
+    let th = Harness::new(config, &ex).await?;
 
     // Retrieve genesis block
     let previous = th.alice.validator.read().await.blockchain.last_block()?;
@@ -44,7 +45,25 @@ async fn sync_blocks_real(ex: Arc<Executor<'_>>) -> Result<()> {
     th.add_blocks(&vec![block1, block2]).await?;
 
     // Validate chains
-    th.validate_chains(3).await?;
+    th.validate_chains(3, 7).await?;
+
+    // We are going to create a third node and try to sync from the previous two
+    let mut settings = Settings::default();
+    settings.localnet = true;
+    let charlie_url = Url::parse("tcp+tls://127.0.0.1:18342")?;
+    settings.inbound_addrs = vec![charlie_url];
+    let alice_url = th.alice.sync_p2p.settings().inbound_addrs[0].clone();
+    let bob_url = th.bob.sync_p2p.settings().inbound_addrs[0].clone();
+    settings.peers = vec![alice_url, bob_url];
+    let charlie = generate_node(&th.vks, &th.validator_config, &settings, &ex).await?;
+    // Verify node synced
+    let genesis_txs_total = th.config.alice_initial + th.config.bob_initial;
+    let alice = &th.alice.validator.read().await;
+    let charlie = &charlie.validator.read().await;
+    charlie.validate_blockchain(genesis_txs_total, vec![]).await?;
+    // TODO: this fails because sync starts before we connect to peers
+    //assert_eq!(alice.blockchain.len(), charlie.blockchain.len());
+    //assert_eq!(alice.blockchain.slots.len(), charlie.blockchain.slots.len());
 
     // Thanks for reading
     Ok(())

bin/darkfid2/src/utils.rs

@@ -16,7 +16,18 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use darkfi::{error::TxVerifyFailed, tx::Transaction, Result};
+use log::info;
+
+use darkfi::{
+    error::TxVerifyFailed,
+    net::{P2p, P2pPtr, Settings, SESSION_ALL},
+    tx::Transaction,
+    validator::{
+        proto::{ProtocolBlock, ProtocolSync, ProtocolTx},
+        ValidatorPtr,
+    },
+    Result,
+};
 use darkfi_consensus_contract::{
     model::ConsensusGenesisStakeParamsV1, ConsensusFunction::GenesisStakeV1,
 };
@@ -57,3 +68,36 @@ pub fn genesis_txs_total(txs: &[Transaction]) -> Result<u64> {
 
     Ok(total)
 }
+
+/// Auxiliary function to generate the sync P2P network and register all its protocols.
+pub async fn spawn_sync_p2p(settings: &Settings, validator: &ValidatorPtr) -> P2pPtr {
+    info!(target: "darkfid", "Registering sync network P2P protocols...");
+    let p2p = P2p::new(settings.clone()).await;
+    let registry = p2p.protocol_registry();
+
+    let _validator = validator.clone();
+    registry
+        .register(SESSION_ALL, move |channel, p2p| {
+            let validator = _validator.clone();
+            async move { ProtocolBlock::init(channel, validator, p2p).await.unwrap() }
+        })
+        .await;
+
+    let _validator = validator.clone();
+    registry
+        .register(SESSION_ALL, move |channel, _p2p| {
+            let validator = _validator.clone();
+            async move { ProtocolSync::init(channel, validator).await.unwrap() }
+        })
+        .await;
+
+    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
+}

src/validator/proto/mod.rs

@@ -20,6 +20,10 @@
 mod protocol_block;
 pub use protocol_block::ProtocolBlock;
 
+/// Validator blockchain sync protocol
+mod protocol_sync;
+pub use protocol_sync::{ProtocolSync, SyncRequest, SyncResponse};
+
 /// Transaction broadcast protocol
 mod protocol_tx;
 pub use protocol_tx::ProtocolTx;

src/validator/proto/protocol_block.rs

@@ -33,6 +33,8 @@ use crate::{
     Result,
 };
 
+impl_p2p_message!(BlockInfo, "block");
+
 pub struct ProtocolBlock {
     block_sub: MessageSubscription<BlockInfo>,
     jobsman: ProtocolJobsManagerPtr,
@@ -41,8 +43,6 @@ pub struct ProtocolBlock {
     channel_address: Url,
 }
 
-impl_p2p_message!(BlockInfo, "block");
-
 impl ProtocolBlock {
     pub async fn init(
         channel: ChannelPtr,

src/validator/proto/protocol_sync.rs

@@ -0,0 +1,148 @@
+/* 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, error};
+use smol::Executor;
+
+use darkfi_serial::{SerialDecodable, SerialEncodable};
+
+use crate::{
+    blockchain::BlockInfo,
+    impl_p2p_message,
+    net::{
+        ChannelPtr, Message, MessageSubscription, ProtocolBase, ProtocolBasePtr,
+        ProtocolJobsManager, ProtocolJobsManagerPtr,
+    },
+    validator::ValidatorPtr,
+    Result,
+};
+
+// Constant defining how many blocks we send during syncing.
+const BATCH: u64 = 10;
+
+/// Auxiliary structure used for blockchain syncing.
+#[derive(Debug, SerialEncodable, SerialDecodable)]
+pub struct SyncRequest {
+    /// Slot UID
+    pub slot: u64,
+    /// Block headerhash of that slot
+    pub block: blake3::Hash,
+}
+
+impl_p2p_message!(SyncRequest, "syncrequest");
+
+/// Auxiliary structure used for blockchain syncing.
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
+pub struct SyncResponse {
+    /// Response blocks
+    pub blocks: Vec<BlockInfo>,
+}
+
+impl_p2p_message!(SyncResponse, "syncresponse");
+
+pub struct ProtocolSync {
+    request_sub: MessageSubscription<SyncRequest>,
+    jobsman: ProtocolJobsManagerPtr,
+    validator: ValidatorPtr,
+    channel: ChannelPtr,
+}
+
+impl ProtocolSync {
+    pub async fn init(channel: ChannelPtr, validator: ValidatorPtr) -> Result<ProtocolBasePtr> {
+        debug!(
+            target: "validator::protocol_sync::init",
+            "Adding ProtocolSync to the protocol registry"
+        );
+        let msg_subsystem = channel.message_subsystem();
+        msg_subsystem.add_dispatch::<SyncRequest>().await;
+
+        let request_sub = channel.subscribe_msg::<SyncRequest>().await?;
+
+        Ok(Arc::new(Self {
+            request_sub,
+            jobsman: ProtocolJobsManager::new("SyncProtocol", channel.clone()),
+            validator,
+            channel,
+        }))
+    }
+
+    async fn handle_receive_request(self: Arc<Self>) -> Result<()> {
+        debug!(target: "validator::protocol_sync::handle_receive_request", "START");
+        loop {
+            let request = match self.request_sub.receive().await {
+                Ok(v) => v,
+                Err(e) => {
+                    debug!(
+                        target: "validator::protocol_sync::handle_receive_request",
+                        "recv fail: {}",
+                        e
+                    );
+                    continue
+                }
+            };
+
+            // Check if node has finished syncing its blockchain
+            if !self.validator.read().await.synced {
+                debug!(
+                    target: "validator::protocol_sync::handle_receive_request",
+                    "Node still syncing blockchain, skipping..."
+                );
+                continue
+            }
+
+            let key = request.slot;
+            let blocks = match self.validator.read().await.blockchain.get_blocks_after(key, BATCH) {
+                Ok(v) => v,
+                Err(e) => {
+                    error!(
+                        target: "validator::protocol_sync::handle_receive_request",
+                        "get_blocks_after fail: {}",
+                        e
+                    );
+                    continue
+                }
+            };
+
+            let response = SyncResponse { blocks };
+            if let Err(e) = self.channel.send(&response).await {
+                error!(
+                    target: "validator::protocol_sync::handle_receive_request",
+                    "channel send fail: {}",
+                    e
+                )
+            };
+        }
+    }
+}
+
+#[async_trait]
+impl ProtocolBase for ProtocolSync {
+    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!(target: "validator::protocol_sync::start", "START");
+        self.jobsman.clone().start(executor.clone());
+        self.jobsman.clone().spawn(self.clone().handle_receive_request(), executor.clone()).await;
+        debug!(target: "validator::protocol_sync::start", "END");
+        Ok(())
+    }
+
+    fn name(&self) -> &'static str {
+        "ProtocolSync"
+    }
+}

src/validator/proto/protocol_tx.rs

@@ -33,6 +33,8 @@ use crate::{
     Result,
 };
 
+impl_p2p_message!(Transaction, "tx");
+
 pub struct ProtocolTx {
     tx_sub: MessageSubscription<Transaction>,
     jobsman: ProtocolJobsManagerPtr,
@@ -41,8 +43,6 @@ pub struct ProtocolTx {
     channel_address: Url,
 }
 
-impl_p2p_message!(Transaction, "tx");
-
 impl ProtocolTx {
     pub async fn init(
         channel: ChannelPtr,