event_graph_v2: implementing sync mechanism and fast head mode

bin/darkirc/src/irc/client.rs

@@ -197,6 +197,9 @@ impl Client {
                                 *self.last_sent.write().await = event_id;
 
                                 // If it fails for some reason, for now, we just note it and pass.
+                                if let Err(e) = self.server.darkirc.event_graph.header_dag_insert(vec![event.header.clone()]).await {
+                                    error!("[IRC CLIENT] Failed inserting new header to Header DAG: {}", e);
+                                }
                                 if let Err(e) = self.server.darkirc.event_graph.dag_insert(slice::from_ref(&event)).await {
                                     error!("[IRC CLIENT] Failed inserting new event to DAG: {e}");
                                 } else {

bin/darkirc/src/main.rs

@@ -128,7 +128,7 @@ struct Args {
 
     #[structopt(long)]
     // Whether to sync headers only or full sync
-    pub fast_mode: bool,
+    fast_mode: bool,
 
     #[structopt(long)]
     /// IRC Password (Encrypted with bcrypt-2b)
@@ -191,6 +191,9 @@ impl DarkIrc {
 
 async_daemonize!(realmain);
 async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
+    if args.fast_mode {
+        info!("fast mode enabled");
+    }
     // Abort the application on panic right away
     std::panic::set_hook(Box::new(panic_hook));
 
@@ -334,6 +337,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         }
     };
     let replay_mode = args.replay_mode;
+    let fast_mode = args.fast_mode;
 
     info!("Instantiating event DAG");
     let sled_db = match sled::open(datastore.clone()) {
@@ -357,6 +361,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         sled_db.clone(),
         replay_datastore.clone(),
         replay_mode,
+        fast_mode,
         "darkirc_dag",
         1,
         ex.clone(),

bin/genev/genevd/src/main.rs

@@ -121,6 +121,7 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
 
     let replay_datastore = expand_path(&settings.replay_datastore)?;
     let replay_mode = settings.replay_mode;
+    let fast_mode = settings.fast_mode;
 
     let sled_db = sled::open(datastore_path.clone())?;
     let p2p_settings: darkfi::net::Settings =
@@ -131,6 +132,7 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
         sled_db.clone(),
         replay_datastore,
         replay_mode,
+        fast_mode,
         "genevd_dag",
         1,
         executor.clone(),

bin/tau/taud/src/main.rs

@@ -516,6 +516,7 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
 
     let replay_datastore = expand_path(&settings.replay_datastore)?;
     let replay_mode = settings.replay_mode;
+    let fast_mode = settings.fast_mode;
 
     info!(target: "taud", "Instantiating event DAG");
     let sled_db = sled::open(datastore)?;
@@ -530,6 +531,7 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
         sled_db.clone(),
         replay_datastore,
         replay_mode,
+        fast_mode,
         "taud_dag",
         0,
         executor.clone(),

src/error.rs

@@ -479,6 +479,9 @@ pub enum Error {
     #[error("Event is invalid")]
     EventIsInvalid,
 
+    #[error("Header is invalid")]
+    HeaderIsInvalid,
+
     // ====================
     // Miscellaneous errors
     // ====================

src/event_graph/event.rs

@@ -20,6 +20,7 @@ use std::{collections::HashSet, time::UNIX_EPOCH};
 
 use darkfi_serial::{async_trait, deserialize_async, Encodable, SerialDecodable, SerialEncodable};
 use sled_overlay::{sled, SledTreeOverlay};
+use tracing::info;
 
 use crate::Result;
 
@@ -74,6 +75,7 @@ impl Header {
     ) -> Result<bool> {
         // Check if the event timestamp is after genesis timestamp
         if self.timestamp < genesis_timestamp - EVENT_TIME_DRIFT {
+            info!("timestampe");
             return Ok(false)
         }
 
@@ -82,6 +84,7 @@ impl Header {
         if days_rotation > 0 {
             let next_genesis_timestamp = next_rotation_timestamp(INITIAL_GENESIS, days_rotation);
             if self.timestamp > next_genesis_timestamp + EVENT_TIME_DRIFT {
+                info!("rotation");
                 return Ok(false)
             }
         }
@@ -95,14 +98,17 @@ impl Header {
 
         for parent_id in self.parents.iter() {
             if parent_id == &NULL_ID {
+                info!("null");
                 continue
             }
 
             if parent_id == &self_id {
+                info!("self");
                 return Ok(false)
             }
 
             if seen.contains(parent_id) {
+                info!("seen");
                 return Ok(false)
             }
 
@@ -112,11 +118,13 @@ impl Header {
                 dag.get(parent_id.as_bytes())?
             };
             if parent_bytes.is_none() {
+                info!("none");
                 return Ok(false)
             }
 
             let parent: Header = deserialize_async(&parent_bytes.unwrap()).await?;
             if self.layer <= parent.layer {
+                info!("layer");
                 return Ok(false)
             }
 
@@ -163,70 +171,14 @@ impl Event {
     /// to use that instead of actual referenced DAG.
     /// TODO: is this necessary? we validate headers and events should
     /// be downloaded into the correct structure.
-    pub async fn validate(
-        &self,
-        dag: &sled::Tree,
-        genesis_timestamp: u64,
-        days_rotation: u64,
-        overlay: Option<&SledTreeOverlay>,
-    ) -> Result<bool> {
+    pub async fn validate(&self, _dag: &sled::Tree) -> Result<bool> {
         // Let's not bother with empty events
         if self.content.is_empty() {
+            info!("empty");
             return Ok(false)
         }
 
-        // Check if the event timestamp is after genesis timestamp
-        if self.header.timestamp < genesis_timestamp - EVENT_TIME_DRIFT {
-            return Ok(false)
-        }
-
-        // If a rotation has been set, check if the event timestamp
-        // is after the next genesis timestamp
-        if days_rotation > 0 {
-            let next_genesis_timestamp = next_rotation_timestamp(INITIAL_GENESIS, days_rotation);
-            if self.header.timestamp > next_genesis_timestamp + EVENT_TIME_DRIFT {
-                return Ok(false)
-            }
-        }
-
-        // Validate the parents. We have to check that at least one parent
-        // is not NULL, that the parents exist, that no two parents are the
-        // same, and that the parent exists in previous layers, to prevent
-        // recursive references(circles).
-        let mut seen = HashSet::new();
-        let self_id = self.header.id();
-
-        for parent_id in self.header.parents.iter() {
-            if parent_id == &NULL_ID {
-                continue
-            }
-
-            if parent_id == &self_id {
-                return Ok(false)
-            }
-
-            if seen.contains(parent_id) {
-                return Ok(false)
-            }
-
-            let parent_bytes = if let Some(overlay) = overlay {
-                overlay.get(parent_id.as_bytes())?
-            } else {
-                dag.get(parent_id.as_bytes())?
-            };
-            if parent_bytes.is_none() {
-                return Ok(false)
-            }
-
-            let parent: Event = deserialize_async(&parent_bytes.unwrap()).await?;
-            if self.header.layer <= parent.header.layer {
-                return Ok(false)
-            }
-
-            seen.insert(parent_id);
-        }
-
-        Ok(!seen.is_empty())
+        Ok(true)
     }
 
     /// Fully validate an event for the correct layout against provided
@@ -234,11 +186,10 @@ impl Event {
     /// possibility for a time drift.
     pub async fn dag_validate(&self, event_graph: &EventGraph) -> Result<bool> {
         // Grab genesis timestamp
-        let genesis_timestamp = event_graph.current_genesis.read().await.header.timestamp;
+        // let genesis_timestamp = event_graph.current_genesis.read().await.header.timestamp;
 
         // Perform validation
-        self.validate(&event_graph.header_dag, genesis_timestamp, event_graph.days_rotation, None)
-            .await
+        self.validate(&event_graph.header_dag).await
     }
 
     /// Validate a new event for the correct layout and enforce relevant age,
@@ -301,7 +252,7 @@ mod tests {
         let ex = Arc::new(Executor::new());
         let p2p = P2p::new(Settings::default(), ex.clone()).await?;
         let sled_db = sled::Config::new().temporary(true).open().unwrap();
-        EventGraph::new(p2p, sled_db, "/tmp".into(), false, "dag", 1, ex).await
+        EventGraph::new(p2p, sled_db, "/tmp".into(), false, false, "dag", 1, ex).await
     }
 
     #[test]

src/event_graph/mod.rs

@@ -16,9 +16,9 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use async_std::stream::from_iter;
+// use async_std::stream::from_iter;
 use futures::{
-    future,
+    // future,
     stream::{FuturesOrdered, FuturesUnordered},
     StreamExt,
 };
@@ -127,6 +127,9 @@ pub struct EventGraph {
     pub deg_enabled: RwLock<bool>,
     /// The publisher for which we can give deg info over
     deg_publisher: PublisherPtr<DegEvent>,
+    /// Run in replay_mode where if set we log Sled DB instructions
+    /// into `datastore`, useful to reacreate a faulty DAG to debug.
+    fast_mode: bool,
 }
 
 impl EventGraph {
@@ -148,6 +151,7 @@ impl EventGraph {
         sled_db: sled::Db,
         datastore: PathBuf,
         replay_mode: bool,
+        fast_mode: bool,
         dag_tree_name: &str,
         days_rotation: u64,
         ex: Arc<Executor<'_>>,
@@ -167,6 +171,7 @@ impl EventGraph {
             main_dag: dag.clone(),
             datastore,
             replay_mode,
+            fast_mode,
             unreferenced_tips,
             broadcasted_ids,
             prune_task: OnceCell::new(),
@@ -216,26 +221,6 @@ impl EventGraph {
         self.days_rotation
     }
 
-    // /// Header sync
-    // async fn retrieve_headers(&self) -> Result<()> {
-    //     let peers = self.p2p.hosts().peers();
-    //     let mut communicated_peers = peers.len();
-    //     info!(target: "event_graph::retrieve_headers()", "Retrieving missing headers from peers...");
-    //     // Communication setup
-    //     let mut peer_subs = vec![];
-    //     for peer in peers {
-    //         match peer.subscribe_msg::<HeaderReq>().await {
-    //             Ok(response_sub) => peer_subs.push(Some(response_sub)),
-    //             Err(e) => {
-    //                 debug!(target: "darkfid::task::sync::retrieve_headers", "Failure during `HeaderSyncResponse` communication setup with peer {peer:?}: {e}");
-    //                 peer_subs.push(None)
-    //             }
-    //         }
-    //     }
-
-    //     Ok(())
-    // }
-
     /// Sync the DAG from connected peers
     pub async fn dag_sync(&self, fast_mode: bool) -> Result<()> {
         // We do an optimistic sync where we ask all our connected peers for
@@ -337,20 +322,22 @@ impl EventGraph {
         }
         drop(tips);
 
-        // Now begin fetching the events backwards.
-        let mut missing_parents = HashSet::new();
-        for tip in considered_tips.iter() {
-            assert!(tip != &NULL_ID);
+        if fast_mode {
+            // Now begin fetching the events backwards.
+            let mut missing_parents = HashSet::new();
+            for tip in considered_tips.iter() {
+                assert!(tip != &NULL_ID);
 
-            if !self.main_dag.contains_key(tip.as_bytes()).unwrap() {
-                missing_parents.insert(*tip);
+                if !self.main_dag.contains_key(tip.as_bytes()).unwrap() {
+                    missing_parents.insert(*tip);
+                }
             }
-        }
 
-        if missing_parents.is_empty() {
-            *self.synced.write().await = true;
-            info!(target: "event_graph::dag_sync", "[EVENTGRAPH] DAG synced successfully!");
-            return Ok(())
+            if missing_parents.is_empty() {
+                *self.synced.write().await = true;
+                info!(target: "event_graph::dag_sync", "[EVENTGRAPH] DAG synced successfully!");
+                return Ok(())
+            }
         }
 
         // Header sync first
@@ -363,15 +350,13 @@ impl EventGraph {
         }
 
         while let Some(peer_headers) = headers_requests.next().await {
-            info!("Received headers {:?}", peer_headers);
             self.header_dag_insert(peer_headers?).await?
         }
 
-        let peers = channels.clone().into_iter().collect::<Vec<_>>();
-
         // start download payload
         if !fast_mode {
             info!(target: "event_graph::dag_sync()", "[EVENTGRAPH] Fetching events");
+            let peers = channels.clone().into_iter().collect::<Vec<_>>();
             let mut header_sorted = vec![];
 
             for iter_elem in self.header_dag.iter() {
@@ -379,35 +364,29 @@ impl EventGraph {
                 let val: Header = deserialize_async(&val).await.unwrap();
                 header_sorted.push(val);
             }
-
             header_sorted.sort_by(|x, y| y.layer.cmp(&x.layer));
-            for i in header_sorted.iter() {
-                info!("header: {}", i.id());
-            }
-            // info!("layer number of last header: {}", header_sorted.last().unwrap().layer);
 
-            // // // 1. Fetch events one by one
+            // 1. Fetch events one by one
             // let mut events_requests = FuturesOrdered::new();
-            // let peer = peer_selection(peers.clone());
-            // for header in header_sorted.iter() {
-            //     events_requests.push_back(request_event(
-            //         peer.clone(),
-            //         vec![header.id()],
-            //         comms_timeout,
-            //     ))
-            // }
+            let peer = peer_selection(peers.clone());
+            // let peer = channels[0].clone();
+            for header in header_sorted.iter() {
+                let received_events =
+                    request_event(peer.clone(), vec![header.id()], comms_timeout).await?;
+                self.dag_insert(&received_events).await?;
+            }
 
-            // let mut rcvd_events = vec![];
+            // let mut received_events = vec![];
             // while let Some(peer_events) = events_requests.next().await {
             //     let events = peer_events?;
-            //     info!("Received events {:?}", events);
             //     for i in events.iter() {
+            //         info!("Received events id: {:?}", i.header.id());
             //         info!("layer: {}", i.header.layer);
             //     }
-            //     rcvd_events.extend(events);
+            //     received_events.extend(events);
             // }
 
-            // self.dag_insert(&rcvd_events).await?;
+            // self.dag_insert(&received_events).await?;
 
             // // 2. split sorted headers into chunks and assign them to each connected peer
             // let mut responses = vec![];
@@ -437,121 +416,6 @@ impl EventGraph {
             // let parents = send_requests(&peers, &missing).await?.concat();
             // info!("fetched parents: {}", parents.len());
         }
-
-        info!(target: "event_graph::dag_sync()", "[EVENTGRAPH] Fetching events");
-        let mut received_events: BTreeMap<u64, Vec<Event>> = BTreeMap::new();
-        let mut received_events_hashes = HashSet::new();
-
-        while !missing_parents.is_empty() {
-            let mut found_event = false;
-
-            for channel in channels.iter() {
-                let url = channel.display_address();
-
-                debug!(
-                    target: "event_graph::dag_sync",
-                    "Requesting {missing_parents:?} from {url}..."
-                );
-
-                let ev_rep_sub = match channel.subscribe_msg::<EventRep>().await {
-                    Ok(v) => v,
-                    Err(e) => {
-                        error!(
-                            target: "event_graph::dag_sync",
-                            "[EVENTGRAPH] Sync: Couldn't subscribe EventRep for peer {url}, skipping ({e})"
-                        );
-                        continue
-                    }
-                };
-
-                let request_missing_events = missing_parents.clone().into_iter().collect();
-                if let Err(e) = channel.send(&EventReq(request_missing_events)).await {
-                    error!(
-                        target: "event_graph::dag_sync",
-                        "[EVENTGRAPH] Sync: Failed communicating EventReq({missing_parents:?}) to {url}: {e}"
-                    );
-                    continue
-                }
-
-                // Node waits for response
-                let Ok(parent) = ev_rep_sub.receive_with_timeout(comms_timeout).await else {
-                    error!(
-                        target: "event_graph::dag_sync",
-                        "[EVENTGRAPH] Sync: Timeout waiting for parents {missing_parents:?} from {url}"
-                    );
-                    continue
-                };
-
-                let parents = parent.0.clone();
-
-                for parent in parents {
-                    let parent_id = parent.header.id();
-                    if !missing_parents.contains(&parent_id) {
-                        error!(
-                            target: "event_graph::dag_sync",
-                            "[EVENTGRAPH] Sync: Peer {url} replied with a wrong event: {}",
-                            parent.header.id()
-                        );
-                        continue
-                    }
-
-                    debug!(
-                        target: "event_graph::dag_sync",
-                        "Got correct parent event {parent_id}"
-                    );
-
-                    if let Some(layer_events) = received_events.get_mut(&parent.header.layer) {
-                        layer_events.push(parent.clone());
-                    } else {
-                        let layer_events = vec![parent.clone()];
-                        received_events.insert(parent.header.layer, layer_events);
-                    }
-                    received_events_hashes.insert(parent_id);
-
-                    missing_parents.remove(&parent_id);
-                    found_event = true;
-
-                    // See if we have the upper parents
-                    for upper_parent in parent.header.parents.iter() {
-                        if upper_parent == &NULL_ID {
-                            continue
-                        }
-
-                        if !missing_parents.contains(upper_parent) &&
-                            !received_events_hashes.contains(upper_parent) &&
-                            !self.main_dag.contains_key(upper_parent.as_bytes()).unwrap()
-                        {
-                            debug!(
-                                target: "event_graph::dag_sync",
-                                "Found upper missing parent event {upper_parent}"
-                            );
-                            missing_parents.insert(*upper_parent);
-                        }
-                    }
-                }
-
-                break
-            }
-
-            if !found_event {
-                error!(
-                    target: "event_graph::dag_sync",
-                    "[EVENTGRAPH] Sync: Failed to get all events",
-                );
-                return Err(Error::DagSyncFailed)
-            }
-        } // <-- while !missing_parents.is_empty
-
-        // At this point we should've got all the events.
-        // We should add them to the DAG.
-        let mut events = vec![];
-        for (_, tips) in received_events {
-            for tip in tips {
-                events.push(tip);
-            }
-        }
-        self.dag_insert(&events).await?;
-
         // <-- end download payload
 
         *self.synced.write().await = true;
@@ -670,25 +534,32 @@ impl EventGraph {
         let mut overlay = SledTreeOverlay::new(&self.main_dag);
 
         // Grab genesis timestamp
-        let genesis_timestamp = self.current_genesis.read().await.header.timestamp;
+        // let genesis_timestamp = self.current_genesis.read().await.header.timestamp;
 
         // Iterate over given events to validate them and
         // write them to the overlay
         for event in events {
             let event_id = event.header.id();
             if event.header.layer == 0 {
-                continue
+                return Ok(vec![])
             }
             debug!(
                 target: "event_graph::dag_insert",
-                "Inserting event {event_id} into the DAG"
+                "Inserting event {event_id} into the DAG layer: {}", event.header.layer
             );
 
-            if !event
-                .validate(&self.main_dag, genesis_timestamp, self.days_rotation, Some(&overlay))
-                .await?
-            {
-                error!(target: "event_graph::dag_insert", "Event {event_id} is invalid!");
+            // check if we already have the event
+            if self.main_dag.contains_key(event_id.as_bytes())? {
+                continue
+            }
+
+            // check if its header is in header's store
+            if !self.header_dag.contains_key(event_id.as_bytes())? {
+                continue
+            }
+
+            if !event.validate(&self.header_dag).await? {
+                error!(target: "event_graph::dag_insert()", "Event {} is invalid!", event_id);
                 return Err(Error::EventIsInvalid)
             }
 
@@ -705,7 +576,10 @@ impl EventGraph {
         }
 
         // Aggregate changes into a single batch
-        let batch = overlay.aggregate().unwrap();
+        let batch = match overlay.aggregate() {
+            Some(x) => x,
+            None => return Ok(vec![]),
+        };
 
         // Atomically apply the batch.
         // Panic if something is corrupted.
@@ -769,13 +643,17 @@ impl EventGraph {
         Ok(ids)
     }
 
-    async fn header_dag_insert(&self, headers: Vec<Header>) -> Result<()> {
+    pub async fn header_dag_insert(&self, headers: Vec<Header>) -> Result<()> {
         // Create an overlay over the DAG tree
         let mut overlay = SledTreeOverlay::new(&self.header_dag);
 
         // Grab genesis timestamp
         let genesis_timestamp = self.current_genesis.read().await.header.timestamp;
 
+        // Acquire exclusive locks to `unreferenced_tips and broadcasted_ids`
+        // let mut unreferenced_header = self.unreferenced_tips.write().await;
+        // let mut broadcasted_ids = self.broadcasted_ids.write().await;
+
         let mut hdrs = headers;
         hdrs.sort_by(|x, y| x.layer.cmp(&y.layer));
 
@@ -788,14 +666,14 @@ impl EventGraph {
             }
             debug!(
                 target: "event_graph::header_dag_insert()",
-                "Inserting event {} into the DAG", header_id,
+                "Inserting header {} into the DAG", header_id,
             );
             if !header
                 .validate(&self.header_dag, genesis_timestamp, self.days_rotation, Some(&overlay))
                 .await?
             {
                 error!(target: "event_graph::header_dag_insert()", "Header {} is invalid!", header_id);
-                return Err(Error::EventIsInvalid)
+                return Err(Error::HeaderIsInvalid)
             }
             let header_se = serialize_async(&header).await;
 
@@ -804,7 +682,10 @@ impl EventGraph {
         }
 
         // Aggregate changes into a single batch
-        let batch = overlay.aggregate().unwrap();
+        let batch = match overlay.aggregate() {
+            Some(x) => x,
+            None => return Ok(()),
+        };
 
         // Atomically apply the batch.
         // Panic if something is corrupted.
@@ -1045,7 +926,7 @@ impl EventGraph {
     }
 }
 
-async fn send_request(peer: &Channel, missing: &Header) -> Result<Vec<Event>> {
+async fn _send_request(peer: &Channel, missing: &Header) -> Result<Vec<Event>> {
     info!("in send_request first missing: {}", missing.id());
     let url = peer.address();
     debug!(target: "event_graph::dag_sync()","Requesting {:?} from {}...", missing, url);
@@ -1089,15 +970,6 @@ async fn request_header(peer: &Channel, comms_timeout: u64) -> Result<Vec<Header
         }
     };
 
-    // let local_tips = self
-    //     .unreferenced_tips
-    //     .read()
-    //     .await
-    //     .values()
-    //     .flat_map(|x| x.iter())
-    //     .cloned()
-    //     .collect();
-
     if let Err(e) = peer.send(&HeaderReq {}).await {
         error!(
             target: "event_graph::dag_sync()",

src/event_graph/proto.rs

@@ -331,6 +331,9 @@ impl ProtocolEventGraph {
                 "Event {event_id} is new"
             );
 
+            // TODO: when in fast mode (headers only mode) if someone
+            // sends us an event this block will try to request all
+            // the events parents
             let mut missing_parents = HashSet::new();
             for parent_id in event.header.parents.iter() {
                 // `event.validate_new()` should have already made sure that
@@ -453,10 +456,18 @@ impl ProtocolEventGraph {
                         events.push(tip);
                     }
                 }
-                if self.event_graph.dag_insert(&events).await.is_err() {
+                let headers = events.iter().map(|x| x.header.clone()).collect();
+                if self.event_graph.header_dag_insert(headers).await.is_err() {
                     self.clone().increase_malicious_count().await?;
                     continue
                 }
+                // FIXME
+                if !self.event_graph.fast_mode {
+                    if self.event_graph.dag_insert(&events).await.is_err() {
+                        self.clone().increase_malicious_count().await?;
+                        continue
+                    }
+                }
             } // <-- !missing_parents.is_empty()
 
             // If we're here, we have all the parents, and we can now
@@ -466,6 +477,11 @@ impl ProtocolEventGraph {
                 target: "event_graph::protocol::handle_event_put",
                 "Got all parents necessary for insertion",
             );
+            if self.event_graph.header_dag_insert(vec![event.header.clone()]).await.is_err() {
+                self.clone().increase_malicious_count().await?;
+                continue
+            }
+
             if self.event_graph.dag_insert(slice::from_ref(&event)).await.is_err() {
                 self.clone().increase_malicious_count().await?;
                 continue

src/event_graph/tests.rs

@@ -32,7 +32,7 @@ use crate::{
         Event, EventGraph,
     },
     net::{session::SESSION_DEFAULT, settings::NetworkProfile, P2p, Settings},
-    system::sleep,
+    system::{msleep, sleep},
     util::logger::{setup_test_logger, Level},
 };
 
@@ -99,7 +99,7 @@ async fn spawn_node(
     let p2p = P2p::new(settings, ex.clone()).await.unwrap();
     let sled_db = sled::Config::new().temporary(true).open().unwrap();
     let event_graph =
-        EventGraph::new(p2p.clone(), sled_db, "/tmp".into(), false, "dag", 1, ex.clone())
+        EventGraph::new(p2p.clone(), sled_db, "/tmp".into(), false, false, "dag", 1, ex.clone())
             .await
             .unwrap();
     *event_graph.synced.write().await = true;
@@ -219,7 +219,6 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     // =========================================
     // 1. Assert that everyone's DAG is the same
     // =========================================
-    info!("dag len is 1");
     assert_dags(&eg_instances, 1, &mut rng).await;
 
     // ==========================================
@@ -229,6 +228,7 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     let event = Event::new(vec![1, 2, 3, 4], random_node).await;
     assert!(event.header.parents.contains(&genesis_event_id));
     // The node adds it to their DAG, on layer 1.
+    random_node.header_dag_insert(vec![event.header.clone()]).await.unwrap();
     let event_id = random_node.dag_insert(slice::from_ref(&event)).await.unwrap()[0];
     let tips_layers = random_node.unreferenced_tips.read().await;
     // Since genesis was referenced, its layer (0) have been removed
@@ -243,7 +243,6 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     // ====================================================
     // 3. Assert that everyone has the new event in the DAG
     // ====================================================
-    info!("dag len is 2");
     assert_dags(&eg_instances, 2, &mut rng).await;
 
     // ==============================================================
@@ -253,10 +252,13 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     // ==============================================================
     let random_node = eg_instances.choose(&mut rng).unwrap();
     let event0 = Event::new(vec![1, 2, 3, 4, 0], random_node).await;
+    random_node.header_dag_insert(vec![event0.header.clone()]).await.unwrap();
     let event0_id = random_node.dag_insert(slice::from_ref(&event0)).await.unwrap()[0];
     let event1 = Event::new(vec![1, 2, 3, 4, 1], random_node).await;
+    random_node.header_dag_insert(vec![event1.header.clone()]).await.unwrap();
     let event1_id = random_node.dag_insert(slice::from_ref(&event1)).await.unwrap()[0];
     let event2 = Event::new(vec![1, 2, 3, 4, 2], random_node).await;
+    random_node.header_dag_insert(vec![event2.header.clone()]).await.unwrap();
     let event2_id = random_node.dag_insert(slice::from_ref(&event2)).await.unwrap()[0];
     // Genesis event + event from 2. + upper 3 events (layer 4)
     assert_eq!(random_node.main_dag.len(), 5);
@@ -280,7 +282,6 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     // ==========================================
     // 5. Assert that everyone has all the events
     // ==========================================
-    info!("dag len is 5");
     assert_dags(&eg_instances, 5, &mut rng).await;
 
     // ===========================================
@@ -290,57 +291,66 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     // =======
     let node1 = eg_instances.choose(&mut rng).unwrap();
     let event0_1 = Event::new(vec![1, 2, 3, 4, 3], node1).await;
+    node1.header_dag_insert(vec![event0_1.header.clone()]).await.unwrap();
     node1.dag_insert(slice::from_ref(&event0_1)).await.unwrap();
     node1.p2p.broadcast(&EventPut(event0_1)).await;
-    sleep(1).await;
+    msleep(300).await;
 
     let event1_1 = Event::new(vec![1, 2, 3, 4, 4], node1).await;
+    node1.header_dag_insert(vec![event1_1.header.clone()]).await.unwrap();
     node1.dag_insert(slice::from_ref(&event1_1)).await.unwrap();
     node1.p2p.broadcast(&EventPut(event1_1)).await;
-    sleep(1).await;
+    msleep(300).await;
 
     let event2_1 = Event::new(vec![1, 2, 3, 4, 5], node1).await;
+    node1.header_dag_insert(vec![event2_1.header.clone()]).await.unwrap();
     node1.dag_insert(slice::from_ref(&event2_1)).await.unwrap();
     node1.p2p.broadcast(&EventPut(event2_1)).await;
-    sleep(1).await;
+    msleep(300).await;
 
     // =======
     // node 2
     // =======
     let node2 = eg_instances.choose(&mut rng).unwrap();
     let event0_2 = Event::new(vec![1, 2, 3, 4, 6], node2).await;
+    node2.header_dag_insert(vec![event0_2.header.clone()]).await.unwrap();
     node2.dag_insert(slice::from_ref(&event0_2)).await.unwrap();
     node2.p2p.broadcast(&EventPut(event0_2)).await;
-    sleep(1).await;
+    msleep(300).await;
 
     let event1_2 = Event::new(vec![1, 2, 3, 4, 7], node2).await;
+    node2.header_dag_insert(vec![event1_2.header.clone()]).await.unwrap();
     node2.dag_insert(slice::from_ref(&event1_2)).await.unwrap();
     node2.p2p.broadcast(&EventPut(event1_2)).await;
-    sleep(1).await;
+    msleep(300).await;
 
     let event2_2 = Event::new(vec![1, 2, 3, 4, 8], node2).await;
+    node2.header_dag_insert(vec![event2_2.header.clone()]).await.unwrap();
     node2.dag_insert(slice::from_ref(&event2_2)).await.unwrap();
     node2.p2p.broadcast(&EventPut(event2_2)).await;
-    sleep(1).await;
+    msleep(300).await;
 
     // =======
     // node 3
     // =======
     let node3 = eg_instances.choose(&mut rng).unwrap();
     let event0_3 = Event::new(vec![1, 2, 3, 4, 9], node3).await;
+    node3.header_dag_insert(vec![event0_3.header.clone()]).await.unwrap();
     node3.dag_insert(slice::from_ref(&event0_3)).await.unwrap();
     node3.p2p.broadcast(&EventPut(event0_3)).await;
-    sleep(1).await;
+    msleep(300).await;
 
     let event1_3 = Event::new(vec![1, 2, 3, 4, 10], node3).await;
+    node3.header_dag_insert(vec![event1_3.header.clone()]).await.unwrap();
     node3.dag_insert(slice::from_ref(&event1_3)).await.unwrap();
     node3.p2p.broadcast(&EventPut(event1_3)).await;
-    sleep(1).await;
+    msleep(300).await;
 
     let event2_3 = Event::new(vec![1, 2, 3, 4, 11], node3).await;
+    node3.header_dag_insert(vec![event2_3.header.clone()]).await.unwrap();
     node3.dag_insert(slice::from_ref(&event2_3)).await.unwrap();
     node3.p2p.broadcast(&EventPut(event2_3)).await;
-    sleep(1).await;
+    msleep(300).await;
 
     // /////
     // //
@@ -360,14 +370,10 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     // node4.p2p.broadcast(&EventPut(event2_4)).await;
     // // sleep(1).await;
 
-    info!("Waiting 5s for events propagation");
-    sleep(5).await;
-
     // ==========================================
     // 7. Assert that everyone has all the events
     // ==========================================
     // 5 events from 2. and 4. + 9 events from 6. = 14
-    info!("dag len is 14 in 7.");
     assert_dags(&eg_instances, 14, &mut rng).await;
 
     // ============================================================
@@ -405,7 +411,6 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     // 9. Assert the new synced DAG has the same contents as others
     // ============================================================
     // 5 events from 2. and 4. + 9 events from 6. = 14
-    info!("dag len is 14 but in 9.");
     assert_dags(&eg_instances, 14, &mut rng).await;
 
     // Stop the P2P network
@@ -438,6 +443,7 @@ async fn eventgraph_chaotic_propagation_real(ex: Arc<Executor<'static>>) {
     for i in 0..n_events {
         let random_node = eg_instances.choose(&mut rng).unwrap();
         let event = Event::new(i.to_be_bytes().to_vec(), random_node).await;
+        random_node.header_dag_insert(vec![event.header.clone()]).await.unwrap();
         random_node.dag_insert(slice::from_ref(&event)).await.unwrap();
         random_node.p2p.broadcast(&EventPut(event)).await;
     }