feat(dkg): Implement bcast (#290)

* fix(proto): generated message implements Name

* fix(testutil): Public ConstReader

* feat(dkg): implement dkg/bcast

* feat(dkg): implement example bcast

* fix: add example instruction

* fix: clippy fmt

* fix: update bcast example ues relay

* refactor(p2p): make read/write protobuf shareable

* fix: address claude comments

* fix: update docs

* fix: remove unnecessary clone

* fix: simplify error

* fix: simplify code behaviour

* fix: fmt

* fix: move bcast new to behaviour

* fix: remove PROTOCOL_ID_PREFIX

* refactor: handler on_connection_event

* fix: using NotifyHandler::Any instead of One

* fix: only handle direct connection, not relay

* fix: enum error

* refactor: add p2p_context to behaviour

* feat: add Complete and Fail event for bcast

* fix: put local_peer_id to p2p_context

* chore: run fmt

* fix: use thiserror for Failure

* fix: local_peer_id is OnceLock

Author: iamquang95

Cargo.lock

@@ -9,6 +9,16 @@ pub struct PushRequest {
     #[prost(message, repeated, tag = "1")]
     pub streams: ::prost::alloc::vec::Vec<Stream>,
 }
+impl ::prost::Name for PushRequest {
+    const NAME: &'static str = "PushRequest";
+    const PACKAGE: &'static str = "loki";
+    fn full_name() -> ::prost::alloc::string::String {
+        "loki.PushRequest".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/loki.PushRequest".into()
+    }
+}
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Stream {
     #[prost(string, tag = "1")]
@@ -18,10 +28,30 @@ pub struct Stream {
     #[prost(uint64, tag = "3")]
     pub hash: u64,
 }
+impl ::prost::Name for Stream {
+    const NAME: &'static str = "Stream";
+    const PACKAGE: &'static str = "loki";
+    fn full_name() -> ::prost::alloc::string::String {
+        "loki.Stream".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/loki.Stream".into()
+    }
+}
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct Entry {
     #[prost(message, optional, tag = "1")]
     pub timestamp: ::core::option::Option<::prost_types::Timestamp>,
     #[prost(string, tag = "2")]
     pub line: ::prost::alloc::string::String,
+}
+impl ::prost::Name for Entry {
+    const NAME: &'static str = "Entry";
+    const PACKAGE: &'static str = "loki";
+    fn full_name() -> ::prost::alloc::string::String {
+        "loki.Entry".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/loki.Entry".into()
+    }
 }

crates/app/src/log/loki/lokipb/v1/loki.rs

@@ -18,7 +18,12 @@ pub fn compile_protos(proto_dir: &str) -> Result<()> {
     }
 
     let mut config = prost_build::Config::new();
-    config.btree_map(["."]).bytes(["."]).out_dir(proto_dir);
+    config
+        .btree_map(["."])
+        .bytes(["."])
+        .enable_type_names()
+        .type_name_domain(["."], "type.googleapis.com")
+        .out_dir(proto_dir);
 
     config.compile_protos(
         &proto_files.iter().map(|p| p.as_path()).collect::<Vec<_>>(),

crates/build-proto/src/lib.rs

@@ -23,6 +23,16 @@ pub struct QbftMsg {
     #[prost(bytes = "bytes", tag = "12")]
     pub prepared_value_hash: ::prost::bytes::Bytes,
 }
+impl ::prost::Name for QbftMsg {
+    const NAME: &'static str = "QBFTMsg";
+    const PACKAGE: &'static str = "consensus";
+    fn full_name() -> ::prost::alloc::string::String {
+        "consensus.QBFTMsg".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/consensus.QBFTMsg".into()
+    }
+}
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct QbftConsensusMsg {
     /// msg is the message that we send
@@ -35,6 +45,16 @@ pub struct QbftConsensusMsg {
     #[prost(message, repeated, tag = "3")]
     pub values: ::prost::alloc::vec::Vec<::prost_types::Any>,
 }
+impl ::prost::Name for QbftConsensusMsg {
+    const NAME: &'static str = "QBFTConsensusMsg";
+    const PACKAGE: &'static str = "consensus";
+    fn full_name() -> ::prost::alloc::string::String {
+        "consensus.QBFTConsensusMsg".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/consensus.QBFTConsensusMsg".into()
+    }
+}
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SniffedConsensusMsg {
     #[prost(message, optional, tag = "1")]
@@ -43,6 +63,16 @@ pub struct SniffedConsensusMsg {
     #[prost(message, optional, tag = "2")]
     pub msg: ::core::option::Option<QbftConsensusMsg>,
 }
+impl ::prost::Name for SniffedConsensusMsg {
+    const NAME: &'static str = "SniffedConsensusMsg";
+    const PACKAGE: &'static str = "consensus";
+    fn full_name() -> ::prost::alloc::string::String {
+        "consensus.SniffedConsensusMsg".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/consensus.SniffedConsensusMsg".into()
+    }
+}
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SniffedConsensusInstance {
     #[prost(message, optional, tag = "1")]
@@ -56,10 +86,30 @@ pub struct SniffedConsensusInstance {
     #[prost(string, tag = "5")]
     pub protocol_id: ::prost::alloc::string::String,
 }
+impl ::prost::Name for SniffedConsensusInstance {
+    const NAME: &'static str = "SniffedConsensusInstance";
+    const PACKAGE: &'static str = "consensus";
+    fn full_name() -> ::prost::alloc::string::String {
+        "consensus.SniffedConsensusInstance".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/consensus.SniffedConsensusInstance".into()
+    }
+}
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SniffedConsensusInstances {
     #[prost(message, repeated, tag = "1")]
     pub instances: ::prost::alloc::vec::Vec<SniffedConsensusInstance>,
     #[prost(string, tag = "2")]
     pub git_hash: ::prost::alloc::string::String,
+}
+impl ::prost::Name for SniffedConsensusInstances {
+    const NAME: &'static str = "SniffedConsensusInstances";
+    const PACKAGE: &'static str = "consensus";
+    fn full_name() -> ::prost::alloc::string::String {
+        "consensus.SniffedConsensusInstances".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/consensus.SniffedConsensusInstances".into()
+    }
 }

crates/core/src/corepb/v1/consensus.rs

@@ -14,6 +14,16 @@ pub struct Duty {
     #[prost(int32, tag = "2")]
     pub r#type: i32,
 }
+impl ::prost::Name for Duty {
+    const NAME: &'static str = "Duty";
+    const PACKAGE: &'static str = "core";
+    fn full_name() -> ::prost::alloc::string::String {
+        "core.Duty".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/core.Duty".into()
+    }
+}
 /// core.UnsignedDataSet
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct UnsignedDataSet {
@@ -24,6 +34,16 @@ pub struct UnsignedDataSet {
         ::prost::bytes::Bytes,
     >,
 }
+impl ::prost::Name for UnsignedDataSet {
+    const NAME: &'static str = "UnsignedDataSet";
+    const PACKAGE: &'static str = "core";
+    fn full_name() -> ::prost::alloc::string::String {
+        "core.UnsignedDataSet".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/core.UnsignedDataSet".into()
+    }
+}
 /// core.ParSignedDataSet
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ParSignedDataSet {
@@ -34,6 +54,16 @@ pub struct ParSignedDataSet {
         ParSignedData,
     >,
 }
+impl ::prost::Name for ParSignedDataSet {
+    const NAME: &'static str = "ParSignedDataSet";
+    const PACKAGE: &'static str = "core";
+    fn full_name() -> ::prost::alloc::string::String {
+        "core.ParSignedDataSet".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/core.ParSignedDataSet".into()
+    }
+}
 /// core.ParSignedData
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ParSignedData {
@@ -46,4 +76,14 @@ pub struct ParSignedData {
     /// int
     #[prost(int32, tag = "3")]
     pub share_idx: i32,
+}
+impl ::prost::Name for ParSignedData {
+    const NAME: &'static str = "ParSignedData";
+    const PACKAGE: &'static str = "core";
+    fn full_name() -> ::prost::alloc::string::String {
+        "core.ParSignedData".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/core.ParSignedData".into()
+    }
 }

crates/core/src/corepb/v1/core.rs

@@ -10,4 +10,14 @@ pub struct ParSigExMsg {
     pub duty: ::core::option::Option<super::core::Duty>,
     #[prost(message, optional, tag = "2")]
     pub data_set: ::core::option::Option<super::core::ParSignedDataSet>,
+}
+impl ::prost::Name for ParSigExMsg {
+    const NAME: &'static str = "ParSigExMsg";
+    const PACKAGE: &'static str = "parsigex";
+    fn full_name() -> ::prost::alloc::string::String {
+        "parsigex.ParSigExMsg".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/parsigex.ParSigExMsg".into()
+    }
 }

crates/core/src/corepb/v1/parsigex.rs

@@ -12,6 +12,16 @@ pub struct PriorityResult {
     #[prost(message, repeated, tag = "2")]
     pub topics: ::prost::alloc::vec::Vec<PriorityTopicResult>,
 }
+impl ::prost::Name for PriorityResult {
+    const NAME: &'static str = "PriorityResult";
+    const PACKAGE: &'static str = "priority";
+    fn full_name() -> ::prost::alloc::string::String {
+        "priority.PriorityResult".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/priority.PriorityResult".into()
+    }
+}
 /// PriorityMsg defines all the priorities and metadata of a single peer in the Prioritiser protocol.
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct PriorityMsg {
@@ -24,6 +34,16 @@ pub struct PriorityMsg {
     #[prost(bytes = "bytes", tag = "4")]
     pub signature: ::prost::bytes::Bytes,
 }
+impl ::prost::Name for PriorityMsg {
+    const NAME: &'static str = "PriorityMsg";
+    const PACKAGE: &'static str = "priority";
+    fn full_name() -> ::prost::alloc::string::String {
+        "priority.PriorityMsg".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/priority.PriorityMsg".into()
+    }
+}
 /// PriorityTopicProposal defines a single peers proposed priorities for a single topic in the Prioritiser protocol.
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct PriorityTopicProposal {
@@ -32,6 +52,16 @@ pub struct PriorityTopicProposal {
     #[prost(message, repeated, tag = "2")]
     pub priorities: ::prost::alloc::vec::Vec<::prost_types::Any>,
 }
+impl ::prost::Name for PriorityTopicProposal {
+    const NAME: &'static str = "PriorityTopicProposal";
+    const PACKAGE: &'static str = "priority";
+    fn full_name() -> ::prost::alloc::string::String {
+        "priority.PriorityTopicProposal".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/priority.PriorityTopicProposal".into()
+    }
+}
 /// PriorityTopicResult defines the cluster wide resulting priorities for a
 /// single topic in the Prioritiser protocol.
 #[derive(Clone, PartialEq, ::prost::Message)]
@@ -42,11 +72,31 @@ pub struct PriorityTopicResult {
     #[prost(message, repeated, tag = "2")]
     pub priorities: ::prost::alloc::vec::Vec<PriorityScoredResult>,
 }
+impl ::prost::Name for PriorityTopicResult {
+    const NAME: &'static str = "PriorityTopicResult";
+    const PACKAGE: &'static str = "priority";
+    fn full_name() -> ::prost::alloc::string::String {
+        "priority.PriorityTopicResult".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/priority.PriorityTopicResult".into()
+    }
+}
 /// PriorityScoredResult defines the cluster wide priority score in the Prioritiser protocol.
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct PriorityScoredResult {
     #[prost(message, optional, tag = "1")]
     pub priority: ::core::option::Option<::prost_types::Any>,
     #[prost(int64, tag = "2")]
     pub score: i64,
+}
+impl ::prost::Name for PriorityScoredResult {
+    const NAME: &'static str = "PriorityScoredResult";
+    const PACKAGE: &'static str = "priority";
+    fn full_name() -> ::prost::alloc::string::String {
+        "priority.PriorityScoredResult".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "type.googleapis.com/priority.PriorityScoredResult".into()
+    }
 }

crates/core/src/corepb/v1/priority.rs

@@ -9,6 +9,16 @@ publish.workspace = true
 [dependencies]
 prost.workspace = true
 prost-types.workspace = true
+thiserror.workspace = true
+libp2p.workspace = true
+futures.workspace = true
+tokio.workspace = true
+sha2.workspace = true
+tracing.workspace = true
+either.workspace = true
+k256.workspace = true
+pluto-k1util.workspace = true
+pluto-p2p.workspace = true
 pluto-cluster.workspace = true
 pluto-crypto.workspace = true
 pluto-eth1wrap.workspace = true
@@ -17,17 +27,22 @@ hex.workspace = true
 rand.workspace = true
 serde.workspace = true
 serde_json.workspace = true
-thiserror.workspace = true
-tokio.workspace = true
-tracing.workspace = true
 url.workspace = true
 zeroize.workspace = true
 
 [build-dependencies]
 pluto-build-proto.workspace = true
 
 [dev-dependencies]
+anyhow.workspace = true
+clap.workspace = true
+hex.workspace = true
 pluto-cluster = { workspace = true, features = ["test-cluster"] }
+pluto-core.workspace = true
+pluto-testutil.workspace = true
+pluto-tracing.workspace = true
+serde_json.workspace = true
+tokio-util.workspace = true
 tempfile.workspace = true
 
 [lints]