[WIP] added event headers, concurrent requests from connected peers and wip sync mechanisms

Cargo.lock

@@ -1826,6 +1826,7 @@ name = "darkfi"
 version = "0.5.0"
 dependencies = [
  "arti-client",
+ "async-std",
  "async-trait",
  "blake3",
  "bs58",

Cargo.toml

@@ -52,6 +52,7 @@ parking_lot = "0.12.5"
 
 # async-runtime
 async-trait = {version = "0.1.89", optional = true}
+async-std = {version = "1.13.1", optional = true}
 futures = {version = "0.3.31", optional = true}
 smol = {version = "2.0.2", optional = true}
 pin-project-lite = {version = "0.2.16", optional = true}
@@ -182,6 +183,7 @@ geode = [
 ]
 
 event-graph = [
+    "async-std",
     "blake3",
     "num-bigint",
     "sled-overlay",

bin/darkirc/src/crypto/rln.rs

@@ -58,7 +58,7 @@ pub fn closest_epoch(timestamp: u64) -> u64 {
 /// Hash message/event modulo `Fp`
 pub fn hash_event(event: &Event) -> pallas::Base {
     let mut buf = [0u8; 64];
-    buf[..blake3::OUT_LEN].copy_from_slice(event.id().as_bytes());
+    buf[..blake3::OUT_LEN].copy_from_slice(event.header.id().as_bytes());
     pallas::Base::from_uniform_bytes(&buf)
 }
 
@@ -102,7 +102,7 @@ impl RlnIdentity {
         proving_key: &ProvingKey,
     ) -> Result<(Proof, Vec<pallas::Base>)> {
         // 1. Construct share
-        let epoch = pallas::Base::from(closest_epoch(event.timestamp));
+        let epoch = pallas::Base::from(closest_epoch(event.header.timestamp));
         let message_id = pallas::Base::from(self.message_id);
         let external_nullifier = poseidon_hash([epoch, RLN_APP_IDENTIFIER]);
         let a_0 = poseidon_hash([self.nullifier, self.trapdoor]);
@@ -132,8 +132,8 @@ impl RlnIdentity {
         let public_inputs =
             vec![epoch, external_nullifier, x, y, internal_nullifier, identity_root.inner()];
 
-        info!(target: "crypto::rln::create_proof", "[RLN] Creating proof for event {}", event.id());
-        let signal_zkbin = ZkBinary::decode(RLN2_SIGNAL_ZKBIN, false)?;
+        info!(target: "crypto::rln::create_proof", "[RLN] Creating proof for event {}", event.header.id());
+        let signal_zkbin = ZkBinary::decode(RLN2_SIGNAL_ZKBIN)?;
         let signal_circuit = ZkCircuit::new(witnesses, &signal_zkbin);
 
         let proof = Proof::create(proving_key, &[signal_circuit], &public_inputs, &mut OsRng)?;

bin/darkirc/src/irc/client.rs

@@ -193,7 +193,7 @@ impl Client {
                         Ok(Some(events)) => {
                             for event in events {
                                 // Update the last sent event.
-                                let event_id = event.id();
+                                let event_id = event.header.id();
                                 *self.last_sent.write().await = event_id;
 
                                 // If it fails for some reason, for now, we just note it and pass.
@@ -210,8 +210,8 @@ impl Client {
                                     // Also I really want GOTO in Rust... Fags.
                                     if let Some(mut rln_identity) = *self.server.rln_identity.write().await {
                                         // If the current epoch is different, we can reset the message counter
-                                        if rln_identity.last_epoch != closest_epoch(event.timestamp) {
-                                            rln_identity.last_epoch = closest_epoch(event.timestamp);
+                                        if rln_identity.last_epoch != closest_epoch(event.header.timestamp) {
+                                            rln_identity.last_epoch = closest_epoch(event.header.timestamp);
                                             rln_identity.message_id = 0;
                                         }
 
@@ -258,7 +258,7 @@ impl Client {
                 // for which the logic for delivery should be kept in sync
                 r = self.incoming.receive().fuse() => {
                     // We will skip this if it's our own message.
-                    let event_id = r.id();
+                    let event_id = r.header.id();
                     if *self.last_sent.read().await == event_id {
                         continue
                     }
@@ -623,7 +623,7 @@ impl Client {
         proof: Proof,
         public_inputs: [pallas::Base; 2],
     ) -> Result<()> {
-        let epoch = pallas::Base::from(closest_epoch(event.timestamp));
+        let epoch = pallas::Base::from(closest_epoch(event.header.timestamp));
         let external_nullifier = poseidon_hash([epoch, RLN_APP_IDENTIFIER]);
         let x = hash_event(event);
         let y = public_inputs[0];

bin/darkirc/src/irc/command.rs

@@ -919,7 +919,7 @@ impl Client {
         let mut replies = vec![];
 
         for event in dag_events.iter() {
-            let event_id = event.id();
+            let event_id = event.header.id();
             // If it was seen, skip
             match self.is_seen(&event_id).await {
                 Ok(true) => continue,

bin/darkirc/src/main.rs

@@ -126,6 +126,10 @@ struct Args {
     /// Flag to skip syncing the DAG (no history)
     skip_dag_sync: bool,
 
+    #[structopt(long)]
+    // Whether to sync headers only or full sync
+    pub fast_mode: bool,
+
     #[structopt(long)]
     /// IRC Password (Encrypted with bcrypt-2b)
     password: Option<String>,
@@ -497,7 +501,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     }
 
     // Initial DAG sync
-    if let Err(e) = sync_task(&p2p, &event_graph, args.skip_dag_sync).await {
+    if let Err(e) = sync_task(&p2p, &event_graph, args.skip_dag_sync, args.fast_mode).await {
         error!("DAG sync task failed to start: {e}");
         return Err(e);
     };
@@ -505,7 +509,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     // Stoppable task to monitor network and resync on disconnect.
     let sync_mon_task = StoppableTask::new();
     sync_mon_task.clone().start(
-        sync_and_monitor(p2p.clone(), event_graph.clone(), args.skip_dag_sync),
+        sync_and_monitor(p2p.clone(), event_graph.clone(), args.skip_dag_sync, args.fast_mode),
         |res| async move {
             match res {
                 Ok(()) | Err(Error::DetachedTaskStopped) => { /* TODO: */ }
@@ -547,7 +551,12 @@ async fn monitor_network(subscription: &Subscription<Error>) -> Result<()> {
 }
 
 /// Async task to endlessly try to sync DAG, returns Ok if done.
-async fn sync_task(p2p: &P2pPtr, event_graph: &EventGraphPtr, skip_dag_sync: bool) -> Result<()> {
+async fn sync_task(
+    p2p: &P2pPtr,
+    event_graph: &EventGraphPtr,
+    skip_dag_sync: bool,
+    fast_mode: bool,
+) -> Result<()> {
     let comms_timeout = p2p.settings().read_arc().await.outbound_connect_timeout_max();
 
     loop {
@@ -556,7 +565,7 @@ async fn sync_task(p2p: &P2pPtr, event_graph: &EventGraphPtr, skip_dag_sync: boo
             // We'll attempt to sync for ever
             if !skip_dag_sync {
                 info!("Syncing event DAG");
-                match event_graph.dag_sync().await {
+                match event_graph.dag_sync(fast_mode).await {
                     Ok(()) => break,
                     Err(e) => {
                         // TODO: Maybe at this point we should prune or something?
@@ -583,6 +592,7 @@ async fn sync_and_monitor(
     p2p: P2pPtr,
     event_graph: EventGraphPtr,
     skip_dag_sync: bool,
+    fast_mode: bool,
 ) -> Result<()> {
     loop {
         let net_subscription = p2p.hosts().subscribe_disconnect().await;
@@ -595,7 +605,7 @@ async fn sync_and_monitor(
                 // Sync node again
                 info!("Network disconnection detected, resyncing...");
                 *event_graph.synced.write().await = false;
-                sync_task(&p2p, &event_graph, skip_dag_sync).await?;
+                sync_task(&p2p, &event_graph, skip_dag_sync, fast_mode).await?;
             }
             Err(e) => return Err(e),
         }

bin/genev/genevd/src/main.rs

@@ -78,6 +78,10 @@ struct Args {
     /// Flag to skip syncing the DAG (no history)
     skip_dag_sync: bool,
 
+    #[structopt(long)]
+    // Whether to sync headers only or full sync
+    pub fast_mode: bool,
+
     #[structopt(short, parse(from_occurrences))]
     /// Increase verbosity (-vvv supported)
     verbose: u8,
@@ -92,7 +96,7 @@ async fn start_sync_loop(
     let seen_events = seen.get().unwrap();
     loop {
         let event = incoming.receive().await;
-        let event_id = event.id();
+        let event_id = event.header.id();
         if *last_sent.read().await == event_id {
             continue
         }
@@ -154,7 +158,7 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
     if !settings.skip_dag_sync {
         for i in 1..=6 {
             info!("Syncing event DAG (attempt #{i})");
-            match event_graph.dag_sync().await {
+            match event_graph.dag_sync(settings.fast_mode).await {
                 Ok(()) => break,
                 Err(e) => {
                     if i == 6 {

bin/genev/genevd/src/rpc.rs

@@ -229,7 +229,7 @@ impl JsonRpcInterface {
         let dag_events = self.event_graph.order_events().await;
 
         for event in dag_events.iter() {
-            let event_id = event.id();
+            let event_id = event.header.id();
             // Try to deserialize it. (Here we skip errors)
             let genevent: GenEvent = match deserialize_async_partial(event.content()).await {
                 Ok((v, _)) => v,

bin/tau/taud/src/main.rs

@@ -332,7 +332,7 @@ async fn start_sync_loop(
             }
             // Process message from the network. These should only be EncryptedTask.
             task_event = incoming.receive().fuse() => {
-                let event_id = task_event.id();
+                let event_id = task_event.header.id();
                 if is_seen(sled_db.clone(), seen.clone(), &event_id).await? {
                     continue
                 }
@@ -557,7 +557,7 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
             // We'll attempt to sync for ever
             if !settings.skip_dag_sync {
                 info!(target: "taud", "Syncing event DAG");
-                match event_graph.dag_sync().await {
+                match event_graph.dag_sync(settings.fast_mode).await {
                     Ok(()) => break,
                     Err(e) => {
                         // TODO: Maybe at this point we should prune or something?
@@ -585,7 +585,7 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
     let dag_events = event_graph.order_events().await;
 
     for event in dag_events.iter() {
-        let event_id = event.id();
+        let event_id = event.header.id();
         // If it was seen, skip
         if is_seen(sled_db.clone(), seen.clone(), &event_id).await? {
             continue

bin/tau/taud/src/settings.rs

@@ -94,6 +94,10 @@ pub struct Args {
     // Whether to pipe notifications or not
     pub piped: bool,
 
+    #[structopt(long)]
+    // Whether to sync headers only or full sync
+    pub fast_mode: bool,
+
     #[structopt(short, long)]
     /// Set log file to ouput into
     pub log: Option<String>,

src/event_graph/event.rs

@@ -28,67 +28,141 @@ use super::{
     N_EVENT_PARENTS,
 };
 
-/// Representation of an event in the Event Graph
 #[derive(Debug, Clone, PartialEq, SerialEncodable, SerialDecodable)]
-pub struct Event {
+pub struct Header {
+    /// Event version
+    // pub version: u8,
     /// Timestamp of the event in whole seconds
     pub timestamp: u64,
-    /// Content of the event
-    pub content: Vec<u8>,
     /// Parent nodes in the event DAG
     pub parents: [blake3::Hash; N_EVENT_PARENTS],
     /// DAG layer index of the event
     pub layer: u64,
 }
 
-impl Event {
-    /// Create a new event with the given data and an [`EventGraph`] reference.
-    /// The timestamp of the event will be the current time, and the parents
-    /// will be `N_EVENT_PARENTS` from the current event graph unreferenced tips.
-    /// The parents can also include NULL, but this should be handled by the rest
-    /// of the codebase.
-    pub async fn new(data: Vec<u8>, event_graph: &EventGraph) -> Self {
+impl Header {
+    // Create a new Header given EventGraph to retrieve the correct layout
+    pub async fn new(event_graph: &EventGraph) -> Self {
         let (layer, parents) = event_graph.get_next_layer_with_parents().await;
-        Self {
-            timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64,
-            content: data,
-            parents,
-            layer,
-        }
+        Self { timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64, parents, layer }
     }
 
-    /// Same as `Event::new()` but allows specifying the timestamp explicitly.
-    pub async fn with_timestamp(timestamp: u64, data: Vec<u8>, event_graph: &EventGraph) -> Self {
+    pub async fn with_timestamp(timestamp: u64, event_graph: &EventGraph) -> Self {
         let (layer, parents) = event_graph.get_next_layer_with_parents().await;
-        Self { timestamp, content: data, parents, layer }
+        Self { timestamp, parents, layer }
     }
 
-    /// Hash the [`Event`] to retrieve its ID
+    /// Hash the [`Header`] to retrieve its ID
     pub fn id(&self) -> blake3::Hash {
         let mut hasher = blake3::Hasher::new();
         self.timestamp.encode(&mut hasher).unwrap();
-        self.content.encode(&mut hasher).unwrap();
         self.parents.encode(&mut hasher).unwrap();
         self.layer.encode(&mut hasher).unwrap();
         hasher.finalize()
     }
 
+    /// Fully validate a header for the correct layout against provided
+    /// DAG [`sled::Tree`] reference and enforce relevant age, assuming
+    /// some possibility for a time drift. Optionally, provide an overlay
+    /// to use that instead of actual referenced DAG.
+    pub async fn validate(
+        &self,
+        dag: &sled::Tree,
+        genesis_timestamp: u64,
+        days_rotation: u64,
+        overlay: Option<&SledTreeOverlay>,
+    ) -> Result<bool> {
+        // Check if the event timestamp is after genesis timestamp
+        if self.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.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.id();
+
+        for parent_id in self.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: Header = deserialize_async(&parent_bytes.unwrap()).await?;
+            if self.layer <= parent.layer {
+                return Ok(false)
+            }
+
+            seen.insert(parent_id);
+        }
+
+        Ok(!seen.is_empty())
+    }
+}
+
+/// Representation of an event in the Event Graph
+#[derive(Debug, Clone, PartialEq, SerialEncodable, SerialDecodable)]
+pub struct Event {
+    pub header: Header,
+    /// Content of the event
+    pub content: Vec<u8>,
+}
+
+impl Event {
+    /// Create a new event with the given data and an [`EventGraph`] reference.
+    /// The timestamp of the event will be the current time, and the parents
+    /// will be `N_EVENT_PARENTS` from the current event graph unreferenced tips.
+    /// The parents can also include NULL, but this should be handled by the rest
+    /// of the codebase.
+    pub async fn new(data: Vec<u8>, event_graph: &EventGraph) -> Self {
+        let header = Header::new(event_graph).await;
+        Self { header, content: data }
+    }
+
+    /// Same as `Event::new()` but allows specifying the timestamp explicitly.
+    pub async fn with_timestamp(timestamp: u64, data: Vec<u8>, event_graph: &EventGraph) -> Self {
+        let header = Header::with_timestamp(timestamp, event_graph).await;
+        Self { header, content: data }
+    }
+
     /// Return a reference to the event's content
     pub fn content(&self) -> &[u8] {
         &self.content
     }
 
-    /*
-    /// Check if an [`Event`] is considered too old.
-    fn is_too_old(&self) -> bool {
-        self.timestamp < UNIX_EPOCH.elapsed().unwrap().as_secs() - ORPHAN_AGE_LIMIT
-    }
-    */
-
     /// Fully validate an event for the correct layout against provided
     /// DAG [`sled::Tree`] reference and enforce relevant age, assuming
     /// some possibility for a time drift. Optionally, provide an overlay
     /// 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,
@@ -102,7 +176,7 @@ impl Event {
         }
 
         // Check if the event timestamp is after genesis timestamp
-        if self.timestamp < genesis_timestamp - EVENT_TIME_DRIFT {
+        if self.header.timestamp < genesis_timestamp - EVENT_TIME_DRIFT {
             return Ok(false)
         }
 
@@ -110,7 +184,7 @@ impl Event {
         // is after the next genesis timestamp
         if days_rotation > 0 {
             let next_genesis_timestamp = next_rotation_timestamp(INITIAL_GENESIS, days_rotation);
-            if self.timestamp > next_genesis_timestamp + EVENT_TIME_DRIFT {
+            if self.header.timestamp > next_genesis_timestamp + EVENT_TIME_DRIFT {
                 return Ok(false)
             }
         }
@@ -120,9 +194,9 @@ impl Event {
         // same, and that the parent exists in previous layers, to prevent
         // recursive references(circles).
         let mut seen = HashSet::new();
-        let self_id = self.id();
+        let self_id = self.header.id();
 
-        for parent_id in self.parents.iter() {
+        for parent_id in self.header.parents.iter() {
             if parent_id == &NULL_ID {
                 continue
             }
@@ -145,7 +219,7 @@ impl Event {
             }
 
             let parent: Event = deserialize_async(&parent_bytes.unwrap()).await?;
-            if self.layer <= parent.layer {
+            if self.header.layer <= parent.header.layer {
                 return Ok(false)
             }
 
@@ -160,10 +234,11 @@ 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.timestamp;
+        let genesis_timestamp = event_graph.current_genesis.read().await.header.timestamp;
 
         // Perform validation
-        self.validate(&event_graph.dag, genesis_timestamp, event_graph.days_rotation, None).await
+        self.validate(&event_graph.header_dag, genesis_timestamp, event_graph.days_rotation, None)
+            .await
     }
 
     /// Validate a new event for the correct layout and enforce relevant age,
@@ -178,8 +253,8 @@ impl Event {
 
         // Check if the event is too old or too new
         let now = UNIX_EPOCH.elapsed().unwrap().as_millis() as u64;
-        let too_old = self.timestamp < now - EVENT_TIME_DRIFT;
-        let too_new = self.timestamp > now + EVENT_TIME_DRIFT;
+        let too_old = self.header.timestamp < now - EVENT_TIME_DRIFT;
+        let too_new = self.header.timestamp > now + EVENT_TIME_DRIFT;
         if too_old || too_new {
             return false
         }
@@ -187,9 +262,9 @@ impl Event {
         // Validate the parents. We have to check that at least one parent
         // is not NULL and that no two parents are the same.
         let mut seen = HashSet::new();
-        let self_id = self.id();
+        let self_id = self.header.id();
 
-        for parent_id in self.parents.iter() {
+        for parent_id in self.header.parents.iter() {
             if parent_id == &NULL_ID {
                 continue
             }
@@ -260,24 +335,24 @@ mod tests {
             assert!(!event_empty_content.dag_validate(&event_graph).await?);
 
             let mut event_timestamp_too_old = valid_event.clone();
-            event_timestamp_too_old.timestamp = 0;
+            event_timestamp_too_old.header.timestamp = 0;
             assert!(!event_timestamp_too_old.dag_validate(&event_graph).await?);
 
             let mut event_timestamp_too_new = valid_event.clone();
-            event_timestamp_too_new.timestamp = u64::MAX;
+            event_timestamp_too_new.header.timestamp = u64::MAX;
             assert!(!event_timestamp_too_new.dag_validate(&event_graph).await?);
 
             let mut event_duplicated_parents = valid_event.clone();
-            event_duplicated_parents.parents[1] = valid_event.parents[0];
+            event_duplicated_parents.header.parents[1] = valid_event.header.parents[0];
             assert!(!event_duplicated_parents.dag_validate(&event_graph).await?);
 
             let mut event_null_parents = valid_event.clone();
             let all_null_parents = [NULL_ID, NULL_ID, NULL_ID, NULL_ID, NULL_ID];
-            event_null_parents.parents = all_null_parents;
+            event_null_parents.header.parents = all_null_parents;
             assert!(!event_null_parents.dag_validate(&event_graph).await?);
 
             let mut event_same_layer_as_parents = valid_event.clone();
-            event_same_layer_as_parents.layer = 0;
+            event_same_layer_as_parents.header.layer = 0;
             assert!(!event_same_layer_as_parents.dag_validate(&event_graph).await?);
 
             // Thanks for reading

src/event_graph/mod.rs

@@ -16,13 +16,22 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+use async_std::stream::from_iter;
+use futures::{
+    future,
+    stream::{FuturesOrdered, FuturesUnordered},
+    StreamExt,
+};
+use rand::{rngs::OsRng, seq::SliceRandom};
 use std::{
     collections::{BTreeMap, HashMap, HashSet, VecDeque},
     path::PathBuf,
     sync::Arc,
 };
 
+use blake3::Hash;
 use darkfi_serial::{deserialize_async, serialize_async};
+use event::Header;
 use num_bigint::BigUint;
 use sled_overlay::{sled, SledTreeOverlay};
 use smol::{
@@ -33,7 +42,7 @@ use tracing::{debug, error, info, warn};
 
 use crate::{
     event_graph::util::replayer_log,
-    net::P2pPtr,
+    net::{channel::Channel, P2pPtr},
     system::{msleep, Publisher, PublisherPtr, StoppableTask, StoppableTaskPtr, Subscription},
     Error, Result,
 };
@@ -53,7 +62,7 @@ pub use event::Event;
 
 /// P2P protocol implementation for the Event Graph
 pub mod proto;
-use proto::{EventRep, EventReq, TipRep, TipReq};
+use proto::{EventRep, EventReq, HeaderRep, HeaderReq, TipRep, TipReq};
 
 /// Utility functions
 pub mod util;
@@ -77,7 +86,7 @@ pub const N_EVENT_PARENTS: usize = 5;
 /// Allowed timestamp drift in milliseconds
 const EVENT_TIME_DRIFT: u64 = 60_000;
 /// Null event ID
-pub const NULL_ID: blake3::Hash = blake3::Hash::from_bytes([0x00; blake3::OUT_LEN]);
+pub const NULL_ID: Hash = Hash::from_bytes([0x00; blake3::OUT_LEN]);
 
 /// Atomic pointer to an [`EventGraph`] instance.
 pub type EventGraphPtr = Arc<EventGraph>;
@@ -86,21 +95,23 @@ pub type EventGraphPtr = Arc<EventGraph>;
 pub struct EventGraph {
     /// Pointer to the P2P network instance
     p2p: P2pPtr,
-    /// Sled tree containing the DAG
-    dag: sled::Tree,
+    /// Sled tree containing the headers
+    header_dag: sled::Tree,
+    /// Main sled tree containing the events
+    main_dag: sled::Tree,
     /// Replay logs path.
     datastore: PathBuf,
     /// Run in replay_mode where if set we log Sled DB instructions
     /// into `datastore`, useful to reacreate a faulty DAG to debug.
     replay_mode: bool,
     /// The set of unreferenced DAG tips
-    unreferenced_tips: RwLock<BTreeMap<u64, HashSet<blake3::Hash>>>,
+    unreferenced_tips: RwLock<BTreeMap<u64, HashSet<Hash>>>,
     /// A `HashSet` containg event IDs and their 1-level parents.
     /// These come from the events we've sent out using `EventPut`.
     /// They are used with `EventReq` to decide if we should reply
     /// or not. Additionally it is also used when we broadcast the
     /// `TipRep` message telling peers about our unreferenced tips.
-    broadcasted_ids: RwLock<HashSet<blake3::Hash>>,
+    broadcasted_ids: RwLock<HashSet<Hash>>,
     /// DAG Pruning Task
     pub prune_task: OnceCell<StoppableTaskPtr>,
     /// Event publisher, this notifies whenever an event is
@@ -141,6 +152,8 @@ impl EventGraph {
         days_rotation: u64,
         ex: Arc<Executor<'_>>,
     ) -> Result<EventGraphPtr> {
+        let hdr_tree_name = format!("headers_{dag_tree_name}");
+        let hdr_dag = sled_db.open_tree(hdr_tree_name)?;
         let dag = sled_db.open_tree(dag_tree_name)?;
         let unreferenced_tips = RwLock::new(BTreeMap::new());
         let broadcasted_ids = RwLock::new(HashSet::new());
@@ -150,7 +163,8 @@ impl EventGraph {
         let current_genesis = generate_genesis(days_rotation);
         let self_ = Arc::new(Self {
             p2p,
-            dag: dag.clone(),
+            header_dag: hdr_dag.clone(),
+            main_dag: dag.clone(),
             datastore,
             replay_mode,
             unreferenced_tips,
@@ -166,7 +180,7 @@ impl EventGraph {
 
         // Check if we have it in our DAG.
         // If not, we can prune the DAG and insert this new genesis event.
-        if !dag.contains_key(current_genesis.id().as_bytes())? {
+        if !dag.contains_key(current_genesis.header.id().as_bytes())? {
             info!(
                 target: "event_graph::new",
                 "[EVENTGRAPH] DAG does not contain current genesis, pruning existing data",
@@ -183,16 +197,16 @@ impl EventGraph {
             let _ = self_.prune_task.set(prune_task.clone()).await;
 
             prune_task.clone().start(
-                self_.clone().dag_prune_task(days_rotation),
-                |res| async move {
-                    match res {
-                        Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
-                        Err(e) => error!(target: "event_graph::_handle_stop", "[EVENTGRAPH] Failed stopping prune task: {e}")
-                    }
-                },
-                Error::DetachedTaskStopped,
-                ex.clone(),
-            );
+                 self_.clone().dag_prune_task(days_rotation),
+                 |res| async move {
+                     match res {
+                         Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                         Err(e) => error!(target: "event_graph::_handle_stop", "[EVENTGRAPH] Failed stopping prune task: {e}")
+                     }
+                 },
+                 Error::DetachedTaskStopped,
+                 ex.clone(),
+             );
         }
 
         Ok(self_)
@@ -202,8 +216,28 @@ 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) -> Result<()> {
+    pub async fn dag_sync(&self, fast_mode: bool) -> Result<()> {
         // We do an optimistic sync where we ask all our connected peers for
         // the latest layer DAG tips (unreferenced events) and then we accept
         // the ones we see the most times.
@@ -227,8 +261,10 @@ impl EventGraph {
             "[EVENTGRAPH] Syncing DAG from {communicated_peers} peers..."
         );
 
+        let comms_timeout = self.p2p.settings().read().await.outbound_connect_timeout_max();
+
         // Here we keep track of the tips, their layers and how many time we've seen them.
-        let mut tips: HashMap<blake3::Hash, (u64, usize)> = HashMap::new();
+        let mut tips: HashMap<Hash, (u64, usize)> = HashMap::new();
 
         // Let's first ask all of our peers for their tips and collect them
         // in our hashmap above.
@@ -256,15 +292,8 @@ impl EventGraph {
                 continue
             };
 
-            let outbound_connect_timeout = self
-                .p2p
-                .settings()
-                .read_arc()
-                .await
-                .outbound_connect_timeout(channel.address().scheme());
             // Node waits for response
-            let Ok(peer_tips) = tip_rep_sub.receive_with_timeout(outbound_connect_timeout).await
-            else {
+            let Ok(peer_tips) = tip_rep_sub.receive_with_timeout(comms_timeout).await else {
                 error!(
                     target: "event_graph::dag_sync",
                     "[EVENTGRAPH] Sync: Peer {url} didn't reply with tips in time, skipping"
@@ -273,7 +302,7 @@ impl EventGraph {
                 continue
             };
 
-            let peer_tips = &peer_tips.0;
+            let peer_tips: &BTreeMap<u64, HashSet<Hash>> = &peer_tips.0;
 
             // Note down the seen tips
             for (layer, layer_tips) in peer_tips {
@@ -313,7 +342,7 @@ impl EventGraph {
         for tip in considered_tips.iter() {
             assert!(tip != &NULL_ID);
 
-            if !self.dag.contains_key(tip.as_bytes()).unwrap() {
+            if !self.main_dag.contains_key(tip.as_bytes()).unwrap() {
                 missing_parents.insert(*tip);
             }
         }
@@ -324,7 +353,92 @@ impl EventGraph {
             return Ok(())
         }
 
-        info!(target: "event_graph::dag_sync", "[EVENTGRAPH] Fetching events");
+        // Header sync first
+        // TODO: requesting headers should be in a way that we wouldn't
+        // recieve the same header(s) again, by sending our tip, other
+        // nodes should send back the ones after it
+        let mut headers_requests = FuturesUnordered::new();
+        for channel in channels.iter() {
+            headers_requests.push(request_header(&channel, comms_timeout))
+        }
+
+        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 mut header_sorted = vec![];
+
+            for iter_elem in self.header_dag.iter() {
+                let (_, val) = iter_elem.unwrap();
+                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
+            // 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 mut rcvd_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!("layer: {}", i.header.layer);
+            //     }
+            //     rcvd_events.extend(events);
+            // }
+
+            // self.dag_insert(&rcvd_events).await?;
+
+            // // 2. split sorted headers into chunks and assign them to each connected peer
+            // let mut responses = vec![];
+            // for header in header_sorted.chunks_exact(peers.len()) {
+            //     // For each peer, create a future that sends a request
+            //     let pairs = peers.iter().zip(header).collect::<Vec<_>>();
+            //     let pair_stream = from_iter(pairs.iter());
+            //     let requests_stream = pair_stream.map(|(peer, header)| send_request(peer, header));
+            //     // Collect all the responses into a vector
+            //     let x = requests_stream.collect::<Vec<_>>().await;
+            //     info!("len of x: {}", x.len());
+            //     // responses.push(x);
+            //     responses.extend(x);
+            // }
+            // // Wait for all the futures to complete
+            // let x = future::join_all(responses).await;
+            // let fetched_parents = x.into_iter().map(|f| f.unwrap()).collect::<Vec<_>>().concat();
+            // info!("fetched parents: {}", fetched_parents.len());
+            // for i in fetched_parents.iter() {
+            //     info!("layer: {}", i.header.layer)
+            // }
+
+            // // 3. Fetch all events at once (just a POC)
+            // let peers = channels.clone().into_iter().collect::<Vec<_>>();
+            // let missing = header_sorted.iter().map(|x| x.id()).collect::<Vec<_>>();
+            // info!("first missing: {}", missing[0]);
+            // 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();
 
@@ -359,15 +473,8 @@ impl EventGraph {
                     continue
                 }
 
-                let outbound_connect_timeout = self
-                    .p2p
-                    .settings()
-                    .read_arc()
-                    .await
-                    .outbound_connect_timeout(channel.address().scheme());
                 // Node waits for response
-                let Ok(parent) = ev_rep_sub.receive_with_timeout(outbound_connect_timeout).await
-                else {
+                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}"
@@ -378,12 +485,12 @@ impl EventGraph {
                 let parents = parent.0.clone();
 
                 for parent in parents {
-                    let parent_id = parent.id();
+                    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.id()
+                            parent.header.id()
                         );
                         continue
                     }
@@ -393,11 +500,11 @@ impl EventGraph {
                         "Got correct parent event {parent_id}"
                     );
 
-                    if let Some(layer_events) = received_events.get_mut(&parent.layer) {
+                    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.layer, layer_events);
+                        received_events.insert(parent.header.layer, layer_events);
                     }
                     received_events_hashes.insert(parent_id);
 
@@ -405,14 +512,14 @@ impl EventGraph {
                     found_event = true;
 
                     // See if we have the upper parents
-                    for upper_parent in parent.parents.iter() {
+                    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.dag.contains_key(upper_parent.as_bytes()).unwrap()
+                            !self.main_dag.contains_key(upper_parent.as_bytes()).unwrap()
                         {
                             debug!(
                                 target: "event_graph::dag_sync",
@@ -445,6 +552,8 @@ impl EventGraph {
         }
         self.dag_insert(&events).await?;
 
+        // <-- end download payload
+
         *self.synced.write().await = true;
 
         info!(target: "event_graph::dag_sync", "[EVENTGRAPH] DAG synced successfully!");
@@ -464,21 +573,37 @@ impl EventGraph {
         let mut broadcasted_ids = self.broadcasted_ids.write().await;
         let mut current_genesis = self.current_genesis.write().await;
 
-        // Atomically clear the DAG and write the new genesis event.
+        // Atomically clear the main and headers DAGs and write the new genesis event.
+        // Header
+        let mut batch = sled::Batch::default();
+        for key in self.header_dag.iter().keys() {
+            batch.remove(key.unwrap());
+        }
+        batch.insert(
+            genesis_event.header.id().as_bytes(),
+            serialize_async(&genesis_event.header).await,
+        );
+
+        debug!(target: "event_graph::dag_prune", "Applying header batch...");
+        if let Err(e) = self.header_dag.apply_batch(batch) {
+            panic!("Failed pruning header DAG, sled apply_batch error: {}", e);
+        }
+
+        // Main
         let mut batch = sled::Batch::default();
-        for key in self.dag.iter().keys() {
+        for key in self.main_dag.iter().keys() {
             batch.remove(key.unwrap());
         }
-        batch.insert(genesis_event.id().as_bytes(), serialize_async(&genesis_event).await);
+        batch.insert(genesis_event.header.id().as_bytes(), serialize_async(&genesis_event).await);
 
-        debug!(target: "event_graph::dag_prune", "Applying batch...");
-        if let Err(e) = self.dag.apply_batch(batch) {
-            panic!("Failed pruning DAG, sled apply_batch error: {e}");
+        debug!(target: "event_graph::dag_prune", "Applying main batch...");
+        if let Err(e) = self.main_dag.apply_batch(batch) {
+            panic!("Failed pruning main DAG, sled apply_batch error: {e}");
         }
 
         // Clear unreferenced tips and bcast ids
         *unreferenced_tips = BTreeMap::new();
-        unreferenced_tips.insert(0, HashSet::from([genesis_event.id()]));
+        unreferenced_tips.insert(0, HashSet::from([genesis_event.header.id()]));
         *current_genesis = genesis_event;
         *broadcasted_ids = HashSet::new();
         drop(unreferenced_tips);
@@ -501,13 +626,10 @@ impl EventGraph {
             // Find the next rotation timestamp:
             let next_rotation = next_rotation_timestamp(INITIAL_GENESIS, days_rotation);
 
+            let header =
+                Header { timestamp: next_rotation, parents: [NULL_ID; N_EVENT_PARENTS], layer: 0 };
             // Prepare the new genesis event
-            let current_genesis = Event {
-                timestamp: next_rotation,
-                content: GENESIS_CONTENTS.to_vec(),
-                parents: [NULL_ID; N_EVENT_PARENTS],
-                layer: 0,
-            };
+            let current_genesis = Event { header, content: GENESIS_CONTENTS.to_vec() };
 
             // Sleep until it's time to rotate.
             let s = millis_until_next_rotation(next_rotation);
@@ -531,7 +653,7 @@ impl EventGraph {
     /// knows that any requests for them are actually legitimate.
     /// TODO: The `broadcasted_ids` set should periodically be pruned, when
     /// some sensible time has passed after broadcasting the event.
-    pub async fn dag_insert(&self, events: &[Event]) -> Result<Vec<blake3::Hash>> {
+    pub async fn dag_insert(&self, events: &[Event]) -> Result<Vec<Hash>> {
         // Sanity check
         if events.is_empty() {
             return Ok(vec![])
@@ -545,22 +667,25 @@ impl EventGraph {
         let mut ids = Vec::with_capacity(events.len());
 
         // Create an overlay over the DAG tree
-        let mut overlay = SledTreeOverlay::new(&self.dag);
+        let mut overlay = SledTreeOverlay::new(&self.main_dag);
 
         // Grab genesis timestamp
-        let genesis_timestamp = self.current_genesis.read().await.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.id();
+            let event_id = event.header.id();
+            if event.header.layer == 0 {
+                continue
+            }
             debug!(
                 target: "event_graph::dag_insert",
                 "Inserting event {event_id} into the DAG"
             );
 
             if !event
-                .validate(&self.dag, genesis_timestamp, self.days_rotation, Some(&overlay))
+                .validate(&self.main_dag, genesis_timestamp, self.days_rotation, Some(&overlay))
                 .await?
             {
                 error!(target: "event_graph::dag_insert", "Event {event_id} is invalid!");
@@ -584,21 +709,21 @@ impl EventGraph {
 
         // Atomically apply the batch.
         // Panic if something is corrupted.
-        if let Err(e) = self.dag.apply_batch(batch) {
+        if let Err(e) = self.main_dag.apply_batch(batch) {
             panic!("Failed applying dag_insert batch to sled: {e}");
         }
 
         // Iterate over given events to update references and
         // send out notifications about them
         for event in events {
-            let event_id = event.id();
+            let event_id = event.header.id();
 
             // Update the unreferenced DAG tips set
             debug!(
                 target: "event_graph::dag_insert",
-                "Event {event_id} parents {:#?}", event.parents,
+                "Event {event_id} parents {:#?}", event.header.parents,
             );
-            for parent_id in event.parents.iter() {
+            for parent_id in event.header.parents.iter() {
                 if parent_id != &NULL_ID {
                     debug!(
                         target: "event_graph::dag_insert",
@@ -611,7 +736,7 @@ impl EventGraph {
                     // assumption is that previous layers unreferenced
                     // tips will be few.
                     for (layer, tips) in unreferenced_tips.iter_mut() {
-                        if layer >= &event.layer {
+                        if layer >= &event.header.layer {
                             continue
                         }
                         tips.remove(parent_id);
@@ -625,12 +750,12 @@ impl EventGraph {
                 "Adding {event_id} to unreferenced tips"
             );
 
-            if let Some(layer_tips) = unreferenced_tips.get_mut(&event.layer) {
+            if let Some(layer_tips) = unreferenced_tips.get_mut(&event.header.layer) {
                 layer_tips.insert(event_id);
             } else {
                 let mut layer_tips = HashSet::new();
                 layer_tips.insert(event_id);
-                unreferenced_tips.insert(event.layer, layer_tips);
+                unreferenced_tips.insert(event.header.layer, layer_tips);
             }
 
             // Send out notifications about the new event
@@ -644,9 +769,55 @@ impl EventGraph {
         Ok(ids)
     }
 
+    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;
+
+        let mut hdrs = headers;
+        hdrs.sort_by(|x, y| x.layer.cmp(&y.layer));
+
+        // Iterate over given events to validate them and
+        // write them to the overlay
+        for header in hdrs {
+            let header_id = header.id();
+            if header.layer == 0 {
+                continue
+            }
+            debug!(
+                target: "event_graph::header_dag_insert()",
+                "Inserting event {} 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)
+            }
+            let header_se = serialize_async(&header).await;
+
+            // Add the event to the overlay
+            overlay.insert(header_id.as_bytes(), &header_se)?;
+        }
+
+        // Aggregate changes into a single batch
+        let batch = overlay.aggregate().unwrap();
+
+        // Atomically apply the batch.
+        // Panic if something is corrupted.
+        if let Err(e) = self.header_dag.apply_batch(batch) {
+            panic!("Failed applying dag_insert batch to sled: {}", e);
+        }
+
+        Ok(())
+    }
+
     /// Fetch an event from the DAG
-    pub async fn dag_get(&self, event_id: &blake3::Hash) -> Result<Option<Event>> {
-        let Some(bytes) = self.dag.get(event_id.as_bytes())? else { return Ok(None) };
+    pub async fn dag_get(&self, event_id: &Hash) -> Result<Option<Event>> {
+        let Some(bytes) = self.main_dag.get(event_id.as_bytes())? else { return Ok(None) };
         let event: Event = deserialize_async(&bytes).await?;
 
         Ok(Some(event))
@@ -656,7 +827,7 @@ impl EventGraph {
     /// tips of the DAG. Since tips are mapped by their layer, we go backwards
     /// until we fill the vector, ensuring we always use latest layers tips as
     /// parents.
-    async fn get_next_layer_with_parents(&self) -> (u64, [blake3::Hash; N_EVENT_PARENTS]) {
+    async fn get_next_layer_with_parents(&self) -> (u64, [Hash; N_EVENT_PARENTS]) {
         let unreferenced_tips = self.unreferenced_tips.read().await;
 
         let mut parents = [NULL_ID; N_EVENT_PARENTS];
@@ -666,7 +837,7 @@ impl EventGraph {
                 parents[index] = *tip;
                 index += 1;
                 if index >= N_EVENT_PARENTS {
-                    break 'outer
+                    break 'outer;
                 }
             }
         }
@@ -678,34 +849,34 @@ impl EventGraph {
     }
 
     /// Find the unreferenced tips in the current DAG state, mapped by their layers.
-    async fn find_unreferenced_tips(&self) -> BTreeMap<u64, HashSet<blake3::Hash>> {
+    async fn find_unreferenced_tips(&self) -> BTreeMap<u64, HashSet<Hash>> {
         // First get all the event IDs
         let mut tips = HashSet::new();
-        for iter_elem in self.dag.iter() {
+        for iter_elem in self.main_dag.iter() {
             let (id, _) = iter_elem.unwrap();
-            let id = blake3::Hash::from_bytes((&id as &[u8]).try_into().unwrap());
+            let id = Hash::from_bytes((&id as &[u8]).try_into().unwrap());
             tips.insert(id);
         }
 
         // Iterate again to find unreferenced IDs
-        for iter_elem in self.dag.iter() {
+        for iter_elem in self.main_dag.iter() {
             let (_, event) = iter_elem.unwrap();
             let event: Event = deserialize_async(&event).await.unwrap();
-            for parent in event.parents.iter() {
+            for parent in event.header.parents.iter() {
                 tips.remove(parent);
             }
         }
 
         // Build the layers map
-        let mut map: BTreeMap<u64, HashSet<blake3::Hash>> = BTreeMap::new();
+        let mut map: BTreeMap<u64, HashSet<Hash>> = BTreeMap::new();
         for tip in tips {
             let event = self.dag_get(&tip).await.unwrap().unwrap();
-            if let Some(layer_tips) = map.get_mut(&event.layer) {
+            if let Some(layer_tips) = map.get_mut(&event.header.layer) {
                 layer_tips.insert(tip);
             } else {
                 let mut layer_tips = HashSet::new();
                 layer_tips.insert(tip);
-                map.insert(event.layer, layer_tips);
+                map.insert(event.header.layer, layer_tips);
             }
         }
 
@@ -713,7 +884,7 @@ impl EventGraph {
     }
 
     /// Internal function used for DAG sorting.
-    async fn get_unreferenced_tips_sorted(&self) -> [blake3::Hash; N_EVENT_PARENTS] {
+    async fn get_unreferenced_tips_sorted(&self) -> [Hash; N_EVENT_PARENTS] {
         let (_, tips) = self.get_next_layer_with_parents().await;
 
         // Convert the hash to BigUint for sorting
@@ -731,7 +902,7 @@ impl EventGraph {
                 bytes.insert(0, 0);
             }
 
-            tips_sorted[i] = blake3::Hash::from_bytes(bytes.try_into().unwrap());
+            tips_sorted[i] = Hash::from_bytes(bytes.try_into().unwrap());
         }
 
         tips_sorted
@@ -751,8 +922,9 @@ impl EventGraph {
 
         let mut ord_events_vec = ordered_events.make_contiguous().to_vec();
         // Order events based on thier layer numbers, or based on timestamp if they are equal
-        ord_events_vec
-            .sort_unstable_by(|a, b| a.0.cmp(&b.0).then(b.1.timestamp.cmp(&a.1.timestamp)));
+        ord_events_vec.sort_unstable_by(|a, b| {
+            a.0.cmp(&b.0).then(b.1.header.timestamp.cmp(&a.1.header.timestamp))
+        });
 
         ord_events_vec.iter().map(|a| a.1.clone()).collect::<Vec<Event>>()
     }
@@ -762,22 +934,22 @@ impl EventGraph {
     async fn dfs_topological_sort(
         &self,
         event: Event,
-        visited: &mut HashSet<blake3::Hash>,
+        visited: &mut HashSet<Hash>,
     ) -> VecDeque<(u64, Event)> {
         let mut ordered_events = VecDeque::new();
         let mut stack = VecDeque::new();
-        let event_id = event.id();
+        let event_id = event.header.id();
         stack.push_back(event_id);
 
         while let Some(event_id) = stack.pop_front() {
             if !visited.contains(&event_id) && event_id != NULL_ID {
                 visited.insert(event_id);
                 if let Some(event) = self.dag_get(&event_id).await.unwrap() {
-                    for parent in event.parents.iter() {
+                    for parent in event.header.parents.iter() {
                         stack.push_back(*parent);
                     }
 
-                    ordered_events.push_back((event.layer, event))
+                    ordered_events.push_back((event.header.layer, event))
                 }
             }
         }
@@ -810,9 +982,9 @@ impl EventGraph {
     #[cfg(feature = "rpc")]
     pub async fn eventgraph_info(&self, id: u16, _params: JsonValue) -> JsonResult {
         let mut graph = HashMap::new();
-        for iter_elem in self.dag.iter() {
+        for iter_elem in self.main_dag.iter() {
             let (id, val) = iter_elem.unwrap();
-            let id = blake3::Hash::from_bytes((&id as &[u8]).try_into().unwrap());
+            let id = Hash::from_bytes((&id as &[u8]).try_into().unwrap());
             let val: Event = deserialize_async(&val).await.unwrap();
             graph.insert(id, val);
         }
@@ -836,7 +1008,7 @@ impl EventGraph {
     /// provided ones.
     pub async fn fetch_successors_of(
         &self,
-        tips: BTreeMap<u64, HashSet<blake3::Hash>>,
+        tips: BTreeMap<u64, HashSet<Hash>>,
     ) -> Result<Vec<Event>> {
         debug!(
              target: "event_graph::fetch_successors_of",
@@ -844,9 +1016,9 @@ impl EventGraph {
         );
 
         let mut graph = HashMap::new();
-        for iter_elem in self.dag.iter() {
+        for iter_elem in self.main_dag.iter() {
             let (id, val) = iter_elem.unwrap();
-            let hash = blake3::Hash::from_bytes((&id as &[u8]).try_into().unwrap());
+            let hash = Hash::from_bytes((&id as &[u8]).try_into().unwrap());
             let event: Event = deserialize_async(&val).await.unwrap();
             graph.insert(hash, event);
         }
@@ -861,14 +1033,186 @@ impl EventGraph {
             }
 
             for (_, ev) in graph.iter() {
-                if ev.layer > *tip.0 && !result.contains(ev) {
+                if ev.header.layer > *tip.0 && !result.contains(ev) {
                     result.push(ev.clone())
                 }
             }
         }
 
-        result.sort_by(|a, b| a.layer.cmp(&b.layer));
+        result.sort_by(|a, b| a.header.layer.cmp(&b.header.layer));
 
         Ok(result)
     }
 }
+
+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);
+    let ev_rep_sub = match peer.subscribe_msg::<EventRep>().await {
+        Ok(v) => v,
+        Err(e) => {
+            error!(target: "event_graph::dag_sync()","[EVENTGRAPH] Sync: Couldn't subscribe EventRep for peer {}, skipping ({})",url, e);
+            return Err(Error::Custom("Couldn't subscribe EventRep".to_string()))
+        }
+    };
+
+    if let Err(e) = peer.send(&EventReq(vec![missing.id()])).await {
+        error!(target: "event_graph::dag_sync()","[EVENTGRAPH] Sync: Failed communicating EventReq({:?}) to {}: {}",missing, url, e);
+        return Err(Error::Custom("Failed communicating EventReq".to_string()))
+    }
+
+    let Ok(parent) = ev_rep_sub.receive_with_timeout(15).await else {
+        error!(
+            target: "event_graph::dag_sync()",
+            "[EVENTGRAPH] Sync: Timeout waiting for parents {:?} from {}",
+            missing, url,
+        );
+        return Err(().into())
+    };
+
+    Ok(parent.0.clone())
+}
+
+async fn request_header(peer: &Channel, comms_timeout: u64) -> Result<Vec<Header>> {
+    let url = peer.address();
+
+    let hdr_rep_sub = match peer.subscribe_msg::<HeaderRep>().await {
+        Ok(v) => v,
+        Err(e) => {
+            error!(
+                target: "event_graph::dag_sync()",
+                "[EVENTGRAPH] Sync: Couldn't subscribe HeaderReq for peer {}, skipping ({})",
+                url, e,
+            );
+            return Err(Error::EventNotFound("Couldn't subscribe HeaderReq".to_owned()));
+        }
+    };
+
+    // 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()",
+            "[EVENTGRAPH] Sync: Couldn't contact peer {}, skipping ({})", url, e,
+        );
+        return Err(Error::EventNotFound("Couldn't contact peer".to_owned()));
+    };
+
+    // Node waits for response
+    let Ok(peer_headers) = hdr_rep_sub.receive_with_timeout(comms_timeout).await else {
+        error!(
+            target: "event_graph::dag_sync()",
+            "[EVENTGRAPH] Sync: Peer {} didn't reply with headers in time, skipping", url,
+        );
+        // communicated_peers -= 1;
+        return Err(Error::EventNotFound("Peer didn't reply with headers in time".to_owned()));
+    };
+
+    let peer_headers = &peer_headers.0;
+    Ok(peer_headers.to_vec())
+}
+
+async fn request_event(
+    peer: Arc<Channel>,
+    headers: Vec<Hash>,
+    comms_timeout: u64,
+) -> Result<Vec<Event>> {
+    let url = peer.address();
+
+    debug!(
+        target: "event_graph::dag_sync()",
+        "Requesting {:?} from {}...", headers, url,
+    );
+
+    let ev_rep_sub = match peer.subscribe_msg::<EventRep>().await {
+        Ok(v) => v,
+        Err(e) => {
+            error!(
+                target: "event_graph::dag_sync()",
+                "[EVENTGRAPH] Sync: Couldn't subscribe EventRep for peer {}, skipping ({})",
+                url, e,
+            );
+            return Err(Error::EventNotFound("Couldn't subscribe EventRep".to_owned()));
+        }
+    };
+
+    // let request_missing_events = missing_parents.clone().into_iter().collect();
+    if let Err(e) = peer.send(&EventReq(headers.clone())).await {
+        error!(
+            target: "event_graph::dag_sync()",
+            "[EVENTGRAPH] Sync: Failed communicating EventReq({:?}) to {}: {}",
+            headers, url, e,
+        );
+        return Err(Error::EventNotFound("Failed communicating EventReq".to_owned()));
+    }
+
+    // Node waits for response
+    let Ok(event) = ev_rep_sub.receive_with_timeout(comms_timeout).await else {
+        error!(
+            target: "event_graph::dag_sync()",
+            "[EVENTGRAPH] Sync: Timeout waiting for parents {:?} from {}",
+            headers, url,
+        );
+        return Err(Error::EventNotFound("Timeout waiting for parents".to_owned()));
+    };
+
+    Ok(event.0.clone())
+}
+
+fn peer_selection(peers: Vec<Arc<Channel>>) -> Arc<Channel> {
+    peers.choose(&mut OsRng).unwrap().clone()
+}
+
+// async fn send_request(peer: &Channel, missing: &[Hash]) -> Result<Vec<Event>> {
+//     info!("in send_request first missing: {}", missing[0]);
+//     let url = peer.address();
+//     debug!(target: "event_graph::dag_sync()","Requesting {:?} from {}...", missing, url);
+//     let ev_rep_sub = match peer.subscribe_msg::<EventRep>().await {
+//         Ok(v) => v,
+//         Err(e) => {
+//             error!(target: "event_graph::dag_sync()","[EVENTGRAPH] Sync: Couldn't subscribe EventRep for peer {}, skipping ({})",url, e);
+//             return Err(Error::Custom("Couldn't subscribe EventRep".to_string()))
+//         }
+//     };
+
+//     if let Err(e) = peer.send(&EventReq(missing.to_vec())).await {
+//         error!(target: "event_graph::dag_sync()","[EVENTGRAPH] Sync: Failed communicating EventReq({:?}) to {}: {}",missing, url, e);
+//         return Err(Error::Custom("Failed communicating EventReq".to_string()))
+//     }
+
+//     let Ok(parent) = ev_rep_sub.receive_with_timeout(15).await else {
+//         error!(
+//             target: "event_graph::dag_sync()",
+//             "[EVENTGRAPH] Sync: Timeout waiting for parents {:?} from {}",
+//             missing, url,
+//         );
+//         return Err(().into())
+//     };
+
+//     Ok(parent.0.clone())
+// }
+
+// // A function that sends requests to multiple peers concurrently
+// async fn send_requests(peers: &[Arc<Channel>], missing: &[Hash]) -> Result<Vec<Vec<Event>>> {
+//     info!("in send_requests first missing: {}", missing[0]);
+//     let chunk_size = (missing.len() as f64 / peers.len() as f64).ceil() as usize;
+//     let pairs = peers.iter().zip(missing.chunks(chunk_size)).collect::<Vec<_>>();
+
+//     // For each peer, create a future that sends a request
+//     let pair_stream = from_iter(pairs.iter());
+//     let requests_stream = pair_stream.map(|(peer, missing)| send_request(peer, missing));
+
+//     // Collect all the responses into a vector
+//     let responses = requests_stream.collect::<Vec<_>>().await;
+
+//     // Wait for all the futures to complete
+//     future::try_join_all(responses).await
+// }

src/event_graph/proto.rs

@@ -25,11 +25,11 @@ use std::{
     },
 };
 
-use darkfi_serial::{async_trait, SerialDecodable, SerialEncodable};
+use darkfi_serial::{async_trait, deserialize_async, SerialDecodable, SerialEncodable};
 use smol::Executor;
 use tracing::{debug, error, trace, warn};
 
-use super::{Event, EventGraphPtr, NULL_ID};
+use super::{event::Header, Event, EventGraphPtr, NULL_ID};
 use crate::{
     impl_p2p_message,
     net::{
@@ -109,6 +109,12 @@ pub struct ProtocolEventGraph {
     ev_req_sub: MessageSubscription<EventReq>,
     /// `MessageSubscriber` for `EventRep`
     ev_rep_sub: MessageSubscription<EventRep>,
+    /// `MessageSubscriber` for `HeaderPut`
+    _hdr_put_sub: MessageSubscription<HeaderPut>,
+    /// `MessageSubscriber` for `HeaderReq`
+    hdr_req_sub: MessageSubscription<HeaderReq>,
+    /// `MessageSubscriber` for `HeaderRep`
+    _hdr_rep_sub: MessageSubscription<HeaderRep>,
     /// `MessageSubscriber` for `TipReq`
     tip_req_sub: MessageSubscription<TipReq>,
     /// `MessageSubscriber` for `TipRep`
@@ -139,6 +145,21 @@ impl_p2p_message!(EventReq, "EventGraph::EventReq", 0, 0, DEFAULT_METERING_CONFI
 pub struct EventRep(pub Vec<Event>);
 impl_p2p_message!(EventRep, "EventGraph::EventRep", 0, 0, DEFAULT_METERING_CONFIGURATION);
 
+/// A P2P message representing publishing an event's header on the network
+#[derive(Clone, SerialEncodable, SerialDecodable)]
+pub struct HeaderPut(pub Header);
+impl_p2p_message!(HeaderPut, "EventGraph::HeaderPut", 0, 0, DEFAULT_METERING_CONFIGURATION);
+
+/// A P2P message representing a header request
+#[derive(Clone, SerialEncodable, SerialDecodable)]
+pub struct HeaderReq {}
+impl_p2p_message!(HeaderReq, "EventGraph::HeaderReq", 0, 0, DEFAULT_METERING_CONFIGURATION);
+
+/// A P2P message representing a header reply
+#[derive(Clone, SerialEncodable, SerialDecodable)]
+pub struct HeaderRep(pub Vec<Header>);
+impl_p2p_message!(HeaderRep, "EventGraph::HeaderRep", 0, 0, DEFAULT_METERING_CONFIGURATION);
+
 /// A P2P message representing a request for a peer's DAG tips
 #[derive(Clone, SerialEncodable, SerialDecodable)]
 pub struct TipReq {}
@@ -155,6 +176,9 @@ impl ProtocolBase for ProtocolEventGraph {
         self.jobsman.clone().start(ex.clone());
         self.jobsman.clone().spawn(self.clone().handle_event_put(), ex.clone()).await;
         self.jobsman.clone().spawn(self.clone().handle_event_req(), ex.clone()).await;
+        // self.jobsman.clone().spawn(self.clone().handle_header_put(), ex.clone()).await;
+        // self.jobsman.clone().spawn(self.clone().handle_header_req(), ex.clone()).await;
+        self.jobsman.clone().spawn(self.clone().handle_header_rep(), ex.clone()).await;
         self.jobsman.clone().spawn(self.clone().handle_tip_req(), ex.clone()).await;
         self.jobsman.clone().spawn(self.clone().broadcast_rate_limiter(), ex.clone()).await;
         Ok(())
@@ -171,12 +195,18 @@ impl ProtocolEventGraph {
         msg_subsystem.add_dispatch::<EventPut>().await;
         msg_subsystem.add_dispatch::<EventReq>().await;
         msg_subsystem.add_dispatch::<EventRep>().await;
+        msg_subsystem.add_dispatch::<HeaderPut>().await;
+        msg_subsystem.add_dispatch::<HeaderReq>().await;
+        msg_subsystem.add_dispatch::<HeaderRep>().await;
         msg_subsystem.add_dispatch::<TipReq>().await;
         msg_subsystem.add_dispatch::<TipRep>().await;
 
         let ev_put_sub = channel.subscribe_msg::<EventPut>().await?;
         let ev_req_sub = channel.subscribe_msg::<EventReq>().await?;
         let ev_rep_sub = channel.subscribe_msg::<EventRep>().await?;
+        let _hdr_put_sub = channel.subscribe_msg::<HeaderPut>().await?;
+        let hdr_req_sub = channel.subscribe_msg::<HeaderReq>().await?;
+        let _hdr_rep_sub = channel.subscribe_msg::<HeaderRep>().await?;
         let tip_req_sub = channel.subscribe_msg::<TipReq>().await?;
         let _tip_rep_sub = channel.subscribe_msg::<TipRep>().await?;
 
@@ -188,6 +218,9 @@ impl ProtocolEventGraph {
             ev_put_sub,
             ev_req_sub,
             ev_rep_sub,
+            _hdr_put_sub,
+            hdr_req_sub,
+            _hdr_rep_sub,
             tip_req_sub,
             _tip_rep_sub,
             malicious_count: AtomicUsize::new(0),
@@ -231,7 +264,7 @@ impl ProtocolEventGraph {
             };
             trace!(
                  target: "event_graph::protocol::handle_event_put",
-                 "Got EventPut: {} [{}]", event.id(), self.channel.display_address(),
+                 "Got EventPut: {} [{}]", event.header.id(), self.channel.display_address(),
             );
 
             // Check if node has finished syncing its DAG
@@ -244,8 +277,8 @@ impl ProtocolEventGraph {
             }
 
             // If we have already seen the event, we'll stay quiet.
-            let event_id = event.id();
-            if self.event_graph.dag.contains_key(event_id.as_bytes()).unwrap() {
+            let event_id = event.header.id();
+            if self.event_graph.main_dag.contains_key(event_id.as_bytes()).unwrap() {
                 debug!(
                     target: "event_graph::protocol::handle_event_put",
                     "Event {event_id} is already known"
@@ -274,12 +307,12 @@ impl ProtocolEventGraph {
             // The genesis event marks the last time the Dag has been pruned of old
             // events. The pruning interval is defined by the days_rotation field
             // of [`EventGraph`].
-            let genesis_timestamp = self.event_graph.current_genesis.read().await.timestamp;
-            if event.timestamp < genesis_timestamp {
+            let genesis_timestamp = self.event_graph.current_genesis.read().await.header.timestamp;
+            if event.header.timestamp < genesis_timestamp {
                 debug!(
                     target: "event_graph::protocol::handle_event_put",
                     "Event {} is older than genesis. Event timestamp: `{}`. Genesis timestamp: `{genesis_timestamp}`",
-                event.id(), event.timestamp
+                event.header.id(), event.header.timestamp
                 );
             }
 
@@ -299,14 +332,14 @@ impl ProtocolEventGraph {
             );
 
             let mut missing_parents = HashSet::new();
-            for parent_id in event.parents.iter() {
+            for parent_id in event.header.parents.iter() {
                 // `event.validate_new()` should have already made sure that
                 // not all parents are NULL, and that there are no duplicates.
                 if parent_id == &NULL_ID {
                     continue
                 }
 
-                if !self.event_graph.dag.contains_key(parent_id.as_bytes()).unwrap() {
+                if !self.event_graph.main_dag.contains_key(parent_id.as_bytes()).unwrap() {
                     missing_parents.insert(*parent_id);
                 }
             }
@@ -362,12 +395,12 @@ impl ProtocolEventGraph {
                     let parents = parents.0.clone();
 
                     for parent in parents {
-                        let parent_id = parent.id();
+                        let parent_id = parent.header.id();
                         if !missing_parents.contains(&parent_id) {
                             error!(
                                 target: "event_graph::protocol::handle_event_put",
                                 "[EVENTGRAPH] Peer {} replied with a wrong event: {}",
-                                self.channel.display_address(), parent.id(),
+                                self.channel.display_address(), parent.header.id(),
                             );
                             self.channel.stop().await;
                             return Err(Error::ChannelStopped)
@@ -375,21 +408,21 @@ impl ProtocolEventGraph {
 
                         debug!(
                             target: "event_graph::protocol::handle_event_put",
-                            "Got correct parent event {}", parent.id(),
+                            "Got correct parent event {}", parent.header.id(),
                         );
 
-                        if let Some(layer_events) = received_events.get_mut(&parent.layer) {
+                        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.layer, layer_events);
+                            received_events.insert(parent.header.layer, layer_events);
                         }
                         received_events_hashes.insert(parent_id);
 
                         missing_parents.remove(&parent_id);
 
                         // See if we have the upper parents
-                        for upper_parent in parent.parents.iter() {
+                        for upper_parent in parent.header.parents.iter() {
                             if upper_parent == &NULL_ID {
                                 continue
                             }
@@ -398,7 +431,7 @@ impl ProtocolEventGraph {
                                 !received_events_hashes.contains(upper_parent) &&
                                 !self
                                     .event_graph
-                                    .dag
+                                    .main_dag
                                     .contains_key(upper_parent.as_bytes())
                                     .unwrap()
                             {
@@ -509,22 +542,22 @@ impl ProtocolEventGraph {
             // Check if the incoming event is older than the genesis event. If so, something
             // has gone wrong. The event should have been pruned during the last
             // rotation.
-            let genesis_timestamp = self.event_graph.current_genesis.read().await.timestamp;
+            let genesis_timestamp = self.event_graph.current_genesis.read().await.header.timestamp;
             let mut bcast_ids = self.event_graph.broadcasted_ids.write().await;
 
             for event in events.iter() {
-                if event.timestamp < genesis_timestamp {
+                if event.header.timestamp < genesis_timestamp {
                     error!(
                         target: "event_graph::protocol::handle_event_req",
                         "Requested event by peer {} is older than previous rotation period. It should have been pruned.
                     Event timestamp: `{}`. Genesis timestamp: `{genesis_timestamp}`",
-                    event.id(), event.timestamp
+                    event.header.id(), event.header.timestamp
                     );
                 }
 
                 // Now let's get the upper level of event IDs. When we reply, we could
                 // get requests for those IDs as well.
-                for parent_id in event.parents.iter() {
+                for parent_id in event.header.parents.iter() {
                     if parent_id != &NULL_ID {
                         bcast_ids.insert(*parent_id);
                     }
@@ -540,6 +573,51 @@ impl ProtocolEventGraph {
         }
     }
 
+    /// Protocol function handling `HeaderReq`.
+    /// This is triggered whenever someone requests syncing headers by
+    /// sending their current headers.
+    async fn handle_header_rep(self: Arc<Self>) -> Result<()> {
+        loop {
+            self.hdr_req_sub.receive().await?;
+            trace!(
+                target: "event_graph::protocol::handle_tip_req",
+                "Got TipReq [{}]", self.channel.display_address(),
+            );
+
+            // Check if node has finished syncing its DAG
+            if !*self.event_graph.synced.read().await {
+                debug!(
+                    target: "event_graph::protocol::handle_tip_req",
+                    "DAG is still syncing, skipping..."
+                );
+                continue
+            }
+
+            // TODO: Rate limit
+
+            // We received header request. Let's find them, add them to
+            // our bcast ids list, and reply with them.
+            let mut headers = vec![];
+            for item in self.event_graph.main_dag.iter() {
+                let (_, event) = item.unwrap();
+                let event: Event = deserialize_async(&event).await.unwrap();
+                if !headers.contains(&event.header) || event.header.layer != 0 {
+                    headers.push(event.header);
+                }
+            }
+            // let mut bcast_ids = self.event_graph.broadcasted_ids.write().await;
+            // for (_, tips) in layers.iter() {
+            //     for tip in tips {
+            //         bcast_ids.insert(*tip);
+            //     }
+            // }
+            // drop(bcast_ids);
+
+            self.channel.send(&HeaderRep(headers)).await?;
+        }
+        // Ok(())
+    }
+
     /// Protocol function handling `TipReq`.
     /// This is triggered when someone requests the current unreferenced
     /// tips of our DAG.

src/event_graph/tests.rs

@@ -168,9 +168,9 @@ async fn assert_dags(eg_instances: &[Arc<EventGraph>], expected_len: usize, rng:
         let node_last_layer_tips =
             eg.unreferenced_tips.read().await.last_key_value().unwrap().1.clone();
         assert!(
-            eg.dag.len() == expected_len,
+            eg.main_dag.len() == expected_len,
             "Node {i}, expected {expected_len} events, have {}",
-            eg.dag.len()
+            eg.main_dag.len()
         );
         assert_eq!(
             node_last_layer_tips, last_layer_tips,
@@ -213,12 +213,13 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
 
     // Grab genesis event
     let random_node = eg_instances.choose(&mut rng).unwrap();
-    let (id, _) = random_node.dag.last().unwrap().unwrap();
+    let (id, _) = random_node.main_dag.last().unwrap().unwrap();
     let genesis_event_id = blake3::Hash::from_bytes((&id as &[u8]).try_into().unwrap());
 
     // =========================================
     // 1. Assert that everyone's DAG is the same
     // =========================================
+    info!("dag len is 1");
     assert_dags(&eg_instances, 1, &mut rng).await;
 
     // ==========================================
@@ -226,7 +227,7 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     // ==========================================
     let random_node = eg_instances.choose(&mut rng).unwrap();
     let event = Event::new(vec![1, 2, 3, 4], random_node).await;
-    assert!(event.parents.contains(&genesis_event_id));
+    assert!(event.header.parents.contains(&genesis_event_id));
     // The node adds it to their DAG, on layer 1.
     let event_id = random_node.dag_insert(slice::from_ref(&event)).await.unwrap()[0];
     let tips_layers = random_node.unreferenced_tips.read().await;
@@ -242,6 +243,7 @@ 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;
 
     // ==============================================================
@@ -257,14 +259,17 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     let event2 = Event::new(vec![1, 2, 3, 4, 2], random_node).await;
     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.dag.len(), 5);
+    assert_eq!(random_node.main_dag.len(), 5);
     let tips_layers = random_node.unreferenced_tips.read().await;
     assert_eq!(tips_layers.len(), 1);
     assert!(tips_layers.get(&4).unwrap().get(&event2_id).is_some());
     drop(tips_layers);
 
-    let event_chain =
-        vec![(event0_id, event0.parents), (event1_id, event1.parents), (event2_id, event2.parents)];
+    let event_chain = vec![
+        (event0_id, event0.header.parents),
+        (event1_id, event1.header.parents),
+        (event2_id, event2.header.parents),
+    ];
 
     info!("Broadcasting event {event2_id}");
     info!("Event chain: {event_chain:#?}");
@@ -275,6 +280,7 @@ 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;
 
     // ===========================================
@@ -286,14 +292,17 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     let event0_1 = Event::new(vec![1, 2, 3, 4, 3], node1).await;
     node1.dag_insert(slice::from_ref(&event0_1)).await.unwrap();
     node1.p2p.broadcast(&EventPut(event0_1)).await;
+    sleep(1).await;
 
     let event1_1 = Event::new(vec![1, 2, 3, 4, 4], node1).await;
     node1.dag_insert(slice::from_ref(&event1_1)).await.unwrap();
     node1.p2p.broadcast(&EventPut(event1_1)).await;
+    sleep(1).await;
 
     let event2_1 = Event::new(vec![1, 2, 3, 4, 5], node1).await;
     node1.dag_insert(slice::from_ref(&event2_1)).await.unwrap();
     node1.p2p.broadcast(&EventPut(event2_1)).await;
+    sleep(1).await;
 
     // =======
     // node 2
@@ -302,14 +311,17 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     let event0_2 = Event::new(vec![1, 2, 3, 4, 6], node2).await;
     node2.dag_insert(slice::from_ref(&event0_2)).await.unwrap();
     node2.p2p.broadcast(&EventPut(event0_2)).await;
+    sleep(1).await;
 
     let event1_2 = Event::new(vec![1, 2, 3, 4, 7], node2).await;
     node2.dag_insert(slice::from_ref(&event1_2)).await.unwrap();
     node2.p2p.broadcast(&EventPut(event1_2)).await;
+    sleep(1).await;
 
     let event2_2 = Event::new(vec![1, 2, 3, 4, 8], node2).await;
     node2.dag_insert(slice::from_ref(&event2_2)).await.unwrap();
     node2.p2p.broadcast(&EventPut(event2_2)).await;
+    sleep(1).await;
 
     // =======
     // node 3
@@ -317,15 +329,36 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     let node3 = eg_instances.choose(&mut rng).unwrap();
     let event0_3 = Event::new(vec![1, 2, 3, 4, 9], node3).await;
     node3.dag_insert(slice::from_ref(&event0_3)).await.unwrap();
-    node2.p2p.broadcast(&EventPut(event0_3)).await;
+    node3.p2p.broadcast(&EventPut(event0_3)).await;
+    sleep(1).await;
 
     let event1_3 = Event::new(vec![1, 2, 3, 4, 10], node3).await;
     node3.dag_insert(slice::from_ref(&event1_3)).await.unwrap();
-    node2.p2p.broadcast(&EventPut(event1_3)).await;
+    node3.p2p.broadcast(&EventPut(event1_3)).await;
+    sleep(1).await;
 
     let event2_3 = Event::new(vec![1, 2, 3, 4, 11], node3).await;
     node3.dag_insert(slice::from_ref(&event2_3)).await.unwrap();
     node3.p2p.broadcast(&EventPut(event2_3)).await;
+    sleep(1).await;
+
+    // /////
+    // //
+    // let node4 = eg_instances.choose(&mut rng).unwrap();
+    // let event0_4 = Event::new(vec![1, 2, 3, 4, 12], node4).await;
+    // node4.dag_insert(&[event0_4.clone()]).await.unwrap();
+    // node4.p2p.broadcast(&EventPut(event0_4)).await;
+    // sleep(1).await;
+
+    // let event1_4 = Event::new(vec![1, 2, 3, 4, 13], node4).await;
+    // node4.dag_insert(&[event1_4.clone()]).await.unwrap();
+    // node4.p2p.broadcast(&EventPut(event1_4)).await;
+    // sleep(1).await;
+
+    // let event2_4 = Event::new(vec![1, 2, 3, 4, 14], node4).await;
+    // node4.dag_insert(&[event2_4.clone()]).await.unwrap();
+    // node4.p2p.broadcast(&EventPut(event2_4)).await;
+    // // sleep(1).await;
 
     info!("Waiting 5s for events propagation");
     sleep(5).await;
@@ -334,6 +367,7 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
     // 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;
 
     // ============================================================
@@ -364,13 +398,14 @@ async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
         info!("Waiting 5s for new node connection");
         sleep(5).await;
 
-        event_graph.dag_sync().await.unwrap()
+        event_graph.dag_sync(false).await.unwrap()
     }
 
     // ============================================================
     // 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
@@ -442,7 +477,7 @@ async fn eventgraph_chaotic_propagation_real(ex: Arc<Executor<'static>>) {
         info!("Waiting 5s for new node connection");
         sleep(5).await;
 
-        event_graph.dag_sync().await.unwrap()
+        event_graph.dag_sync(false).await.unwrap()
     }
 
     // ============================================================

src/event_graph/util.rs

@@ -45,6 +45,8 @@ use {
     tracing::error,
 };
 
+use super::event::Header;
+
 /// MilliSeconds in a day
 pub(super) const DAY: i64 = 86_400_000;
 
@@ -131,12 +133,8 @@ pub fn generate_genesis(days_rotation: u64) -> Event {
         // Calculate the timestamp of the most recent event
         INITIAL_GENESIS + (rotations_since_genesis * days_rotation * DAY as u64)
     };
-    Event {
-        timestamp,
-        content: GENESIS_CONTENTS.to_vec(),
-        parents: [NULL_ID; N_EVENT_PARENTS],
-        layer: 0,
-    }
+    let header = Header { timestamp, parents: [NULL_ID; N_EVENT_PARENTS], layer: 0 };
+    Event { header, content: GENESIS_CONTENTS.to_vec() }
 }
 
 pub(super) fn replayer_log(datastore: &Path, cmd: String, value: Vec<u8>) -> Result<()> {
@@ -179,7 +177,7 @@ pub async fn recreate_from_replayer_log(datastore: &Path) -> JsonResult {
             let v = base64::decode(line[1]).unwrap();
             let v: Event = deserialize(&v).unwrap();
             let v_se = serialize(&v);
-            dag.insert(v.id().as_bytes(), v_se).unwrap();
+            dag.insert(v.header.id().as_bytes(), v_se).unwrap();
         }
     }
 

src/rpc/from_impl.rs

@@ -174,12 +174,12 @@ impl From<net::dnet::DnetEvent> for JsonValue {
 impl From<event_graph::Event> for JsonValue {
     fn from(event: event_graph::Event) -> JsonValue {
         let parents =
-            event.parents.into_iter().map(|id| JsonStr(id.to_string())).collect::<Vec<_>>();
+            event.header.parents.into_iter().map(|id| JsonStr(id.to_string())).collect::<Vec<_>>();
         json_map([
-            ("timestamp", JsonNum(event.timestamp as f64)),
+            ("timestamp", JsonNum(event.header.timestamp as f64)),
             ("content", JsonStr(bs58::encode(event.content()).into_string())),
             ("parents", JsonArray(parents)),
-            ("layer", JsonNum(event.layer as f64)),
+            ("layer", JsonNum(event.header.layer as f64)),
         ])
     }
 }