script/evgrd: dropped

script/evgrd/.gitignore

@@ -1,2 +0,0 @@
-/target
-Cargo.lock

script/evgrd/Cargo.toml

@@ -1,74 +0,0 @@
-[package]
-name = "evgrd"
-description = "Event graph daemon"
-version = "0.5.1"
-edition = "2021"
-authors = ["Dyne.org foundation <foundation@dyne.org>"]
-license = "AGPL-3.0-only"
-homepage = "https://dark.fi"
-repository = "https://codeberg.org/darkrenaissance/darkfi"
-
-[[bin]]
-name = "evgrd"
-path = "bin/evgrd.rs"
-required-features = ["build-daemon"]
-
-[[example]]
-name = "recv"
-path = "example/recv.rs"
-
-[[example]]
-name = "send"
-path = "example/send.rs"
-
-[dependencies]
-darkfi = {path = "../../", features = ["event-graph"]}
-darkfi-serial = {path = "../../src/serial", features = ["async"]}
-
-# Event Graph DB
-sled-overlay = "0.1.20"
-
-# Crypto
-blake3 = "1.8.5"
-
-# Misc
-tracing = "0.1.44"
-url = "2.5.8"
-
-# Daemon
-smol = "2.0.2"
-
-# evgrd deps
-async-trait = {version = "0.1.89", optional = true}
-futures = {version = "0.3.32", optional = true}
-semver = {version = "1.0.28", optional = true}
-easy-parallel = {version = "3.3.1", optional = true}
-signal-hook-async-std = {version = "0.4.0", optional = true}
-signal-hook = {version = "0.4.4", optional = true}
-tracing-subscriber = { version = "0.3.23", default-features = false, features = ["fmt"], optional = true }
-tracing-appender = { version = "0.2.5", optional = true }
-serde = {version = "1.0.228", features = ["derive"], optional = true}
-structopt = {version = "0.3.26", optional = true}
-structopt-toml = {version = "0.5.1", optional = true}
-
-[features]
-build-daemon = [
-    "darkfi/async-daemonize",
-    "async-trait",
-    "futures",
-    "semver",
-    "easy-parallel",
-    "signal-hook-async-std",
-    "signal-hook",
-    "serde",
-    "structopt",
-    "structopt-toml",
-    "tracing-subscriber",
-    "tracing-appender"
-]
-
-#[lints]
-#workspace = true
-
-# Temp stuff
-[workspace]

script/evgrd/Makefile

@@ -1,14 +0,0 @@
-.POSIX:
-
-# Cargo binary
-CARGO = cargo
-
-BIN = evgrd
-
-all: $(BIN)
-
-evgrd:
-	$(CARGO) run --bin evgrd --features=build-daemon
-
-.PHONY: all
-

script/evgrd/bin/evgrd.rs

