dkg: add dkg_join + test

Author: version513

Diff for: src/cli.rs

@@ -7,16 +7,19 @@ use crate::key::Scheme;
 
 use crate::net::control;
 use crate::net::control::ControlClient;
+use crate::net::dkg_control::DkgControlClient;
 use crate::net::protocol;
 use crate::net::utils::Address;
 use crate::net::utils::ControlListener;
 use crate::net::utils::NodeListener;
 
-use anyhow::bail;
-use anyhow::Result;
 use clap::arg;
 use clap::command;
 use clap::Parser;
+use clap::Subcommand;
+
+use anyhow::bail;
+use anyhow::Result;
 use energon::drand::schemes::*;
 
 /// Generate the long-term keypair (drand.private, drand.public) for this node, and load it on the drand daemon if it is up and running
@@ -39,7 +42,7 @@ pub struct KeyGenConfig {
 }
 
 /// Start the drand daemon.
-#[derive(Debug, Parser, Clone)] //TODO: mv Clone to tests
+#[derive(Debug, Parser, Clone)]
 pub struct Config {
     /// Folder to keep all drand cryptographic information, with absolute path.
     #[arg(long, default_value_t = FileStore::drand_home())]
@@ -81,6 +84,19 @@ pub struct SyncConfig {
     pub follow: bool,
 }
 
+#[derive(Subcommand, Clone, Debug)]
+/// Commands for interacting with the DKG
+pub enum Dkg {
+    Join {
+        /// Set the port you want to listen to for control port commands. If not specified, we will use the default value.
+        #[arg(long, default_value = control::DEFAULT_CONTROL_PORT)]
+        control: String,
+        /// Indicates the id for the randomness generation process which will be started
+        #[arg(long)]
+        id: String,
+    },
+}
+
 #[derive(Debug, Parser, Clone)]
 #[command(name = "git")]
 #[command(about = "", long_about = None)]
@@ -114,6 +130,8 @@ pub enum Cmd {
         id: String,
     },
     Sync(SyncConfig),
+    #[command(subcommand)]
+    Dkg(Dkg),
 }
 
 impl CLI {
@@ -128,6 +146,9 @@ impl CLI {
             Cmd::Load { control, id } => load_cmd(&control, &id).await?,
             Cmd::Stop { control, id } => stop_cmd(&control, id.as_deref()).await?,
             Cmd::Sync(config) => sync_cmd(config).await?,
+            Cmd::Dkg(dkg) => match dkg {
+                Dkg::Join { control, id } => join_cmd(&control, &id).await?,
+            },
         }
 
         Ok(())
@@ -200,7 +221,7 @@ async fn load_cmd(control_port: &str, beacon_id: &str) -> Result<()> {
     Ok(())
 }
 
-pub async fn stop_cmd(control_port: &str, beacon_id: Option<&str>) -> anyhow::Result<()> {
+async fn stop_cmd(control_port: &str, beacon_id: Option<&str>) -> anyhow::Result<()> {
     let mut conn = ControlClient::new(control_port).await?;
 
     match conn.shutdown(beacon_id).await {
@@ -229,3 +250,10 @@ async fn sync_cmd(config: SyncConfig) -> Result<()> {
 
     Ok(())
 }
+
+async fn join_cmd(control_port: &str, beacon_id: &str) -> Result<()> {
+    let mut client = DkgControlClient::new(control_port).await?;
+    client.dkg_join(beacon_id).await?;
+
+    Ok(())
+}

Diff for: src/core/beacon.rs

@@ -9,7 +9,7 @@ use crate::key::toml::Toml;
 use crate::key::ConversionError;
 use crate::key::Scheme;
 
-use crate::dkg::dkg_handler::DkgActions;
+use crate::dkg::dkg_handler::Actions;
 use crate::dkg::dkg_handler::DkgHandler;
 use crate::store::ChainStore;
 use crate::store::NewStore;
@@ -48,7 +48,7 @@ pub enum BeaconCmd {
     IdentityRequest(Callback<IdentityResponse, ConversionError>),
     Sync(Callback<Arc<ChainStore>, &'static str>),
     ChainInfo(Callback<ChainInfoPacket, &'static str>),
-    DkgActions(DkgActions),
+    DkgActions(Actions),
 }
 
 #[derive(thiserror::Error, Debug)]

Diff for: src/dkg/actions_active.rs

@@ -0,0 +1,55 @@
+//! This module contains all the DKG actions that require user interaction:
+//! creating a network, accepting or rejecting a DKG, getting the status, etc.
+//! Both leader and follower interactions are contained herein
+
+use super::dkg_handler::ActionsError;
+use super::state::State;
+
+use crate::key::group::Group;
+use crate::key::keys::Identity;
+use crate::key::Scheme;
+
+use crate::protobuf::dkg::DkgStatusResponse;
+use crate::protobuf::dkg::JoinOptions;
+use crate::transport::dkg::Command;
+
+use tracing::info;
+
+pub(super) async fn command<S: Scheme>(
+    cmd: Command,
+    state: &mut State<S>,
+    me: &Identity<S>,
+) -> Result<(), ActionsError> {
+    info!("running DKG command: {cmd}, beaconID: {}", state.id());
+
+    match cmd {
+        Command::Join(join_options) => start_join(me, state, join_options).await,
+        _ => super::dkg_handler::todo_request(&cmd),
+    }
+}
+
+pub(super) async fn start_join<S: Scheme>(
+    me: &Identity<S>,
+    state: &mut State<S>,
+    options: JoinOptions,
+) -> Result<(), ActionsError> {
+    let prev_group: Option<Group<S>> = if state.epoch() > 1 {
+        // TODO: reshape
+        drop(options);
+        panic!("start_join: epoch can not be bigger than 1, reshape is not implemented yet")
+    } else {
+        None
+    };
+
+    state
+        .joined(me, prev_group)
+        .map_err(ActionsError::DBState)?;
+    // joiners don't need to gossip anything
+
+    Ok(())
+}
+
+/// TODO: this method should make request to dkg.db
+pub(super) fn status<S: Scheme>(state: &State<S>) -> Result<DkgStatusResponse, ActionsError> {
+    state.status().map_err(ActionsError::DBState)
+}

Diff for: src/dkg/state_machine/convert.rs renamed to src/dkg/convert.rs

@@ -1,61 +1,85 @@
-use super::state_machine::db_state::DBState;
-use super::state_machine::db_state::DBStateError;
+use std::fmt::Display;
+
+use super::actions_active as active;
+use super::state::DBStateError;
+use super::state::State;
 
 use crate::core::beacon::BeaconProcess;
 use crate::key::keys::Identity;
 use crate::key::Scheme;
 
 use crate::protobuf::dkg::packet::Bundle;
 use crate::protobuf::dkg::DkgStatusResponse;
-use crate::transport::dkg::DkgCommand;
+use crate::transport::dkg::Command;
 use crate::transport::dkg::GossipData;
 use crate::transport::dkg::GossipPacket;
 
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 use tracing::debug;
 use tracing::error;
+use tracing::warn;
+
+/// Callback for DKG actions
+pub struct Callback<T> {
+    inner: oneshot::Sender<Result<T, ActionsError>>,
+}
+
+/// For diagnostic purposes, this should never happen
+const ERR_SEND: &str = "dkg callback receiver is dropped";
 
-/// Callback for all dkg actions
-pub type DkgCallback<T> = tokio::sync::oneshot::Sender<Result<T, DkgActionsError>>;
+impl<T> Callback<T> {
+    /// Sends a response back through the callback channel.
+    fn reply(self, result: Result<T, ActionsError>) {
+        if let Err(err) = &result {
+            error!("failed to proceed dkg request: {err}")
+        }
+        if self.inner.send(result).is_err() {
+            error!("{ERR_SEND}");
+        };
+    }
+}
 
-/// For diagnostic purposes, should be never possible
-const ERR_SEND: &str = "dkg receiver is dropped";
+impl<T> From<oneshot::Sender<Result<T, ActionsError>>> for Callback<T> {
+    fn from(tx: oneshot::Sender<Result<T, ActionsError>>) -> Self {
+        Self { inner: tx }
+    }
+}
 
 pub struct DkgHandler {
-    sender: mpsc::Sender<DkgActions>,
+    sender: mpsc::Sender<Actions>,
 }
 
-pub enum DkgActions {
-    Gossip(GossipPacket, DkgCallback<()>),
-    Shutdown(DkgCallback<()>),
-    Command(DkgCommand, DkgCallback<()>),
-    Broadcast(Bundle, DkgCallback<()>),
-    Status(DkgCallback<DkgStatusResponse>),
+pub enum Actions {
+    Gossip(GossipPacket, Callback<()>),
+    Shutdown(Callback<()>),
+    Command(Command, Callback<()>),
+    Broadcast(Bundle, Callback<()>),
+    Status(Callback<DkgStatusResponse>),
 }
 
 #[derive(thiserror::Error, Debug)]
-pub enum DkgActionsError {
+pub enum ActionsError {
     #[error("db state error: {0}")]
     DBState(#[from] DBStateError),
     #[error("TODO: this dkg action is not implemented yet")]
     TODO,
 }
 
 impl DkgHandler {
-    pub async fn new_action(&self, action: DkgActions) {
+    pub async fn new_action(&self, action: Actions) {
         if self.sender.send(action).await.is_err() {
             error!("{ERR_SEND}")
         }
     }
 
     pub fn fresh_install<S: Scheme>(bp: BeaconProcess<S>) -> Self {
-        let (sender, mut receiver) = mpsc::channel::<DkgActions>(1);
+        let (sender, mut receiver) = mpsc::channel::<Actions>(1);
         let tracker = bp.tracker().clone();
 
         tracker.spawn(async move {
             let me = bp.keypair.public_identity();
-            let mut db_state = DBState::<S>::new_fresh(bp.id());
+            let mut db_state = State::<S>::new_fresh(bp.id());
 
             while let Some(actions) = receiver.recv().await {
                 handle_actions(&mut db_state, me, actions).await;
@@ -68,59 +92,33 @@ impl DkgHandler {
     }
 }
 
-async fn handle_actions<S: Scheme>(db: &mut DBState<S>, me: &Identity<S>, actions: DkgActions) {
-    match actions {
-        DkgActions::Shutdown(callback) => todo_request(callback, "shutdown"),
-        DkgActions::Status(callback) => status(db, callback),
-        DkgActions::Command(command, callback) => handle_command(command, callback).await,
-        DkgActions::Gossip(gossip, callback) => handle_gossip(db, me, gossip, callback).await,
-        DkgActions::Broadcast(_bundle, callback) => todo_request(callback, "broadcast"),
+async fn handle_actions<S: Scheme>(db: &mut State<S>, me: &Identity<S>, request: Actions) {
+    match request {
+        Actions::Shutdown(callback) => callback.reply(todo_request("shutdown")),
+        Actions::Status(callback) => callback.reply(active::status(db)),
+        Actions::Command(cmd, callback) => callback.reply(active::command(cmd, db, me).await),
+        Actions::Gossip(packet, callback) => callback.reply(gossip(db, me, packet).await),
+        Actions::Broadcast(bundle, callback) => callback.reply(todo_request(&bundle.to_string())),
     }
 }
 
-/// Dkg status request
-fn status<S: Scheme>(db: &mut DBState<S>, callback: DkgCallback<DkgStatusResponse>) {
-    if callback
-        .send(db.status().map_err(|err| {
-            error!("dkg status request, id: {}, error: {err}", db.id());
-            err.into()
-        }))
-        .is_err()
-    {
-        error!("dkg status request, id: {}, error: {ERR_SEND}", db.id());
-    };
-}
-
-async fn handle_command(dkg_command: DkgCommand, callback: DkgCallback<()>) {
-    let _metadata = &dkg_command.metadata;
-    todo_request(callback, dkg_command.command.to_string().as_str())
-}
-
-async fn handle_gossip<S: Scheme>(
-    db: &mut DBState<S>,
+async fn gossip<S: Scheme>(
+    db: &mut State<S>,
     me: &Identity<S>,
-    gossip: crate::transport::dkg::GossipPacket,
-    callback: oneshot::Sender<Result<(), DkgActionsError>>,
-) {
-    let metadata = &gossip.metadata;
-    match gossip.data {
-        GossipData::Proposal(proposal) => {
-            if callback
-                .send(db.proposed(me, proposal, metadata).map_err(|err| {
-                    error!("received dkg proposal, id: {}, error: {err}", db.id());
-                    err.into()
-                }))
-                .is_err()
-            {
-                error!("received dkg proposal, id: {}, error: {ERR_SEND}", db.id());
-            };
-        }
-        _ => todo_request(callback, gossip.data.to_string().as_str()),
+    packet: GossipPacket,
+) -> Result<(), ActionsError> {
+    let meta = &packet.metadata;
+    match packet.data {
+        GossipData::Proposal(proposal) => db.proposed(me, proposal, meta)?,
+        _ => todo_request(&packet.data)?,
     }
+
+    Ok(())
 }
 
-fn todo_request<T>(callback: DkgCallback<T>, kind: &str) {
-    if callback.send(Err(DkgActionsError::TODO)).is_err() {
-        error!("failed to proceed {kind}, error: {ERR_SEND}",);
-    };
+/// Temporary tracker for unfinished logic
+pub fn todo_request<D: Display + ?Sized>(kind: &D) -> Result<(), ActionsError> {
+    warn!("received TODO request: {kind}");
+
+    Err(ActionsError::TODO)
 }

Diff for: src/dkg/dkg_handler.rs

@@ -1,2 +1,5 @@
+pub mod actions_active;
+pub(super) mod convert;
 pub mod dkg_handler;
-pub mod state_machine;
+pub mod state;
+pub mod status;