feat(dkg): Port node signature exchange from Charon

crates/dkg/Cargo.toml

@@ -13,6 +13,7 @@ thiserror.workspace = true
 libp2p.workspace = true
 futures.workspace = true
 tokio.workspace = true
+tokio-util.workspace = true
 sha2.workspace = true
 tracing.workspace = true
 either.workspace = true
@@ -35,6 +36,7 @@ pluto-build-proto.workspace = true
 
 [dev-dependencies]
 anyhow.workspace = true
+test-case.workspace = true
 clap.workspace = true
 hex.workspace = true
 pluto-cluster = { workspace = true, features = ["test-cluster"] }

crates/dkg/examples/bcast.rs

@@ -299,14 +299,16 @@ async fn register_message(component: &Component, local_node_number: u32) -> bcas
                 Ok(())
             }),
             Box::new(move |peer_id, received_msg_id, msg| {
-                info!(
-                    local_node = local_node_number,
-                    sender = %peer_id,
-                    msg_id = received_msg_id,
-                    msg = ?msg,
-                    "Received broadcast"
-                );
-                Ok(())
+                Box::pin(async move {
+                    info!(
+                        local_node = local_node_number,
+                        sender = %peer_id,
+                        msg_id = received_msg_id,
+                        msg = ?msg,
+                        "Received broadcast"
+                    );
+                    Ok(())
+                })
             }),
         )
         .await

crates/dkg/src/bcast/behaviour.rs

@@ -585,15 +585,18 @@ mod tests {
                 "timestamp",
                 Box::new(|_peer_id, _msg| Ok(())),
                 Box::new(move |peer_id, msg_id, msg| {
-                    receipt_tx
-                        .send(Receipt {
-                            target: node_index,
-                            source: peer_id,
-                            msg_id: msg_id.to_string(),
-                            seconds: msg.seconds,
-                        })
-                        .map_err(|_| Error::ReceiptChannelClosed)?;
-                    Ok(())
+                    let receipt_tx = receipt_tx.clone();
+                    Box::pin(async move {
+                        receipt_tx
+                            .send(Receipt {
+                                target: node_index,
+                                source: peer_id,
+                                msg_id,
+                                seconds: msg.seconds,
+                            })
+                            .map_err(|_| Error::ReceiptChannelClosed)?;
+                        Ok(())
+                    })
                 }),
             )
             .await

crates/dkg/src/bcast/component.rs

@@ -2,6 +2,7 @@
 
 use std::{collections::HashMap, sync::Arc};
 
+use futures::future::BoxFuture;
 use libp2p::PeerId;
 use prost::{Message, Name};
 use prost_types::Any;
@@ -13,7 +14,12 @@ use super::error::{Error, Result};
 pub type CheckFn<M> = Box<dyn Fn(PeerId, &M) -> Result<()> + Send + Sync + 'static>;
 
 /// Typed message callback invoked for validated broadcast messages.
-pub type CallbackFn<M> = Box<dyn Fn(PeerId, &str, M) -> Result<()> + Send + Sync + 'static>;
+///
+/// The returned future is awaited by the inbound message handler, allowing
+/// the callback to perform async operations (e.g. waiting for state that
+/// becomes available later).
+pub type CallbackFn<M> =
+    Box<dyn Fn(PeerId, String, M) -> BoxFuture<'static, Result<()>> + Send + Sync + 'static>;
 
 pub(crate) type Registry = Arc<RwLock<HashMap<String, Arc<dyn RegisteredMessage>>>>;
 
@@ -33,7 +39,8 @@ pub(crate) trait RegisteredMessage: Send + Sync {
     fn check(&self, peer_id: PeerId, any: &Any) -> Result<()>;
 
     /// Dispatches the incoming wrapped protobuf message to the typed callback.
-    fn callback(&self, peer_id: PeerId, msg_id: &str, any: &Any) -> Result<()>;
+    fn callback(&self, peer_id: PeerId, msg_id: String, any: Any)
+    -> BoxFuture<'static, Result<()>>;
 }
 
 struct TypedRegistration<M> {
@@ -50,9 +57,16 @@ where
         (self.check)(peer_id, &message)
     }
 
