darkfid2: skip-sync flag added, block on sync to wait for peers connection

bin/darkfid2/darkfid_config.toml

@@ -12,6 +12,9 @@ rpc_listen = "tcp://127.0.0.1:18340"
 # Participate in the consensus protocol
 consensus = false
 
+# Skip syncing process and start node right away
+skip_sync = false
+
 # Enable testing mode for local testing
 testing_mode = false
 

bin/darkfid2/src/main.rs

@@ -71,6 +71,10 @@ struct Args {
     /// Participate in the consensus protocol
     consensus: bool,
 
+    #[structopt(long)]
+    /// Skip syncing process and start node right away
+    skip_sync: bool,
+
     /// Syncing network settings
     #[structopt(flatten)]
     sync_net: SettingsOpt,
@@ -189,12 +193,11 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
     }
 
     // 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?;
+    if !args.skip_sync {
+        sync_task(&darkfid).await?;
+    } else {
+        darkfid.validator.write().await.synced = true;
+    }
 
     // Signal handling for graceful termination.
     let (signals_handler, signals_task) = SignalHandler::new()?;

bin/darkfid2/src/task/sync.rs

@@ -17,6 +17,7 @@
  */
 
 use darkfi::{
+    util::async_util::sleep,
     validator::proto::{SyncRequest, SyncResponse},
     Result,
 };
@@ -27,43 +28,55 @@ 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...");
+    // Block until at least node is connected to at least one peer
+    loop {
+        if !node.sync_p2p.channels().lock().await.is_empty() {
+            break
+        }
+        warn!(target: "darkfid::task::sync_task", "Node is not connected to other nodes, waiting to retry...");
+        sleep(10).await;
+    }
+
     // 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?;
+    let channel = node.sync_p2p.random_channel().await.unwrap();
 
-            // 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?;
+    // Communication setup
+    let msg_subsystem = channel.message_subsystem();
+    msg_subsystem.add_dispatch::<SyncResponse>().await;
+    let block_response_sub = channel.subscribe_msg::<SyncResponse>().await?;
 
-                // Node stores response data
-                let response = block_response_sub.receive().await?;
+    // TODO: make this parallel and use a head selection method,
+    // for example use a manual known head and only connect to nodes
+    // that follow that. Also use a random peer on every block range
+    // we sync.
 
-                // 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?;
+    // 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?;
 
-                let last_received = node.validator.read().await.blockchain.last()?;
-                info!(target: "darkfid::task::sync_task", "Last received block: {:?} - {:?}", last_received.0, last_received.1);
+        // TODO: add a timeout here to retry
+        // Node waits for response
+        let response = block_response_sub.receive().await?;
 
-                if last == last_received {
-                    break
-                }
+        // Verify and store retrieved blocks
+        debug!(target: "darkfid::task::sync_task", "Processing received blocks");
+        node.validator.read().await.add_blocks(&response.blocks).await?;
 
-                last = last_received;
-            }
+        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
         }
-        None => warn!(target: "darkfid::task::sync_task", "Node is not connected to other nodes"),
-    };
+
+        last = last_received;
+    }
 
     node.validator.write().await.synced = true;
     info!(target: "darkfid::task::sync_task", "Blockchain synced!");

bin/darkfid2/src/tests/harness.rs

@@ -91,13 +91,13 @@ 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, &validator_config, &settings, ex).await?;
+        let alice = generate_node(&vks, &validator_config, &settings, ex, true).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, &validator_config, &settings, ex).await?;
+        let bob = generate_node(&vks, &validator_config, &settings, ex, false).await?;
 
         Ok(Self { config, vks, validator_config, alice, bob })
     }
@@ -189,6 +189,7 @@ pub async fn generate_node(
     config: &ValidatorConfig,
     settings: &Settings,
     ex: &Arc<smol::Executor<'_>>,
+    skip_sync: bool,
 ) -> Result<Darkfid> {
     let sled_db = sled::Config::new().temporary(true).open()?;
     vks::inject(&sled_db, &vks)?;
@@ -203,9 +204,11 @@ pub async fn generate_node(
         }
     })
     .detach();
-    // TODO: enable this when ready
-    //sync_p2p.wait_for_outbound(ex).await?;
-    sync_task(&node).await?;
+    if !skip_sync {
+        sync_task(&node).await?;
+    } else {
+        node.validator.write().await.synced = true;
+    }
 
     Ok(node)
 }

bin/darkfid2/src/tests/mod.rs

@@ -55,15 +55,14 @@ async fn sync_blocks_real(ex: Arc<Executor<'_>>) -> Result<()> {
     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?;
+    let charlie = generate_node(&th.vks, &th.validator_config, &settings, &ex, false).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());
+    assert_eq!(alice.blockchain.len(), charlie.blockchain.len());
+    assert_eq!(alice.blockchain.slots.len(), charlie.blockchain.slots.len());
 
     // Thanks for reading
     Ok(())

src/blockchain/block_store.rs