@@ -1,465 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2026 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, convert::TryInto, path::PathBuf, sync::Arc};
-
-use darkfi::{
-    async_daemonize, cli_desc,
-    event_graph::{
-        proto::{EventPut, ProtocolEventGraph},
-        Event, EventGraph, EventGraphPtr,
-    },
-    net::{
-        session::SESSION_DEFAULT,
-        settings::SettingsOpt as NetSettingsOpt,
-        transport::{Listener, PtListener, PtStream},
-        P2p, P2pPtr,
-    },
-    rpc::{
-        jsonrpc::JsonSubscriber,
-        server::{listen_and_serve, RequestHandler},
-        settings::RpcSettingsOpt,
-    },
-    system::{sleep, StoppableTask, StoppableTaskPtr},
-    util::path::expand_path,
-    Error, Result,
-};
-use darkfi_serial::{AsyncDecodable, AsyncEncodable};
-use futures::{AsyncWriteExt, FutureExt};
-use sled_overlay::sled;
-use smol::{fs, lock::Mutex, stream::StreamExt, Executor};
-use structopt_toml::{serde::Deserialize, structopt::StructOpt, StructOptToml};
-use tracing::{debug, error, info};
-use url::Url;
-
-use evgrd::{FetchEventsMessage, VersionMessage, MSG_EVENT, MSG_FETCHEVENTS, MSG_SENDEVENT};
-
-mod rpc;
-
-const CONFIG_FILE: &str = "evgrd.toml";
-const CONFIG_FILE_CONTENTS: &str = include_str!("../evgrd.toml");
-
-#[derive(Clone, Debug, Deserialize, StructOpt, StructOptToml)]
-#[serde(default)]
-#[structopt(name = "evgrd", about = cli_desc!())]
-struct Args {
-    #[structopt(short, parse(from_occurrences))]
-    /// Increase verbosity (-vvv supported)
-    verbose: u8,
-
-    #[structopt(short, long)]
-    /// Configuration file to use
-    config: Option<String>,
-
-    #[structopt(long)]
-    /// Set log file output
-    log: Option<String>,
-
-    #[structopt(long, default_value = "tcp://127.0.0.1:5588")]
-    /// RPC server listen address
-    daemon_listen: Vec<Url>,
-
-    #[structopt(short, long, default_value = "~/.local/share/darkfi/evgrd_db")]
-    /// Datastore (DB) path
-    datastore: String,
-
-    #[structopt(short, long, default_value = "~/.local/share/darkfi/replayed_evgrd_db")]
-    /// Replay logs (DB) path
-    replay_datastore: String,
-
-    #[structopt(long)]
-    /// Flag to store Sled DB instructions
-    replay_mode: bool,
-
-    #[structopt(long)]
-    /// Flag to skip syncing the DAG (no history)
-    skip_dag_sync: bool,
-
-    #[structopt(long, default_value = "5")]
-    /// Number of attempts to sync the DAG
-    sync_attempts: u8,
-
-    #[structopt(long, default_value = "15")]
-    /// Number of seconds to wait before trying again if sync fails
-    sync_timeout: u8,
-
-    #[structopt(flatten)]
-    /// P2P network settings
-    net: NetSettingsOpt,
-
-    #[structopt(flatten)]
-    /// JSON-RPC settings
-    rpc: RpcSettingsOpt,
-}
-
-pub struct Daemon {
-    /// P2P network pointer
-    p2p: P2pPtr,
-    ///// Sled DB (also used in event_graph and for RLN)
-    //sled: sled::Db,
-    /// Event Graph instance
-    event_graph: EventGraphPtr,
-    /// JSON-RPC connection tracker
-    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
-    /// dnet JSON-RPC subscriber
-    dnet_sub: JsonSubscriber,
-    /// deg JSON-RPC subscriber
-    deg_sub: JsonSubscriber,
-    /// Replay logs (DB) path
-    replay_datastore: PathBuf,
-}
-
-impl Daemon {
-    fn new(
-        p2p: P2pPtr,
-        //sled: sled::Db,
-        event_graph: EventGraphPtr,
-        dnet_sub: JsonSubscriber,
-        deg_sub: JsonSubscriber,
-        replay_datastore: PathBuf,
-    ) -> Self {
-        Self {
-            p2p,
-            //sled,
-            event_graph,
-            rpc_connections: Mutex::new(HashSet::new()),
-            dnet_sub,
-            deg_sub,
-            replay_datastore,
-        }
-    }
-}
-
-async fn rpc_serve(
-    listener: Box<dyn PtListener>,
-    daemon: Arc<Daemon>,
-    ex: Arc<Executor<'_>>,
-) -> Result<()> {
-    loop {
-        let connection = match listener.next().await {
-            Ok(negotiation) => negotiation.await,
-            Err(err) => Err(err),
-        };
-
-        match connection {
-            Ok((stream, url)) => {
-                info!(target: "evgrd", "Accepted connection from {url}");
-                let daemon = daemon.clone();
-                ex.spawn(async move {
-                    if let Err(e) = handle_connect(stream, daemon).await {
-                        error!(target: "evgrd", "Handle connect exited: {e}");
-                    }
-                })
-                .detach();
-            }
-
-            // Errors we didn't handle above:
-            Err(e) => {
-                error!(
-                    target: "evgrd",
-                    "Unhandled listener.next() error: {}", e,
-                );
-                continue
-            }
-        }
-    }
-}
-
-async fn handle_connect(mut stream: Box<dyn PtStream>, daemon: Arc<Daemon>) -> Result<()> {
-    let client_version = VersionMessage::decode_async(&mut stream).await?;
-    info!(target: "evgrd", "Client version: {}", client_version.protocol_version);
-
-    let version = VersionMessage::new();
-    version.encode_async(&mut stream).await?;
-    stream.flush().await?;
-    debug!(target: "darkirc", "Sent version: {version:?}");
-
-    let event_sub = daemon.event_graph.event_pub.clone().subscribe().await;
-
-    loop {
-        futures::select! {
-            ev = event_sub.receive().fuse() => {
-                MSG_EVENT.encode_async(&mut stream).await?;
-                stream.flush().await?;
-                ev.encode_async(&mut stream).await?;
-                stream.flush().await?;
-            }
-            msg_type = u8::decode_async(&mut stream).fuse() => {
-                debug!(target: "evgrd", "Received msg_type: {msg_type:?}");
-                let msg_type = msg_type?;
-                match msg_type {
-                    MSG_FETCHEVENTS => fetch_events(&mut stream, &daemon).await?,
-                    MSG_SENDEVENT => send_event(&mut stream, &daemon).await?,
-                    _ => error!(target: "evgrd", "Skipping unhandled msg_type: {msg_type}")
-                }
-            }
-        }
-    }
-}
-
-async fn fetch_events(stream: &mut Box<dyn PtStream>, daemon: &Daemon) -> Result<()> {
-    let fetchevs = FetchEventsMessage::decode_async(stream).await?;
-    info!(target: "evgrd", "Fetch events: {fetchevs:?}");
-    let events = daemon.event_graph.fetch_successors_of(fetchevs.unref_tips).await?;
-
-    let n_events = events.len();
-    for event in events {
-        MSG_EVENT.encode_async(stream).await?;
-        stream.flush().await?;
-        event.encode_async(stream).await?;
-        stream.flush().await?;
-    }
-    debug!(target: "evgrd", "Sent {n_events} for fetch");
-    Ok(())
-}
-
-async fn send_event(stream: &mut Box<dyn PtStream>, daemon: &Daemon) -> Result<()> {
-    let timestamp = u64::decode_async(stream).await?;
-    let content = Vec::<u8>::decode_async(stream).await?;
-    info!(target: "evgrd", "send_event: {timestamp}, {content:?}");
-
-    let event = Event::with_timestamp(timestamp, content, &daemon.event_graph).await;
-    daemon.event_graph.dag_insert(&[event.clone()]).await.unwrap();
-
-    info!(target: "evgrd", "Broadcasting event put: {event:?}");
-    //daemon.p2p.broadcast(&EventPut(event)).await;
-
-    let p2p = daemon.p2p.clone();
-    let self_version = p2p.settings().read().await.app_version.clone();
-    let connected_peers = p2p.hosts().peers();
-    let mut peers_with_matched_version = vec![];
-    let mut peers_with_different_version = vec![];
-    for peer in connected_peers {
-        let peer_version = peer.version.get();
-        if let Some(peer_version) = peer_version {
-            if self_version == peer_version.version {
-                peers_with_matched_version.push(peer)
-            } else {
-                peers_with_different_version.push(peer)
-            }
-        }
-    }
-
-    if !peers_with_matched_version.is_empty() {
-        p2p.broadcast_to(&EventPut(event.clone()), &peers_with_matched_version).await?;
-    }
-    if !peers_with_different_version.is_empty() {
-        let mut event = event;
-        event.timestamp /= 1000;
-        p2p.broadcast_to(&EventPut(event), &peers_with_different_version).await?;
-    }
-
-    Ok(())
-}
-
-async_daemonize!(realmain);
-async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
-    info!(target: "evgrd", "Starting evgrd node");
-
-    // Create datastore path if not there already.
-    let datastore = expand_path(&args.datastore)?;
-    fs::create_dir_all(&datastore).await?;
-
-    let replay_datastore = expand_path(&args.replay_datastore)?;
-    let replay_mode = args.replay_mode;
-
-    info!(target: "evgrd", "Instantiating event DAG");
-    let sled_db = sled::open(datastore)?;
-    let mut p2p_settings: darkfi::net::Settings =
-        (env!("CARGO_PKG_NAME"), env!("CARGO_PKG_VERSION"), args.net).try_into()?;
-    p2p_settings.seeds.push(url::Url::parse("tcp+tls://lilith1.dark.fi:5262").unwrap());
-    let p2p = P2p::new(p2p_settings, ex.clone()).await?;
-    let event_graph = EventGraph::new(
-        p2p.clone(),
-        sled_db.clone(),
-        replay_datastore.clone(),
-        replay_mode,
-        "evgrd_dag",
-        1,
-        ex.clone(),
-    )
-    .await?;
-
-    // Adding some events
-    // for i in 1..6 {
-    //     let event = Event::new(vec![1, 2, 3, i], &event_graph).await;
-    //     event_graph.dag_insert(&[event.clone()]).await.unwrap();
-    // }
-
-    let prune_task = event_graph.prune_task.get().unwrap();
-
-    info!(target: "evgrd", "Registering EventGraph P2P protocol");
-    let event_graph_ = Arc::clone(&event_graph);
-    let registry = p2p.protocol_registry();
-    registry
-        .register(SESSION_DEFAULT, move |channel, _| {
-            let event_graph_ = event_graph_.clone();
-            async move { ProtocolEventGraph::init(event_graph_, channel).await.unwrap() }
-        })
-        .await;
-
-    info!(target: "evgrd", "Starting dnet subs task");
-    let dnet_sub = JsonSubscriber::new("dnet.subscribe_events");
-    let dnet_sub_ = dnet_sub.clone();
-    let p2p_ = p2p.clone();
-    let dnet_task = StoppableTask::new();
-    dnet_task.clone().start(
-        async move {
-            let dnet_sub = p2p_.dnet_subscribe().await;
-            loop {
-                let event = dnet_sub.receive().await;
-                debug!(target: "evgrd", "Got dnet event: {:?}", event);
-                dnet_sub_.notify(vec![event.into()].into()).await;
-            }
-        },
-        |res| async {
-            match res {
-                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
-                Err(e) => panic!("{}", e),
-            }
-        },
-        Error::DetachedTaskStopped,
-        ex.clone(),
-    );
-
-    info!(target: "evgrd", "Starting deg subs task");
-    let deg_sub = JsonSubscriber::new("deg.subscribe_events");
-    let deg_sub_ = deg_sub.clone();
-    let event_graph_ = event_graph.clone();
-    let deg_task = StoppableTask::new();
-    deg_task.clone().start(
-        async move {
-            let deg_sub = event_graph_.deg_subscribe().await;
-            loop {
-                let event = deg_sub.receive().await;
-                debug!(target: "evgrd", "Got deg event: {:?}", event);
-                deg_sub_.notify(vec![event.into()].into()).await;
-            }
-        },
-        |res| async {
-            match res {
-                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
-                Err(e) => panic!("{}", e),
-            }
-        },
-        Error::DetachedTaskStopped,
-        ex.clone(),
-    );
-
-    info!(target: "evgrd", "Starting JSON-RPC server");
-    let daemon = Arc::new(Daemon::new(
-        p2p.clone(),
-        //sled_db.clone(),
-        event_graph.clone(),
-        dnet_sub,
-        deg_sub,
-        replay_datastore.clone(),
-    ));
-
-    // Used for deg and dnet
-    let daemon_ = daemon.clone();
-    let rpc_task = StoppableTask::new();
-    rpc_task.clone().start(
-        listen_and_serve(args.rpc.into(), daemon.clone(), None, ex.clone()),
-        |res| async move {
-            match res {
-                Ok(()) | Err(Error::RpcServerStopped) => daemon_.stop_connections().await,
-                Err(e) => error!(target: "evgrd", "Failed stopping JSON-RPC server: {}", e),
-            }
-        },
-        Error::RpcServerStopped,
-        ex.clone(),
-    );
-
-    info!(target: "evgrd", "Starting evgrd server");
-    let mut rpc_tasks = vec![];
-    for listen_url in args.daemon_listen {
-        let listener = Listener::new(listen_url, None).await?;
-        let ptlistener = listener.listen().await?;
-
-        let rpc_task = StoppableTask::new();
-        rpc_task.clone().start(
-            rpc_serve(ptlistener, daemon.clone(), ex.clone()),
-            |res| async move {
-                match res {
-                    Ok(()) => panic!("Acceptor task should never complete without error status"),
-                    //Err(Error::RpcServerStopped) => daemon_.stop_connections().await,
-                    Err(e) => error!(target: "evgrd", "Failed stopping RPC server: {}", e),
-                }
-            },
-            Error::RpcServerStopped,
-            ex.clone(),
-        );
-        rpc_tasks.push(rpc_task);
-    }
-
-    info!(target: "evgrd", "Starting P2P network");
-    p2p.clone().start().await?;
-
-    info!(target: "evgrd", "Waiting for some P2P connections...");
-    sleep(5).await;
-
-    // We'll attempt to sync {sync_attempts} times
-    if !args.skip_dag_sync {
-        for i in 1..=args.sync_attempts {
-            info!(target: "evgrd", "Syncing event DAG (attempt #{})", i);
-            match event_graph.dag_sync().await {
-                Ok(()) => break,
-                Err(e) => {
-                    if i == args.sync_attempts {
-                        error!(target: "evgrd", "Failed syncing DAG. Exiting.");
-                        p2p.stop().await;
-                        return Err(Error::DagSyncFailed)
-                    } else {
-                        // TODO: Maybe at this point we should prune or something?
-                        // TODO: Or maybe just tell the user to delete the DAG from FS.
-                        error!(target: "evgrd", "Failed syncing DAG ({}), retrying in {}s...", e, args.sync_timeout);
-                        sleep(args.sync_timeout.into()).await;
-                    }
-                }
-            }
-        }
-    } else {
-        *event_graph.synced.write().await = true;
-    }
-
-    // Signal handling for graceful termination.
-    let (signals_handler, signals_task) = SignalHandler::new(ex)?;
-    signals_handler.wait_termination(signals_task).await?;
-    info!(target: "evgrd", "Caught termination signal, cleaning up and exiting...");
-
-    info!(target: "evgrd", "Stopping P2P network");
-    p2p.stop().await;
-
-    info!(target: "evgrd", "Stopping RPC server");
-    for rpc_task in rpc_tasks {
-        rpc_task.stop().await;
-    }
-    dnet_task.stop().await;
-    deg_task.stop().await;
-
-    info!(target: "evgrd", "Stopping IRC server");
-    prune_task.stop().await;
-
-    info!(target: "evgrd", "Flushing sled database...");
-    let flushed_bytes = sled_db.flush_async().await?;
-    info!(target: "evgrd", "Flushed {} bytes", flushed_bytes);
-
-    info!(target: "evgrd", "Shut down successfully");
-    Ok(())
-}

