dkg: add initial handler with requests

Author: version513

Diff for: src/core/beacon.rs

@@ -8,11 +8,15 @@ use crate::key::toml::PairToml;
 use crate::key::toml::Toml;
 use crate::key::ConversionError;
 use crate::key::Scheme;
-use crate::protobuf::drand::ChainInfoPacket;
-use crate::protobuf::drand::IdentityResponse;
+
+use crate::dkg::dkg_handler::DkgActions;
+use crate::dkg::dkg_handler::DkgHandler;
 use crate::store::ChainStore;
 use crate::store::NewStore;
 
+use crate::protobuf::drand::ChainInfoPacket;
+use crate::protobuf::drand::IdentityResponse;
+
 use tracing::debug;
 use tracing::error;
 use tracing::info_span;
@@ -44,14 +48,15 @@ pub enum BeaconCmd {
     IdentityRequest(Callback<IdentityResponse, ConversionError>),
     Sync(Callback<Arc<ChainStore>, &'static str>),
     ChainInfo(Callback<ChainInfoPacket, &'static str>),
+    DkgActions(DkgActions),
 }
 
 #[derive(thiserror::Error, Debug)]
 /// For diagnostic, should not be possible.
 #[error("callback receiver has already been dropped")]
 struct SendError;
 
-#[derive(PartialEq, Eq, Clone)]
+#[derive(PartialEq, Eq)]
 pub struct BeaconID {
     inner: Arc<str>,
 }
@@ -101,10 +106,6 @@ impl<S: Scheme> BeaconProcess<S> {
         Ok(process)
     }
 
-    pub fn tracker(&self) -> &TaskTracker {
-        &self.tracker
-    }
-
     pub fn run(fs: FileStore, pair: PairToml, id: &str) -> Result<BeaconHandler, FileStoreError> {
         let chain_handler = ChainHandler::try_init(&fs, id)?;
         let chain_store = Arc::new(ChainStore::new(&fs.db_path(), S::Beacon::is_chained())?);
@@ -113,6 +114,8 @@ impl<S: Scheme> BeaconProcess<S> {
 
         let (tx, mut rx) = mpsc::channel::<BeaconCmd>(5);
         tracker.spawn(async move {
+            let dkg_handler = DkgHandler::fresh_install(node.clone());
+
             while let Some(cmd) = rx.recv().await {
                 match cmd {
                     BeaconCmd::Shutdown(callback) => {
@@ -136,6 +139,7 @@ impl<S: Scheme> BeaconProcess<S> {
                             error!("failed to send chain_info, receiver is dropped")
                         }
                     }
+                    BeaconCmd::DkgActions(action) => dkg_handler.new_action(action).await,
                 }
             }
         });
@@ -173,6 +177,14 @@ impl<S: Scheme> BeaconProcess<S> {
 
         Ok(())
     }
+
+    pub fn tracker(&self) -> &TaskTracker {
+        &self.tracker
+    }
+
+    pub fn id(&self) -> &BeaconID {
+        &self.beacon_id
+    }
 }
 
 impl<S: Scheme> std::ops::Deref for BeaconProcess<S> {
@@ -182,3 +194,17 @@ impl<S: Scheme> std::ops::Deref for BeaconProcess<S> {
         &self.0
     }
 }
+
+impl<S: Scheme> Clone for BeaconProcess<S> {
+    fn clone(&self) -> Self {
+        Self(Arc::clone(&self.0))
+    }
+}
+
+impl Clone for BeaconID {
+    fn clone(&self) -> Self {
+        Self {
+            inner: Arc::clone(&self.inner),
+        }
+    }
+}

Diff for: src/dkg/dkg_handler.rs

@@ -0,0 +1,126 @@
+use super::state_machine::db_state::DBState;
+use super::state_machine::db_state::DBStateError;
+
+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::GossipData;
+use crate::transport::dkg::GossipPacket;
+
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+use tracing::debug;
+use tracing::error;
+
+/// Callback for all dkg actions
+pub type DkgCallback<T> = tokio::sync::oneshot::Sender<Result<T, DkgActionsError>>;
+
+/// For diagnostic purposes, should be never possible
+const ERR_SEND: &str = "dkg receiver is dropped";
+
+pub struct DkgHandler {
+    sender: mpsc::Sender<DkgActions>,
+}
+
+pub enum DkgActions {
+    Gossip(GossipPacket, DkgCallback<()>),
+    Shutdown(DkgCallback<()>),
+    Command(DkgCommand, DkgCallback<()>),
+    Broadcast(Bundle, DkgCallback<()>),
+    Status(DkgCallback<DkgStatusResponse>),
+}
+
+#[derive(thiserror::Error, Debug)]
+pub enum DkgActionsError {
+    #[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) {
+        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 tracker = bp.tracker().clone();
+
+        tracker.spawn(async move {
+            let me = bp.keypair.public_identity();
+            let mut db_state = DBState::<S>::new_fresh(bp.id());
+
+            while let Some(actions) = receiver.recv().await {
+                handle_actions(&mut db_state, me, actions).await;
+            }
+
+            debug!("dkg handler is shutting down")
+        });
+
+        Self { sender }
+    }
+}
+
+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"),
+    }
+}
+
+/// 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>,
+    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()),
+    }
+}
+
+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}",);
+    };
+}

