event-graph: Alternative DAG WIP

Cargo.lock

@@ -425,6 +425,17 @@ dependencies = [
  "windows-sys 0.48.0",
 ]
 
+[[package]]
+name = "async-recursion"
+version = "1.0.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5fd55a5ba1179988837d24ab4c7cc8ed6efdeff578ede0416b4225a5fca35bd0"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.29",
+]
+
 [[package]]
 name = "async-rustls"
 version = "0.4.0"
@@ -1399,6 +1410,7 @@ name = "darkfi"
 version = "0.4.1"
 dependencies = [
  "arti-client",
+ "async-recursion",
  "async-rustls",
  "async-trait",
  "blake3",

Cargo.toml

@@ -57,6 +57,7 @@ log = "0.4.20"
 thiserror = "1.0.47"
 
 # async-runtime
+async-recursion = {version = "1.0.5", optional = true}
 async-trait = {version = "0.1.73", optional = true}
 futures = {version = "0.3.28", optional = true}
 smol = {version = "1.3.0", optional = true}
@@ -188,8 +189,10 @@ geode = [
 
 event-graph = [
     "async-trait",
+    "async-recursion",
     "blake3",
     "rand",
+    "sled",
     "smol",
     "tinyjson",
 

src/event_graph2/mod.rs

@@ -0,0 +1,233 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::{
+    collections::{HashSet, VecDeque},
+    sync::Arc,
+};
+
+use async_recursion::async_recursion;
+use darkfi_serial::{
+    async_trait, deserialize_async, serialize_async, Encodable, SerialDecodable, SerialEncodable,
+};
+use smol::lock::RwLock;
+
+use crate::{net::P2pPtr, util::time::Timestamp, Result};
+
+/// P2P protocol implementation for the Event Graph
+pub mod proto;
+
+#[cfg(test)]
+mod tests;
+
+/// The number of parents an event is supposed to have.
+const N_EVENT_PARENTS: usize = 5;
+/// Allowed timestamp drift in seconds
+const EVENT_TIME_DRIFT: u64 = 60;
+// Allowed orphan age limit in seconds
+//const ORPHAN_AGE_LIMIT: u64 = 60 * 5;
+/// Null event ID
+const NULL_ID: blake3::Hash = blake3::Hash::from_bytes([0x00; blake3::OUT_LEN]);
+
+/// Atomic pointer to an [`EventGraph`] instance.
+pub type EventGraphPtr = Arc<EventGraph>;
+
+/// An Event Graph instance
+pub struct EventGraph {
+    /// Pointer to the P2P network instance
+    p2p: P2pPtr,
+    /// Sled tree containing the DAG
+    dag: sled::Tree,
+    /// The set of unreferenced DAG tips
+    unreferenced_tips: RwLock<HashSet<blake3::Hash>>,
+    /// The last event ID inserted into the DAG
+    last_event: RwLock<blake3::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.
+    broadcasted_ids: RwLock<HashSet<blake3::Hash>>,
+}
+
+impl EventGraph {
+    /// Create a new [`EventGraph`] instance
+    pub fn new(p2p: P2pPtr, sled_db: &sled::Db, dag_tree_name: &str) -> Result<EventGraphPtr> {
+        let dag = sled_db.open_tree(dag_tree_name)?;
+        let unreferenced_tips = RwLock::new(HashSet::new());
+        let last_event = RwLock::new(NULL_ID);
+        let broadcasted_ids = RwLock::new(HashSet::new());
+
+        Ok(Arc::new(Self { p2p, dag, unreferenced_tips, last_event, broadcasted_ids }))
+    }
+
+    /// Insert an event into the DAG
+    pub async fn dag_insert(&self, event: &Event) -> Result<blake3::Hash> {
+        let event_id = event.id();
+        let s_event = serialize_async(event).await;
+
+        let mut unreferenced_tips = self.unreferenced_tips.write().await;
+        let mut bcast_ids = self.broadcasted_ids.write().await;
+        for parent_id in event.parents.iter() {
+            if parent_id != &NULL_ID {
+                unreferenced_tips.remove(parent_id);
+                bcast_ids.insert(*parent_id);
+            }
+        }
+        unreferenced_tips.insert(event_id);
+        drop(unreferenced_tips);
+        drop(bcast_ids);
+
+        self.dag.insert(event_id.as_bytes(), s_event).unwrap();
+        *self.last_event.write().await = event_id;
+
+        Ok(event_id)
+    }
+
+    /// Get a set of unreferenced tips used to produce a new [`Event`]
+    async fn get_unreferenced_tips(&self) -> [blake3::Hash; N_EVENT_PARENTS] {
+        let mut tips = [NULL_ID; N_EVENT_PARENTS];
+        let unreferenced_tips = self.unreferenced_tips.read().await;
+
+        for (i, tip) in unreferenced_tips.iter().enumerate() {
+            if i == N_EVENT_PARENTS - 1 {
+                break
+            }
+
+            tips[i] = *tip;
+        }
+
+        assert!(tips.iter().any(|x| x != &NULL_ID));
+        tips
+    }
+
+    /// Perform a topological sort of the DAG.
+    pub async fn order_events(&self) -> Vec<blake3::Hash> {
+        let mut ordered_events = VecDeque::new();
+        let mut visited = HashSet::new();
+
+        for tip in self.get_unreferenced_tips().await {
+            if !visited.contains(&tip) && tip != NULL_ID {
+                let tip = self.dag.get(tip.as_bytes()).unwrap().unwrap();
+                let tip = deserialize_async(&tip).await.unwrap();
+                self.dfs_topological_sort(tip, &mut visited, &mut ordered_events).await;
+            }
+        }
+
+        ordered_events.make_contiguous().to_vec()
+    }
+
+    /// <https://en.wikipedia.org/wiki/Depth-first_search>
+    #[async_recursion]
+    async fn dfs_topological_sort(
+        &self,
+        event: Event,
+        visited: &mut HashSet<blake3::Hash>,
+        ordered_events: &mut VecDeque<blake3::Hash>,
+    ) {
+        let event_id = event.id();
+        visited.insert(event_id);
+
+        for parent_id in event.parents.iter() {
+            if !visited.contains(parent_id) && parent_id != &NULL_ID {
+                let p_event = self.dag.get(parent_id.as_bytes()).unwrap().unwrap();
+                let p_event = deserialize_async(&p_event).await.unwrap();
+                self.dfs_topological_sort(p_event, visited, ordered_events).await;
+            }
+        }
+
+        // Once all the parents are visited, add the current event
+        // to the start of the list
+        ordered_events.push_front(event_id);
+    }
+}
+
+/// Representation of an event in the Event Graph
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
+pub struct Event {
+    /// Timestamp of the event
+    timestamp: Timestamp,
+    /// Content of the event
+    content: Vec<u8>,
+    /// Parent nodes in the event DAG
+    parents: [blake3::Hash; N_EVENT_PARENTS],
+}
+
+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.
+    pub async fn new(data: Vec<u8>, event_graph: EventGraphPtr) -> Self {
+        Self {
+            timestamp: Timestamp::current_time(),
+            content: data,
+            parents: event_graph.get_unreferenced_tips().await,
+        }
+    }
+
+    /// Hash the [`Event`] 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();
+        hasher.finalize()
+    }
+
+    /*
+    /// Check if an [`Event`] is considered too old.
+    fn is_too_old(&self) -> bool {
+        self.timestamp.0 < Timestamp::current_time().0 - ORPHAN_AGE_LIMIT
+    }
+    */
+
+    /// Validate a new event for the correct layout.
+    pub fn validate(&self) -> bool {
+        // Let's not bother with empty events
+        if self.content.is_empty() {
+            return false
+        }
+
+        // Check if the event is too old or too new
+        let now = Timestamp::current_time().0;
+        let too_old = self.timestamp.0 < now - EVENT_TIME_DRIFT;
+        let too_new = self.timestamp.0 > now + EVENT_TIME_DRIFT;
+
+        if too_old || too_new {
+            return false
+        }
+
+        // Check there is at least one valid parent.
+        // TODO: It's possible multiple parents are the same and not NULL.
+        //       Should we consider this invalid?
+        let mut has_valid_parent = false;
+        let self_id = self.id();
+        for parent_id in self.parents.iter() {
+            // If it's recursing to us, obviously it's malicious
+            if parent_id == &self_id {
+                return false
+            }
+
+            // Check that at least one parent is not NULL
+            if parent_id != &NULL_ID {
+                has_valid_parent = true;
+            }
+        }
+
+        has_valid_parent
+    }
+}

src/event_graph2/proto.rs

@@ -0,0 +1,298 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::{
+    collections::{HashMap, HashSet},
+    sync::{
+        atomic::{AtomicUsize, Ordering::SeqCst},
+        Arc,
+    },
+    time::Duration,
+};
+
+use darkfi_serial::{async_trait, deserialize_async, SerialDecodable, SerialEncodable};
+use log::{debug, error};
+use smol::Executor;
+
+use super::{Event, EventGraphPtr, NULL_ID};
+use crate::{impl_p2p_message, net::*, system::timeout::timeout, Error, Result};
+
+/// Malicious behaviour threshold. If the threshold is reached, we will
+/// drop the peer from our P2P connection.
+const MALICIOUS_THRESHOLD: usize = 5;
+/// Time to wait for a parent ID reply
+const REPLY_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// P2P protocol implementation for the Event Graph.
+pub struct ProtocolEventGraph {
+    /// Pointer to the connected peer
+    channel: ChannelPtr,
+    /// Pointer to the Event Graph instance
+    event_graph: EventGraphPtr,
+    /// `MessageSubscriber` for `EventPut`
+    ev_put_sub: MessageSubscription<EventPut>,
+    /// `MessageSubscriber` for `EventReq`
+    ev_req_sub: MessageSubscription<EventReq>,
+    /// `MessageSubscriber` for `EventRep`
+    ev_rep_sub: MessageSubscription<EventRep>,
+    /// Peer malicious message count
+    malicious_count: AtomicUsize,
+    /// P2P jobs manager pointer
+    jobsman: ProtocolJobsManagerPtr,
+}
+
+/// A P2P message representing publishing an event on the network
+#[derive(Clone, SerialEncodable, SerialDecodable)]
+pub struct EventPut(pub Event);
+impl_p2p_message!(EventPut, "EventGraph::EventPut");
+
+/// A P2P message representing an event request
+#[derive(Clone, SerialEncodable, SerialDecodable)]
+pub struct EventReq(pub blake3::Hash);
+impl_p2p_message!(EventReq, "EventGraph::EventReq");
+
+/// A P2P message representing an event reply
+#[derive(Clone, SerialEncodable, SerialDecodable)]
+pub struct EventRep(pub Event);
+impl_p2p_message!(EventRep, "EventGraph::EventRep");
+
+#[async_trait]
+impl ProtocolBase for ProtocolEventGraph {
+    async fn start(self: Arc<Self>, ex: Arc<Executor<'_>>) -> Result<()> {
+        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;
+        Ok(())
+    }
+
+    fn name(&self) -> &'static str {
+        "ProtocolEventGraph"
+    }
+}
+
+impl ProtocolEventGraph {
+    pub async fn init(event_graph: EventGraphPtr, channel: ChannelPtr) -> Result<ProtocolBasePtr> {
+        let msg_subsystem = channel.message_subsystem();
+        msg_subsystem.add_dispatch::<EventPut>().await;
+        msg_subsystem.add_dispatch::<EventReq>().await;
+        msg_subsystem.add_dispatch::<EventRep>().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?;
+
+        Ok(Arc::new(Self {
+            channel: channel.clone(),
+            event_graph,
+            ev_put_sub,
+            ev_req_sub,
+            ev_rep_sub,
+            malicious_count: AtomicUsize::new(0),
+            jobsman: ProtocolJobsManager::new("ProtocolEventGraph", channel.clone()),
+        }))
+    }
+
+    async fn handle_event_put(self: Arc<Self>) -> Result<()> {
+        loop {
+            let event = match self.ev_put_sub.receive().await {
+                Ok(v) => v.0.clone(),
+                Err(e) => {
+                    error!(
+                        target: "event_graph::handle_event_put()",
+                        "[EVENTGRAPH] handle_event_put() recv fail: {}", e,
+                    );
+                    continue
+                }
+            };
+
+            // We received an event. Check if we already have it in our DAG.
+            // Also check if we have the event's parents. In the case we do
+            // not have the parents, we'll request them from the peer that has
+            // sent this event to us. In case they do not reply in time, we drop
+            // the event.
+
+            // Validate the event first. If we do not consider it valid, we
+            // will just drop it and stay quiet. If the malicious threshold
+            // is reached, we will stop the connection.
+            if !event.validate() {
+                let malicious_count = self.malicious_count.fetch_add(1, SeqCst);
+                if malicious_count + 1 == MALICIOUS_THRESHOLD {
+                    error!(
+                        target: "event_graph::handle_event_put()",
+                        "[EVENTGRAPH] Peer {} reached malicious threshold. Dropping connection.",
+                        self.channel.address(),
+                    );
+                    self.channel.stop().await;
+                    return Err(Error::ChannelStopped)
+                }
+
+                continue
+            }
+
+            // 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() {
+                debug!(target: "event_graph::handle_event_put()", "Got known event");
+                continue
+            }
+
+            // At this point, this is a new event to us. Let's see if we
+            // have all of its parents.
+            /*
+            info!(
+                target: "event_graph::handle_event_put()",
+                "[EVENTGRAPH] Got new event"
+            );
+            */
+            let mut missing_parents = HashSet::new();
+            for parent_id in event.parents.iter() {
+                // `event.validate()` should have already made sure that
+                // not all parents are NULL.
+                if parent_id == &NULL_ID {
+                    continue
+                }
+
+                if !self.event_graph.dag.contains_key(parent_id.as_bytes()).unwrap() {
+                    missing_parents.insert(*parent_id);
+                }
+            }
+
+            // If we have missing parents, then we have to attempt to
+            // fetch them from this peer.
+            if !missing_parents.is_empty() {
+                debug!(
+                    target: "event_graph::handle_event_put()",
+                    "Event has {} missing parents. Requesting...", missing_parents.len(),
+                );
+                let mut received_events = HashMap::new();
+                for parent_id in missing_parents.iter() {
+                    debug!(
+                        target: "event_graph::handle_event_put()",
+                        "Requesting {}", parent_id,
+                    );
+                    self.channel.send(&EventReq(*parent_id)).await?;
+                    let parent = match timeout(REPLY_TIMEOUT, self.ev_rep_sub.receive()).await {
+                        Ok(parent) => parent?,
+                        Err(_) => {
+                            error!(
+                                target: "event_graph::handle_event_put()",
+                                "[EVENTGRAPH] Timeout while waiting for parent {} from {}",
+                                parent_id, self.channel.address(),
+                            );
+                            self.channel.stop().await;
+                            return Err(Error::ChannelStopped)
+                        }
+                    };
+                    let parent = parent.0.clone();
+
+                    if &parent.id() != parent_id {
+                        error!(
+                            target: "event_graph::handle_event_put()",
+                            "[EVENTGRAPH] Peer {} replied with a wrong event: {}",
+                            self.channel.address(), parent.id(),
+                        );
+                        self.channel.stop().await;
+                        return Err(Error::ChannelStopped)
+                    }
+
+                    debug!(
+                        target: "event_graph::handle_event_put()",
+                        "Got correct parent event {}", parent.id(),
+                    );
+
+                    received_events.insert(parent.id(), parent);
+                }
+                // At this point we should've got all the events.
+                // We should add them to the DAG.
+                // TODO: FIXME: Also validate these events.
+                for event in received_events.values() {
+                    self.event_graph.dag_insert(event).await.unwrap();
+                }
+            } // <-- !missing_parents.is_empty()
+
+            // If we're here, we have all the parents, and we can now
+            // add the actual event to the DAG.
+            self.event_graph.dag_insert(&event).await.unwrap();
+
+            // Relay the event to other peers
+            self.event_graph
+                .p2p
+                .broadcast_with_exclude(&EventPut(event), &[self.channel.address().clone()])
+                .await;
+        }
+    }
+
+    async fn handle_event_req(self: Arc<Self>) -> Result<()> {
+        loop {
+            let event_id = match self.ev_req_sub.receive().await {
+                Ok(v) => v.0,
+                Err(e) => {
+                    error!(
+                        target: "event_graph::handle_event_req()",
+                        "[EVENTGRAPH] handle_event_req() recv fail: {}", e,
+                    );
+                    continue
+                }
+            };
+
+            // We received an event request from somebody.
+            // If we do have ti, we will send it back to them as `EventRep`.
+            // Otherwise, we'll stay quiet. An honest node should always have
+            // something to reply with provided that the request is legitimate,
+            // i.e. we've sent something to them and they did not have some of
+            // the parents.
+
+            // Check if we expected this request to come around.
+            // I dunno if this is a good idea, but it seems it will help
+            // against malicious event requests where they want us to keep
+            // reading our db and steal our bandwidth.
+            if !self.event_graph.broadcasted_ids.read().await.contains(&event_id) {
+                let malicious_count = self.malicious_count.fetch_add(1, SeqCst);
+                if malicious_count + 1 == MALICIOUS_THRESHOLD {
+                    error!(
+                        target: "event_graph::handle_event_req()",
+                        "[EVENTGRAPH] Peer {} reached malicious threshold. Dropping connection.",
+                        self.channel.address(),
+                    );
+                    self.channel.stop().await;
+                    return Err(Error::ChannelStopped)
+                }
+
+                continue
+            }
+
+            // At this point we should have it in our DAG.
+            // This code panics if this is not the case.
+            let event = self.event_graph.dag.get(event_id.as_bytes()).unwrap().unwrap();
+            let event: Event = deserialize_async(&event).await.unwrap();
+
+            // Now let's get the upper level of event IDs. When we reply, we could
+            // get requests for those IDs as well.
+            let mut bcast_ids = self.event_graph.broadcasted_ids.write().await;
+            for parent_id in event.parents.iter() {
+                if parent_id != &NULL_ID {
+                    bcast_ids.insert(event_id);
+                }
+            }
+            drop(bcast_ids);
+
+            // Reply with the event
+            self.channel.send(&EventRep(event)).await?;
+        }
+    }
+}