-    fn callback(&self, peer_id: PeerId, msg_id: &str, any: &Any) -> Result<()> {
-        let message = any.to_msg::<M>()?;
-        (self.callback)(peer_id, msg_id, message)
+    fn callback(
+        &self,
+        peer_id: PeerId,
+        msg_id: String,
+        any: Any,
+    ) -> BoxFuture<'static, Result<()>> {
+        match any.to_msg::<M>() {
+            Ok(message) => (self.callback)(peer_id, msg_id, message),
+            Err(e) => Box::pin(async move { Err(e.into()) }),
+        }
     }
 }
 
@@ -138,7 +152,7 @@ mod tests {
             .register_message::<prost_types::Timestamp>(
                 "timestamp",
                 Box::new(|_, _| Ok(())),
-                Box::new(|_, _, _| Ok(())),
+                Box::new(|_, _, _| Box::pin(async { Ok(()) })),
             )
             .await
             .unwrap();
@@ -147,7 +161,7 @@ mod tests {
             .register_message::<prost_types::Timestamp>(
                 "timestamp",
                 Box::new(|_, _| Ok(())),
-                Box::new(|_, _, _| Ok(())),
+                Box::new(|_, _, _| Box::pin(async { Ok(()) })),
             )
             .await
             .unwrap_err();

crates/dkg/src/bcast/error.rs

@@ -46,6 +46,25 @@ impl Failure {
     }
 }
 
+/// Peer IDs involved in an [`Error::InvalidSenderPeerIndex`] error.
+#[derive(Debug)]
+pub struct SenderPeerMismatch {
+    /// The peer ID of the actual sender.
+    pub sender: PeerId,
+    /// The peer ID expected at the claimed index.
+    pub expected: PeerId,
+}
+
+impl fmt::Display for SenderPeerMismatch {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "sender peer ID ({}) does not match claimed peer index {}",
+            self.sender, self.expected
+        )
+    }
+}
+
 /// User-facing reliable-broadcast error.
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
@@ -95,6 +114,15 @@ pub enum Error {
     #[error("invalid signature for peer {0}")]
     InvalidSignature(PeerId),
 
+    /// The peer index in the message is out of range or matches the local node.
+    #[error("invalid peer index: {0}")]
+    InvalidPeerIndex(PeerId),
+
+    /// The sender's peer index in the message does not match the sender's
+    /// actual index.
+    #[error("{0}")]
+    InvalidSenderPeerIndex(Box<SenderPeerMismatch>),
+
     /// The repeated hash for the same `(peer, msg_id)` differed.
     #[error("duplicate id with mismatching hash")]
     DuplicateMismatchingHash,
@@ -119,6 +147,10 @@ pub enum Error {
     #[error("receipt channel closed")]
     ReceiptChannelClosed,
 
+    /// The operation was cancelled.
+    #[error("cancelled")]
+    Cancelled,
+
     /// A required message body field was absent.
     #[error("missing protobuf field: {0}")]
     MissingField(&'static str),

crates/dkg/src/bcast/handler.rs

@@ -441,7 +441,7 @@ async fn handle_inbound_msg(
             .ok_or_else(|| Error::UnknownMessageId(message.id.clone()))?
     };
 
-    handler.callback(peer_id, &message.id, &any)?;
+    handler.callback(peer_id, message.id, any).await?;
     stream.close().await?;
     Ok(())
 }

crates/dkg/src/bcast/mod.rs

@@ -10,7 +10,7 @@ mod protocol;
 
 pub use behaviour::{Behaviour, Event};
 pub use component::{CallbackFn, CheckFn, Component};
-pub use error::{Error, Failure, Result};
+pub use error::{Error, Failure, Result, SenderPeerMismatch};
 
 /// The request-response protocol used to gather peer signatures.
 pub const SIG_PROTOCOL_NAME: StreamProtocol = StreamProtocol::new("/charon/dkg/bcast/1.0.0/sig");

crates/dkg/src/lib.rs

@@ -17,5 +17,8 @@ pub mod disk;
 /// Main DKG protocol implementation.
 pub mod dkg;
 
+/// Node signature exchange over the lock hash.
+pub mod nodesigs;
+
 /// Shares distributed to each node in the cluster.
 pub mod share;