@@ -21,7 +21,7 @@ use darkfi_serial::{deserialize, serialize, SerialDecodable, SerialEncodable};
 
 use crate::{tx::Transaction, Error, Result};
 
-use super::{parse_record, validate_slot, Header, SledDbOverlayPtr};
+use super::{parse_record, parse_u64_key_record, validate_slot, Header, SledDbOverlayPtr};
 
 /// Block version number
 pub const BLOCK_VERSION: u8 = 1;
@@ -405,7 +405,7 @@ impl BlockOrderStore {
         let mut order = vec![];
 
         for record in self.0.iter() {
-            order.push(parse_record(record.unwrap())?);
+            order.push(parse_u64_key_record(record.unwrap())?);
         }
 
         Ok(order)
@@ -421,7 +421,7 @@ impl BlockOrderStore {
         let mut counter = 0;
         while counter <= n {
             if let Some(found) = self.0.get_gt(key.to_be_bytes())? {
-                let (number, hash) = parse_record(found)?;
+                let (number, hash) = parse_u64_key_record(found)?;
                 key = number;
                 ret.push(hash);
                 counter += 1;
@@ -440,7 +440,7 @@ impl BlockOrderStore {
             Some(s) => s,
             None => return Err(Error::BlockNumberNotFound(0)),
         };
-        let (number, hash) = parse_record(found)?;
+        let (number, hash) = parse_u64_key_record(found)?;
 
         Ok((number, hash))
     }
@@ -449,7 +449,7 @@ impl BlockOrderStore {
     /// implementation for `Vec<u8>`.
     pub fn get_last(&self) -> Result<(u64, blake3::Hash)> {
         let found = self.0.last()?.unwrap();
-        let (number, hash) = parse_record(found)?;
+        let (number, hash) = parse_u64_key_record(found)?;
 
         Ok((number, hash))
     }
@@ -519,7 +519,7 @@ impl BlockOrderStoreOverlay {
     /// implementation for `Vec<u8>`.
     pub fn get_last(&self) -> Result<(u64, blake3::Hash)> {
         let found = self.0.lock().unwrap().last(SLED_BLOCK_ORDER_TREE)?.unwrap();
-        let (number, hash) = parse_record(found)?;
+        let (number, hash) = parse_u64_key_record(found)?;
 
         Ok((number, hash))
     }

src/blockchain/mod.rs

@@ -542,6 +542,15 @@ impl BlockchainOverlay {
     }
 }
 
+/// Parse a sled record with a u64 keyin the form of a tuple (`key`, `value`).
+pub fn parse_u64_key_record<T: Decodable>(record: (sled::IVec, sled::IVec)) -> Result<(u64, T)> {
+    let key_bytes: [u8; 8] = record.0.as_ref().try_into().unwrap();
+    let key = u64::from_be_bytes(key_bytes);
+    let value = deserialize(&record.1)?;
+
+    Ok((key, value))
+}
+
 /// Parse a sled record in the form of a tuple (`key`, `value`).
 pub fn parse_record<T1: Decodable, T2: Decodable>(
     record: (sled::IVec, sled::IVec),

src/blockchain/slot_store.rs

@@ -22,7 +22,7 @@ use darkfi_serial::{deserialize, serialize};
 
 use crate::{validator::consensus::pid::slot_pid_output, Error, Result};
 
-use super::{parse_record, SledDbOverlayPtr};
+use super::{parse_u64_key_record, SledDbOverlayPtr};
 
 /// A slot is considered valid when the following rules apply:
 ///     1. Id increments previous slot id
@@ -152,7 +152,7 @@ impl SlotStore {
         let mut slots = vec![];
 
         for slot in self.0.iter() {
-            let (_, slot): ([u8; 8], Slot) = parse_record(slot.unwrap())?;
+            let (_, slot): (u64, Slot) = parse_u64_key_record(slot.unwrap())?;
             slots.push(slot);
         }
 
@@ -169,7 +169,7 @@ impl SlotStore {
         let mut counter = 0;
         while counter <= n {
             if let Some(found) = self.0.get_gt(key.to_be_bytes())? {
-                let (id, slot) = parse_record(found)?;
+                let (id, slot) = parse_u64_key_record(found)?;
                 key = id;
                 ret.push(slot);
                 counter += 1;

src/blockchain/tx_store.rs

@@ -22,7 +22,7 @@ use darkfi_serial::{deserialize, serialize};
 
 use crate::{tx::Transaction, Error, Result};
 
-use super::{parse_record, SledDbOverlayPtr};
+use super::{parse_record, parse_u64_key_record, SledDbOverlayPtr};
 
 const SLED_TX_TREE: &[u8] = b"_transactions";
 const SLED_PENDING_TX_TREE: &[u8] = b"_pending_transactions";
@@ -320,7 +320,7 @@ impl PendingTxOrderStore {
         let mut txs = vec![];
 
         for tx in self.0.iter() {
-            txs.push(parse_record(tx.unwrap())?);
+            txs.push(parse_u64_key_record(tx.unwrap())?);
         }
 
         Ok(txs)