script/evgrd/bin/rpc.rs

@@ -1,176 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2026 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;
-
-use async_trait::async_trait;
-use darkfi::{
-    event_graph::util::recreate_from_replayer_log,
-    net::P2pPtr,
-    rpc::{
-        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
-        p2p_method::HandlerP2p,
-        server::RequestHandler,
-        util::JsonValue,
-    },
-    system::StoppableTaskPtr,
-};
-use smol::lock::MutexGuard;
-use tracing::debug;
-
-use super::Daemon;
-
-#[async_trait]
-impl RequestHandler<()> for Daemon {
-    async fn handle_request(&self, req: JsonRequest) -> JsonResult {
-        debug!(target: "darkirc::rpc", "--> {}", req.stringify().unwrap());
-
-        match req.method.as_str() {
-            "ping" => self.pong(req.id, req.params).await,
-            "dnet.switch" => self.dnet_switch(req.id, req.params).await,
-            "dnet.subscribe_events" => self.dnet_subscribe_events(req.id, req.params).await,
-            "p2p.get_info" => self.p2p_get_info(req.id, req.params).await,
-
-            "deg.switch" => self.deg_switch(req.id, req.params).await,
-            "deg.subscribe_events" => self.deg_subscribe_events(req.id, req.params).await,
-            "eventgraph.get_info" => self.eg_get_info(req.id, req.params).await,
-            "eventgraph.replay" => self.eg_rep_info(req.id, req.params).await,
-
-            _ => JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
-        }
-    }
-
-    async fn connections_mut(&self) -> MutexGuard<'_, HashSet<StoppableTaskPtr>> {
-        self.rpc_connections.lock().await
-    }
-}
-
-impl Daemon {
-    // RPCAPI:
-    // Activate or deactivate dnet in the P2P stack.
-    // By sending `true`, dnet will be activated, and by sending `false` dnet
-    // will be deactivated. Returns `true` on success.
-    //
-    // --> {"jsonrpc": "2.0", "method": "dnet.switch", "params": [true], "id": 42}
-    // <-- {"jsonrpc": "2.0", "result": true, "id": 42}
-    async fn dnet_switch(&self, id: u16, params: JsonValue) -> JsonResult {
-        let params = params.get::<Vec<JsonValue>>().unwrap();
-        if params.len() != 1 || !params[0].is_bool() {
-            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-        }
-
-        let switch = params[0].get::<bool>().unwrap();
-
-        if *switch {
-            self.p2p.dnet_enable();
-        } else {
-            self.p2p.dnet_disable();
-        }
-
-        JsonResponse::new(JsonValue::Boolean(true), id).into()
-    }
-
-    // RPCAPI:
-    // Initializes a subscription to p2p dnet events.
-    // Once a subscription is established, `darkirc` will send JSON-RPC notifications of
-    // new network events to the subscriber.
-    //
-    // --> {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [], "id": 1}
-    // <-- {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [`event`]}
-    pub async fn dnet_subscribe_events(&self, id: u16, params: JsonValue) -> JsonResult {
-        let params = params.get::<Vec<JsonValue>>().unwrap();
-        if !params.is_empty() {
-            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-        }
-
-        self.dnet_sub.clone().into()
-    }
-
-    // RPCAPI:
-    // Initializes a subscription to deg events.
-    // Once a subscription is established, apps using eventgraph will send JSON-RPC notifications of
-    // new eventgraph events to the subscriber.
-    //
-    // --> {"jsonrpc": "2.0", "method": "deg.subscribe_events", "params": [], "id": 1}
-    // <-- {"jsonrpc": "2.0", "method": "deg.subscribe_events", "params": [`event`]}
-    pub async fn deg_subscribe_events(&self, id: u16, params: JsonValue) -> JsonResult {
-        let params = params.get::<Vec<JsonValue>>().unwrap();
-        if !params.is_empty() {
-            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-        }
-
-        self.deg_sub.clone().into()
-    }
-
-    // RPCAPI:
-    // Activate or deactivate deg in the EVENTGRAPH.
-    // By sending `true`, deg will be activated, and by sending `false` deg
-    // will be deactivated. Returns `true` on success.
-    //
-    // --> {"jsonrpc": "2.0", "method": "deg.switch", "params": [true], "id": 42}
-    // <-- {"jsonrpc": "2.0", "result": true, "id": 42}
-    async fn deg_switch(&self, id: u16, params: JsonValue) -> JsonResult {
-        let params = params.get::<Vec<JsonValue>>().unwrap();
-        if params.len() != 1 || !params[0].is_bool() {
-            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-        }
-
-        let switch = params[0].get::<bool>().unwrap();
-
-        if *switch {
-            self.event_graph.deg_enable().await;
-        } else {
-            self.event_graph.deg_disable().await;
-        }
-
-        JsonResponse::new(JsonValue::Boolean(true), id).into()
-    }
-
-    // RPCAPI:
-    // Get EVENTGRAPH info.
-    //
-    // --> {"jsonrpc": "2.0", "method": "deg.switch", "params": [true], "id": 42}
-    // <-- {"jsonrpc": "2.0", "result": true, "id": 42}
-    async fn eg_get_info(&self, id: u16, params: JsonValue) -> JsonResult {
-        let params_ = params.get::<Vec<JsonValue>>().unwrap();
-        if !params_.is_empty() {
-            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-        }
-
-        self.event_graph.eventgraph_info(id, params).await
-    }
-
-    // RPCAPI:
-    // Get replayed EVENTGRAPH info.
-    //
-    // --> {"jsonrpc": "2.0", "method": "eventgraph.replay", "params": ..., "id": 42}
-    // <-- {"jsonrpc": "2.0", "result": true, "id": 42}
-    async fn eg_rep_info(&self, id: u16, params: JsonValue) -> JsonResult {
-        let params_ = params.get::<Vec<JsonValue>>().unwrap();
-        if !params_.is_empty() {
-            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-        }
-
-        recreate_from_replayer_log(&self.replay_datastore).await
-    }
-}
-
-impl HandlerP2p for Daemon {
-    fn p2p(&self) -> P2pPtr {
-        self.p2p.clone()
-    }
-}