Diff for: src/dkg/mod.rs

@@ -1 +1,2 @@
+pub mod dkg_handler;
 pub mod state_machine;

Diff for: src/dkg/state_machine/db_state.rs

@@ -1,6 +1,7 @@
 use super::status::StateError;
 use super::status::Status;
 
+use crate::core::beacon::BeaconID;
 use crate::key::group::minimum_t;
 use crate::key::group::Group;
 use crate::key::keys::Identity;
@@ -17,6 +18,7 @@ use crate::transport::dkg::Timestamp;
 
 use crate::net::utils::Seconds;
 use std::time::SystemTime;
+use tracing::debug;
 
 #[derive(thiserror::Error, Debug)]
 pub enum DBStateError {
@@ -106,7 +108,7 @@ pub enum DBStateError {
     ReceivedRejection,
     #[error("dkg state error: {0}")]
     InvalidStateChange(#[from] StateError),
-    #[error("dkg: conversion: {0}")]
+    #[error("conversion: {0}")]
     ConversionError(#[from] ConversionError),
     #[error("the key's scheme may not match the beacon's scheme")]
     InvalidKeyScheme,
@@ -115,7 +117,7 @@ pub enum DBStateError {
 #[derive(PartialEq)]
 pub struct DBState<S: Scheme> {
     // Parameters
-    beacon_id: String,
+    beacon_id: BeaconID,
     epoch: u32,
     state: Status,
     threshold: u32,
@@ -137,6 +139,10 @@ pub struct DBState<S: Scheme> {
 }
 
 impl<S: Scheme> DBState<S> {
+    pub fn id(&self) -> &BeaconID {
+        &self.beacon_id
+    }
+
     pub fn show_status(&self) -> &super::status::Status {
         &self.state
     }
@@ -152,7 +158,7 @@ impl<S: Scheme> DBState<S> {
     }
 
     /// Returns default DBState representation which is used only at fresh state.
-    pub fn new_fresh(beacon_id: &str) -> Self {
+    pub fn new_fresh(beacon_id: &BeaconID) -> Self {
         Self {
             beacon_id: beacon_id.to_owned(),
             genesis_time: Timestamp {
@@ -223,7 +229,7 @@ impl<S: Scheme> DBState<S> {
             // TODO: Complete entry should be updated if dkg is finished.
             complete: None,
             current: Some(DkgEntry {
-                beacon_id: beacon_id.to_owned(),
+                beacon_id: beacon_id.to_string(),
                 state: *state as u32,
                 epoch: *epoch,
                 threshold: *threshold,
@@ -312,6 +318,7 @@ impl<S: Scheme> DBState<S> {
             return Err(DBStateError::SelfMissingFromProposal);
         }
 
+        debug!("received proposal is valid");
         *self = proposed;
 
         Ok(())
@@ -363,7 +370,7 @@ impl<S: Scheme> TryFrom<ProposalTerms> for DBState<S> {
         } = value;
 
         let state = DBState::<S> {
-            beacon_id,
+            beacon_id: BeaconID::new(beacon_id),
             epoch,
             state: Status::Proposed,
             threshold,