src/event_graph2/tests.rs

@@ -0,0 +1,214 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::{collections::HashMap, sync::Arc};
+
+use log::info;
+use rand::{prelude::SliceRandom, Rng};
+use smol::{channel, future, Executor};
+use url::Url;
+
+use crate::{
+    event_graph2::{
+        proto::{EventPut, ProtocolEventGraph},
+        Event, EventGraph, NULL_ID, N_EVENT_PARENTS,
+    },
+    net::{P2p, Settings, SESSION_ALL},
+    system::sleep,
+    util::time::Timestamp,
+};
+
+/// Number of nodes to spawn
+const N_NODES: usize = 50;
+//const N_NODES: usize = 2;
+/// Number of peers each node connects to
+const N_CONNS: usize = N_NODES / 3;
+//const N_CONNS: usize = 1;
+
+#[test]
+#[ignore]
+fn eventgraph_propagation() {
+    let mut cfg = simplelog::ConfigBuilder::new();
+    cfg.add_filter_ignore("net::protocol_ping".to_string());
+    cfg.add_filter_ignore("net::channel::subscribe_stop()".to_string());
+    cfg.add_filter_ignore("net::hosts".to_string());
+    cfg.add_filter_ignore("net::message_subscriber".to_string());
+    cfg.add_filter_ignore("net::protocol_address".to_string());
+
+    simplelog::TermLogger::init(
+        //simplelog::LevelFilter::Info,
+        simplelog::LevelFilter::Debug,
+        //simplelog::LevelFilter::Trace,
+        cfg.build(),
+        simplelog::TerminalMode::Mixed,
+        simplelog::ColorChoice::Auto,
+    )
+    .unwrap();
+
+    let ex = Arc::new(Executor::new());
+    let ex_ = ex.clone();
+    let (signal, shutdown) = channel::unbounded::<()>();
+
+    // Run a thread for each node.
+    easy_parallel::Parallel::new()
+        .each(0..N_NODES, |_| future::block_on(ex.run(shutdown.recv())))
+        .finish(|| {
+            future::block_on(async {
+                eventgraph_propagation_real(ex_).await;
+                drop(signal);
+            })
+        });
+}
+
+async fn eventgraph_propagation_real(ex: Arc<Executor<'static>>) {
+    let mut eg_instances = vec![];
+    let mut rng = rand::thread_rng();
+
+    let genesis_event = Event {
+        timestamp: Timestamp::current_time(),
+        content: vec![0xff, 0xba, 0xfe, 0xf1],
+        parents: [NULL_ID; N_EVENT_PARENTS],
+    };
+
+    // Initialize the nodes
+    for i in 0..N_NODES {
+        // Everyone will connect to N_CONNS random peers.
+        let mut peers = vec![];
+        for _ in 0..N_CONNS {
+            let mut port = 13200 + i;
+            while port == 13200 + i {
+                port = 13200 + rng.gen_range(0..N_NODES);
+            }
+            peers.push(Url::parse(&format!("tcp://127.0.0.1:{}", port)).unwrap());
+        }
+
+        let settings = Settings {
+            localnet: true,
+            inbound_addrs: vec![Url::parse(&format!("tcp://127.0.0.1:{}", 13200 + i)).unwrap()],
+            outbound_connections: 0,
+            outbound_connect_timeout: 2,
+            inbound_connections: usize::MAX,
+            peers,
+            allowed_transports: vec!["tcp".to_string()],
+            ..Default::default()
+        };
+
+        let p2p = P2p::new(settings, ex.clone()).await;
+        let sled_db = sled::Config::new().temporary(true).open().unwrap();
+        let event_graph = EventGraph::new(p2p.clone(), &sled_db, "dag").unwrap();
+        let event_graph_ = event_graph.clone();
+
+        // Everyone initializes the event graph with a genesis event.
+        event_graph.dag_insert(&genesis_event).await.unwrap();
+
+        // Register the P2P protocols
+        let registry = p2p.protocol_registry();
+        registry
+            .register(SESSION_ALL, move |channel, _| {
+                let event_graph_ = event_graph_.clone();
+                async move { ProtocolEventGraph::init(event_graph_, channel).await.unwrap() }
+            })
+            .await;
+
+        eg_instances.push(event_graph);
+    }
+
+    // Start the P2P network
+    for eg in eg_instances.iter() {
+        eg.p2p.clone().start().await.unwrap();
+    }
+
+    info!("Waiting 10s until all peers connect");
+    sleep(10).await;
+
+    // Now we get to the logic.
+    //for i in 1..1001_u16 {
+    for i in 1..3_u16 {
+        // A random node creates an event
+        let random_node = eg_instances.choose(&mut rand::thread_rng()).unwrap();
+        let event = Event::new(i.to_le_bytes().to_vec(), random_node.clone()).await;
+
+        // The node adds it to their DAG
+        let event_id = random_node.dag_insert(&event).await.unwrap();
+        // The node broadcasts it
+        info!("Broadcasting {}", event_id);
+        random_node.p2p.broadcast(&EventPut(event)).await;
+
+        // Another random node creates five events and sends them out of order.
+        let random_node = eg_instances.choose(&mut rand::thread_rng()).unwrap();
+
+        let event0 = Event::new(i.to_le_bytes().to_vec(), random_node.clone()).await;
+        let event0_id = random_node.dag_insert(&event0).await.unwrap();
+
+        let event1 = Event::new(i.to_le_bytes().to_vec(), random_node.clone()).await;
+        let event1_id = random_node.dag_insert(&event1).await.unwrap();
+
+        let event2 = Event::new(i.to_le_bytes().to_vec(), random_node.clone()).await;
+        let event2_id = random_node.dag_insert(&event2).await.unwrap();
+
+        let event3 = Event::new(i.to_le_bytes().to_vec(), random_node.clone()).await;
+        let event3_id = random_node.dag_insert(&event3).await.unwrap();
+
+        let event4 = Event::new(i.to_le_bytes().to_vec(), random_node.clone()).await;
+        let event4_id = random_node.dag_insert(&event4).await.unwrap();
+
+        info!("Broadcasting {}", event3_id);
+        random_node.p2p.broadcast(&EventPut(event3)).await;
+        info!("Broadcasting {}", event2_id);
+        random_node.p2p.broadcast(&EventPut(event2)).await;
+        info!("Broadcasting {}", event4_id);
+        random_node.p2p.broadcast(&EventPut(event4)).await;
+        info!("Broadcasting {}", event1_id);
+        random_node.p2p.broadcast(&EventPut(event1)).await;
+        info!("Broadcasting {}", event0_id);
+        random_node.p2p.broadcast(&EventPut(event0)).await;
+    }
+
+    info!("Waiting 10s until the p2p broadcasts settle");
+    sleep(10).await;
+
+    // Assert that everyone has the same DAG
+    let mut contents = HashMap::new();
+    for (i, eg) in eg_instances.iter().enumerate() {
+        let mut ids = vec![];
+        for r in eg.dag.iter() {
+            let (id, _) = r.unwrap();
+            ids.push(blake3::Hash::from_bytes((&id as &[u8]).try_into().unwrap()));
+        }
+
+        contents.insert(i, ids);
+    }
+    let value = contents.values().next().unwrap();
+    assert!(contents.values().all(|v| v == value));
+
+    // Assert that everyone's DAG sorts the same.
+    let mut orders = HashMap::new();
+    for (i, eg) in eg_instances.iter().enumerate() {
+        let order = eg.order_events().await;
+        orders.insert(i, order);
+    }
+    let value = orders.values().next().unwrap();
+    for (i, order) in orders.iter() {
+        assert!(order == value, "{} has wrong order:\n{:#?}\nvs{:#?}", i, order, value);
+    }
+
+    // Stop the P2P network
+    for eg in eg_instances.iter() {
+        eg.p2p.clone().stop().await;
+    }
+}

src/lib.rs

@@ -36,6 +36,9 @@ pub mod geode;
 #[cfg(feature = "event-graph")]
 pub mod event_graph;
 
+#[cfg(feature = "event-graph")]
+pub mod event_graph2;
+
 #[cfg(feature = "net")]
 pub mod net;