script/evgrd/evgrd.toml

@@ -1,7 +0,0 @@
-[net]
-datastore = "~/.local/share/darkfi/evgrd"
-hostlist = "~/.local/share/darkfi/evgrd/p2p_hostlist.tsv"
-
-[rpc]
-rpc_listen = "tcp://127.0.0.1:26690"
-rpc_disabled_methods = ["p2p.get_info"]

script/evgrd/example/recv.rs

@@ -1,108 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2026 Dyne.org foundation
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
- */
-
-use darkfi::{
-    event_graph::{self},
-    net::transport::Dialer,
-    util::path::expand_path,
-    Error, Result,
-};
-use darkfi_serial::{
-    async_trait, deserialize_async_partial, AsyncDecodable, AsyncEncodable, SerialDecodable,
-    SerialEncodable,
-};
-use sled_overlay::sled;
-use smol::fs;
-use tracing::{error, info};
-use url::Url;
-
-use evgrd::{FetchEventsMessage, LocalEventGraph, VersionMessage, MSG_EVENT, MSG_FETCHEVENTS};
-
-#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
-pub struct Privmsg {
-    pub channel: String,
-    pub nick: String,
-    pub msg: String,
-}
-
-async fn amain() -> Result<()> {
-    info!("Instantiating event DAG");
-    let ex = std::sync::Arc::new(smol::Executor::new());
-    let datastore = expand_path("~/.local/share/darkfi/evgrd-test-client")?;
-    fs::create_dir_all(&datastore).await?;
-    let sled_db = sled::open(datastore)?;
-
-    let evgr = LocalEventGraph::new(sled_db.clone(), "evgrd_testdag", 1, ex.clone()).await?;
-
-    let endpoint = "tcp://127.0.0.1:5588";
-    let endpoint = Url::parse(endpoint)?;
-
-    let dialer = Dialer::new(endpoint, None).await?;
-    let timeout = std::time::Duration::from_secs(60);
-
-    println!("Connecting...");
-    let mut stream = dialer.dial(Some(timeout)).await?;
-    println!("Connected!");
-
-    let version = VersionMessage::new();
-    version.encode_async(&mut stream).await?;
-
-    let server_version = VersionMessage::decode_async(&mut stream).await?;
-    println!("Server version: {}", server_version.protocol_version);
-
-    let unref_tips = evgr.unreferenced_tips.read().await.clone();
-    let fetchevs = FetchEventsMessage::new(unref_tips);
-    MSG_FETCHEVENTS.encode_async(&mut stream).await?;
-    fetchevs.encode_async(&mut stream).await?;
-
-    loop {
-        let msg_type = u8::decode_async(&mut stream).await?;
-        println!("Received: {msg_type:?}");
-        if msg_type != MSG_EVENT {
-            error!("Received invalid msg_type: {msg_type}");
-            return Err(Error::MalformedPacket)
-        }
-
-        let ev = event_graph::Event::decode_async(&mut stream).await?;
-
-        let genesis_timestamp = evgr.current_genesis.read().await.clone().timestamp;
-        let ev_id = ev.id();
-        if !evgr.dag.contains_key(ev_id.as_bytes()).unwrap() &&
-            ev.validate(&evgr.dag, genesis_timestamp, evgr.days_rotation, None).await?
-        {
-            println!("got {ev:?}");
-            evgr.dag_insert(&[ev.clone()]).await.unwrap();
-
-            let privmsg: Privmsg = match deserialize_async_partial(ev.content()).await {
-                Ok((v, _)) => v,
-                Err(e) => {
-                    println!("Failed deserializing incoming Privmsg event: {}", e);
-                    continue
-                }
-            };
-
-            println!("privmsg: {privmsg:?}");
-        } else {
-            println!("Event is invalid!")
-        }
-    }
-}
-
-fn main() {
-    let _ = smol::block_on(amain());
-}

script/evgrd/example/send.rs

@@ -1,71 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2026 Dyne.org foundation
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
- */
-
-use darkfi::{net::transport::Dialer, Result};
-use darkfi_serial::{
-    async_trait, serialize_async, AsyncDecodable, AsyncEncodable, SerialDecodable, SerialEncodable,
-};
-use std::time::UNIX_EPOCH;
-use url::Url;
-
-use evgrd::{VersionMessage, MSG_SENDEVENT};
-
-#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
-pub struct Privmsg {
-    pub channel: String,
-    pub nick: String,
-    pub msg: String,
-}
-
-async fn amain() -> Result<()> {
-    let endpoint = "tcp://127.0.0.1:5588";
-    let endpoint = Url::parse(endpoint)?;
-
-    let dialer = Dialer::new(endpoint, None).await?;
-    let timeout = std::time::Duration::from_secs(60);
-
-    println!("Connecting...");
-    let mut stream = dialer.dial(Some(timeout)).await?;
-    println!("Connected!");
-
-    let version = VersionMessage::new();
-    version.encode_async(&mut stream).await?;
-
-    let server_version = VersionMessage::decode_async(&mut stream).await?;
-    println!("Server version: {}", server_version.protocol_version);
-
-    let msg = Privmsg {
-        channel: "#random".to_string(),
-        nick: "anon".to_string(),
-        msg: "i'm so random!".to_string(),
-    };
-    let timestamp = UNIX_EPOCH.elapsed().unwrap().as_millis() as u64;
-
-    MSG_SENDEVENT.encode_async(&mut stream).await?;
-    timestamp.encode_async(&mut stream).await?;
-
-    let content: Vec<u8> = serialize_async(&msg).await;
-    content.encode_async(&mut stream).await?;
-
-    Ok(())
-}
-
-fn main() {
-    let is_success = smol::block_on(amain());
-    println!("Finished: {is_success:?}");
-}

script/evgrd/src/lib.rs

@@ -1,382 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2026 Dyne.org foundation
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
- */
-
-use darkfi::{
-    event_graph::{
-        util::{generate_genesis, millis_until_next_rotation, next_rotation_timestamp},
-        Event, GENESIS_CONTENTS, INITIAL_GENESIS, NULL_ID, N_EVENT_PARENTS,
-    },
-    system::{msleep, Publisher, PublisherPtr, StoppableTask, StoppableTaskPtr},
-    Error, Result,
-};
-use darkfi_serial::{
-    async_trait, deserialize_async, serialize_async, SerialDecodable, SerialEncodable,
-};
-use sled_overlay::{sled, SledTreeOverlay};
-use smol::{
-    lock::{OnceCell, RwLock},
-    Executor,
-};
-use std::{
-    collections::{BTreeMap, HashSet},
-    sync::Arc,
-};
-use tracing::{debug, error, info};
-
-pub const PROTOCOL_VERSION: u32 = 1;
-
-/// Atomic pointer to an [`EventGraph`] instance.
-pub type LocalEventGraphPtr = Arc<LocalEventGraph>;
-
-pub struct LocalEventGraph {
-    /// Sled tree containing the DAG
-    pub dag: sled::Tree,
-    /// The set of unreferenced DAG tips
-    pub unreferenced_tips: RwLock<BTreeMap<u64, HashSet<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. Additionally it is also used when we broadcast the
-    /// `TipRep` message telling peers about our unreferenced tips.
-    broadcasted_ids: RwLock<HashSet<blake3::Hash>>,
-    /// DAG Pruning Task
-    pub prune_task: OnceCell<StoppableTaskPtr>,
-    /// Event publisher, this notifies whenever an event is
-    /// inserted into the DAG
-    pub event_pub: PublisherPtr<Event>,
-    /// Current genesis event
-    pub current_genesis: RwLock<Event>,
-    /// Currently configured DAG rotation, in days
-    pub days_rotation: u64,
-    /// Flag signalling DAG has finished initial sync
-    pub synced: RwLock<bool>,
-    /// Enable graph debugging
-    pub deg_enabled: RwLock<bool>,
-}
-
-impl LocalEventGraph {
-    pub async fn new(
-        sled_db: sled::Db,
-        dag_tree_name: &str,
-        days_rotation: u64,
-        ex: Arc<Executor<'_>>,
-    ) -> Result<LocalEventGraphPtr> {
-        let dag = sled_db.open_tree(dag_tree_name)?;
-        let unreferenced_tips = RwLock::new(BTreeMap::new());
-        let broadcasted_ids = RwLock::new(HashSet::new());
-        let event_pub = Publisher::new();
-
-        // Create the current genesis event based on the `days_rotation`
-        let current_genesis = generate_genesis(days_rotation);
-        let self_ = Arc::new(Self {
-            dag: dag.clone(),
-            unreferenced_tips,
-            broadcasted_ids,
-            prune_task: OnceCell::new(),
-            event_pub,
-            current_genesis: RwLock::new(current_genesis.clone()),
-            days_rotation,
-            synced: RwLock::new(false),
-            deg_enabled: RwLock::new(false),
-        });
-
-        // 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())? {
-            info!(
-                target: "event_graph::new",
-                "[EVENTGRAPH] DAG does not contain current genesis, pruning existing data",
-            );
-            self_.dag_prune(current_genesis).await?;
-        }
-
-        // Find the unreferenced tips in the current DAG state.
-        *self_.unreferenced_tips.write().await = self_.find_unreferenced_tips().await;
-
-        // Spawn the DAG pruning task
-        if days_rotation > 0 {
-            let prune_task = StoppableTask::new();
-            let _ = self_.prune_task.set(prune_task.clone()).await;
-
-            prune_task.clone().start(
-                self_.clone().dag_prune_task(days_rotation),
-                |_| async move {
-                    info!(target: "event_graph::_handle_stop", "[EVENTGRAPH] Prune task stopped, flushing sled")
-                },
-                Error::DetachedTaskStopped,
-                ex.clone(),
-            );
-        }
-
-        Ok(self_)
-    }
-
-    async fn dag_prune(&self, genesis_event: Event) -> Result<()> {
-        debug!(target: "event_graph::dag_prune", "Pruning DAG...");
-
-        // Acquire exclusive locks to unreferenced_tips, broadcasted_ids and
-        // current_genesis while this operation is happening. We do this to
-        // ensure that during the pruning operation, no other operations are
-        // able to access the intermediate state which could lead to producing
-        // the wrong state after pruning.
-        let mut unreferenced_tips = self.unreferenced_tips.write().await;
-        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.
-        let mut batch = sled::Batch::default();
-        for key in self.dag.iter().keys() {
-            batch.remove(key.unwrap());
-        }
-        batch.insert(genesis_event.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);
-        }
-
-        // Clear unreferenced tips and bcast ids
-        *unreferenced_tips = BTreeMap::new();
-        unreferenced_tips.insert(0, HashSet::from([genesis_event.id()]));
-        *current_genesis = genesis_event;
-        *broadcasted_ids = HashSet::new();
-        drop(unreferenced_tips);
-        drop(broadcasted_ids);
-        drop(current_genesis);
-
-        debug!(target: "event_graph::dag_prune", "DAG pruned successfully");
-        Ok(())
-    }
-
-    /// Background task periodically pruning the DAG.
-    async fn dag_prune_task(self: Arc<Self>, days_rotation: u64) -> Result<()> {
-        // The DAG should periodically be pruned. This can be a configurable
-        // parameter. By pruning, we should deterministically replace the
-        // genesis event (can use a deterministic timestamp) and drop everything
-        // in the DAG, leaving just the new genesis event.
-        debug!(target: "event_graph::dag_prune_task", "Spawned background DAG pruning task");
-
-        loop {
-            // Find the next rotation timestamp:
-            let next_rotation = next_rotation_timestamp(INITIAL_GENESIS, days_rotation);
-
-            // 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,
-            };
-
-            // Sleep until it's time to rotate.
-            let s = millis_until_next_rotation(next_rotation);
-
-            debug!(target: "event_graph::dag_prune_task", "Sleeping {}s until next DAG prune", s);
-            msleep(s).await;
-            debug!(target: "event_graph::dag_prune_task", "Rotation period reached");
-
-            // Trigger DAG prune
-            self.dag_prune(current_genesis).await?;
-        }
-    }
-
-    /// Find the unreferenced tips in the current DAG state, mapped by their layers.
-    async fn find_unreferenced_tips(&self) -> BTreeMap<u64, HashSet<blake3::Hash>> {
-        // First get all the event IDs
-        let mut tips = HashSet::new();
-        for iter_elem in self.dag.iter() {
-            let (id, _) = iter_elem.unwrap();
-            let id = blake3::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() {
-            let (_, event) = iter_elem.unwrap();
-            let event: Event = deserialize_async(&event).await.unwrap();
-            for parent in event.parents.iter() {
-                tips.remove(parent);
-            }
-        }
-
-        // Build the layers map
-        let mut map: BTreeMap<u64, HashSet<blake3::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) {
-                layer_tips.insert(tip);
-            } else {
-                let mut layer_tips = HashSet::new();
-                layer_tips.insert(tip);
-                map.insert(event.layer, layer_tips);
-            }
-        }
-
-        map
-    }
-
-    pub async fn dag_insert(&self, events: &[Event]) -> Result<Vec<blake3::Hash>> {
-        // Sanity check
-        if events.is_empty() {
-            return Ok(vec![])
-        }
-
-        // Acquire exclusive locks to `unreferenced_tips and broadcasted_ids`
-        let mut unreferenced_tips = self.unreferenced_tips.write().await;
-        let mut broadcasted_ids = self.broadcasted_ids.write().await;
-
-        // Here we keep the IDs to return
-        let mut ids = Vec::with_capacity(events.len());
-
-        // Create an overlay over the DAG tree
-        let mut overlay = SledTreeOverlay::new(&self.dag);
-
-        // Grab genesis timestamp
-        let genesis_timestamp = self.current_genesis.read().await.timestamp;
-
-        // Iterate over given events to validate them and
-        // write them to the overlay
-        for event in events {
-            let event_id = event.id();
-            debug!(
-                target: "event_graph::dag_insert",
-                "Inserting event {} into the DAG", event_id,
-            );
-
-            if !event
-                .validate(&self.dag, genesis_timestamp, self.days_rotation, Some(&overlay))
-                .await?
-            {
-                error!(target: "event_graph::dag_insert", "Event {} is invalid!", event_id);
-                return Err(Error::EventIsInvalid)
-            }
-
-            let event_se = serialize_async(event).await;
-
-            // Add the event to the overlay
-            overlay.insert(event_id.as_bytes(), &event_se)?;
-
-            // Note down the event ID to return
-            ids.push(event_id);
-        }
-
-        // 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.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();
-
-            // Update the unreferenced DAG tips set
-            debug!(
-                target: "event_graph::dag_insert",
-                "Event {} parents {:#?}", event_id, event.parents,
-            );
-            for parent_id in event.parents.iter() {
-                if parent_id != &NULL_ID {
-                    debug!(
-                        target: "event_graph::dag_insert",
-                        "Removing {} from unreferenced_tips", parent_id,
-                    );
-
-                    // Iterate over unreferenced tips in previous layers
-                    // and remove the parent
-                    // NOTE: this might be too exhaustive, but the
-                    // assumption is that previous layers unreferenced
-                    // tips will be few.
-                    for (layer, tips) in unreferenced_tips.iter_mut() {
-                        if layer >= &event.layer {
-                            continue
-                        }
-                        tips.remove(parent_id);
-                    }
-                    broadcasted_ids.insert(*parent_id);
-                }
-            }
-            unreferenced_tips.retain(|_, tips| !tips.is_empty());
-            debug!(
-                target: "event_graph::dag_insert",
-                "Adding {} to unreferenced tips", event_id,
-            );
-
-            if let Some(layer_tips) = unreferenced_tips.get_mut(&event.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);
-            }
-
-            // Send out notifications about the new event
-            self.event_pub.notify(event.clone()).await;
-        }
-
-        // Drop the exclusive locks
-        drop(unreferenced_tips);
-        drop(broadcasted_ids);
-
-        Ok(ids)
-    }
-
-    /// 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) };
-        let event: Event = deserialize_async(&bytes).await?;
-
-        Ok(Some(event))
-    }
-}
-
-#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct VersionMessage {
-    pub protocol_version: u32,
-}
-
-impl VersionMessage {
-    pub fn new() -> Self {
-        Self { protocol_version: PROTOCOL_VERSION }
-    }
-}
-
-impl Default for VersionMessage {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct FetchEventsMessage {
-    pub unref_tips: BTreeMap<u64, HashSet<blake3::Hash>>,
-}
-
-impl FetchEventsMessage {
-    pub fn new(unref_tips: BTreeMap<u64, HashSet<blake3::Hash>>) -> Self {
-        Self { unref_tips }
-    }
-}
-
-pub const MSG_EVENT: u8 = 1;
-pub const MSG_FETCHEVENTS: u8 = 2;
-pub const MSG_SENDEVENT: u8 = 3;