Implement randomness generation protocol in Narwhal. (MystenLabs#14435)

## Description Implements on-chain randomness generation in narwhal using the DKG added in an earlier PR. ## Test Plan Manual testing in private testnet.
a00012025 · Nov 7, 2023 · 429cb9a · 429cb9a
1 parent 2ef6417
commit 429cb9a
Show file tree

Hide file tree

Showing 16 changed files with 826 additions and 293 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -521,9 +521,9 @@ move-stackless-bytecode = { path = "external-crates/move/crates/move-stackless-b
 move-symbol-pool = { path = "external-crates/move/crates/move-symbol-pool" }
 move-abstract-stack = { path = "external-crates/move/crates/move-abstract-stack" }
 
-fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "802c1ac98061687d6ce024849c747a250dbeea52" }
-fastcrypto-tbls = { git = "https://github.com/MystenLabs/fastcrypto", rev = "802c1ac98061687d6ce024849c747a250dbeea52"}
-fastcrypto-zkp = { git = "https://github.com/MystenLabs/fastcrypto", rev = "802c1ac98061687d6ce024849c747a250dbeea52", package = "fastcrypto-zkp" }
+fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "517ec93ad2eba1807c9508f14953814a33523edc" }
+fastcrypto-tbls = { git = "https://github.com/MystenLabs/fastcrypto", rev = "517ec93ad2eba1807c9508f14953814a33523edc"}
+fastcrypto-zkp = { git = "https://github.com/MystenLabs/fastcrypto", rev = "517ec93ad2eba1807c9508f14953814a33523edc", package = "fastcrypto-zkp" }
 
 # anemo dependencies
 anemo = { git = "https://github.com/mystenlabs/anemo.git", rev = "1169850e6af127397068cd86764c29b1d49dbe35" }

diff --git a/crates/workspace-hack/Cargo.toml b/crates/workspace-hack/Cargo.toml
@@ -261,9 +261,9 @@ expect-test = { version = "1", default-features = false }
 eyre = { version = "0.6" }
 fail = { version = "0.4", default-features = false }
 fast_chemail = { version = "0.9", default-features = false }
-fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "802c1ac98061687d6ce024849c747a250dbeea52", features = ["beacon-dkg", "copy_key"] }
-fastcrypto-tbls = { git = "https://github.com/MystenLabs/fastcrypto", rev = "802c1ac98061687d6ce024849c747a250dbeea52" }
-fastcrypto-zkp = { git = "https://github.com/MystenLabs/fastcrypto", rev = "802c1ac98061687d6ce024849c747a250dbeea52", default-features = false }
+fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "517ec93ad2eba1807c9508f14953814a33523edc", features = ["beacon-dkg", "copy_key"] }
+fastcrypto-tbls = { git = "https://github.com/MystenLabs/fastcrypto", rev = "517ec93ad2eba1807c9508f14953814a33523edc" }
+fastcrypto-zkp = { git = "https://github.com/MystenLabs/fastcrypto", rev = "517ec93ad2eba1807c9508f14953814a33523edc", default-features = false }
 fastrand-dff4ba8e3ae991db = { package = "fastrand", version = "1", default-features = false }
 fastrand-f595c2ba2a3f28df = { package = "fastrand", version = "2" }
 fd-lock = { version = "3", default-features = false }
@@ -1064,10 +1064,10 @@ expect-test = { version = "1", default-features = false }
 eyre = { version = "0.6" }
 fail = { version = "0.4", default-features = false }
 fast_chemail = { version = "0.9", default-features = false }
-fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "802c1ac98061687d6ce024849c747a250dbeea52", features = ["beacon-dkg", "copy_key"] }
-fastcrypto-derive = { git = "https://github.com/MystenLabs/fastcrypto", rev = "802c1ac98061687d6ce024849c747a250dbeea52", default-features = false }
-fastcrypto-tbls = { git = "https://github.com/MystenLabs/fastcrypto", rev = "802c1ac98061687d6ce024849c747a250dbeea52" }
-fastcrypto-zkp = { git = "https://github.com/MystenLabs/fastcrypto", rev = "802c1ac98061687d6ce024849c747a250dbeea52", default-features = false }
+fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "517ec93ad2eba1807c9508f14953814a33523edc", features = ["beacon-dkg", "copy_key"] }
+fastcrypto-derive = { git = "https://github.com/MystenLabs/fastcrypto", rev = "517ec93ad2eba1807c9508f14953814a33523edc", default-features = false }
+fastcrypto-tbls = { git = "https://github.com/MystenLabs/fastcrypto", rev = "517ec93ad2eba1807c9508f14953814a33523edc" }
+fastcrypto-zkp = { git = "https://github.com/MystenLabs/fastcrypto", rev = "517ec93ad2eba1807c9508f14953814a33523edc", default-features = false }
 fastrand-dff4ba8e3ae991db = { package = "fastrand", version = "1", default-features = false }
 fastrand-f595c2ba2a3f28df = { package = "fastrand", version = "2" }
 fd-lock = { version = "3", default-features = false }

diff --git a/narwhal/crypto/src/lib.rs b/narwhal/crypto/src/lib.rs
@@ -43,6 +43,9 @@ pub type KeyPair = bls12381::min_sig::BLS12381KeyPair;
 
 pub type NetworkPublicKey = ed25519::Ed25519PublicKey;
 pub type NetworkKeyPair = ed25519::Ed25519KeyPair;
+
+pub type RandomnessSignature = fastcrypto_tbls::types::Signature;
+pub type RandomnessPartialSignature = fastcrypto_tbls::tbls::PartialSignature<RandomnessSignature>;
 pub type RandomnessPrivateKey =
     fastcrypto_tbls::ecies::PrivateKey<fastcrypto::groups::bls12381::G2Element>;
 

diff --git a/narwhal/primary/Cargo.toml b/narwhal/primary/Cargo.toml
@@ -23,6 +23,7 @@ prometheus.workspace = true
 rand = { workspace = true, features = ["small_rng"] }
 thiserror.workspace = true
 tokio = { workspace = true, features = ["sync", "rt", "macros", "time", "test-util"] }
+tokio-stream.workspace = true
 tower.workspace = true
 tracing.workspace = true
 tap.workspace = true

diff --git a/narwhal/primary/src/metrics.rs b/narwhal/primary/src/metrics.rs
@@ -69,6 +69,8 @@ pub struct PrimaryChannelMetrics {
     pub tx_new_certificates: IntGauge,
     /// occupancy of the channel signaling own committed headers
     pub tx_committed_own_headers: IntGauge,
+    /// occupancy of the channel from the `primary::PrimaryReceiverHandler` to the `primary::StateHandler`
+    pub tx_randomness_partial_signatures: IntGauge,
     /// An internal synchronizer channel. Occupancy of the channel sending certificates to the internal
     /// task that accepts certificates.
     pub tx_certificate_acceptor: IntGauge,
@@ -96,6 +98,8 @@ pub struct PrimaryChannelMetrics {
     pub tx_new_certificates_total: IntCounter,
     /// total received on the channel signaling own committed headers
     pub tx_committed_own_headers_total: IntCounter,
+    /// total received on the channel from the `primary::PrimaryReceiverHandler` to the `primary::StateHandler`
+    pub tx_randomness_partial_signatures_total: IntCounter,
     /// Total received by the channel sending certificates to the internal task that accepts certificates.
     pub tx_certificate_acceptor_total: IntCounter,
     /// Total received the channel to synchronize missing batches
@@ -172,6 +176,11 @@ impl PrimaryChannelMetrics {
                 "occupancy of the channel signaling own committed headers.",
                 registry
             ).unwrap(),
+            tx_randomness_partial_signatures: register_int_gauge_with_registry!(
+                "tx_randomness_partial_signatures",
+                "occupancy of the channel from the `primary::PrimaryReceiverHandler` to the `primary::StateHandler`",
+                registry
+            ).unwrap(),
             tx_certificate_acceptor: register_int_gauge_with_registry!(
                 "tx_certificate_acceptor",
                 "occupancy of the internal synchronizer channel that is accepting new certificates.",
@@ -234,6 +243,11 @@ impl PrimaryChannelMetrics {
                 "total received on channel signaling own committed headers.",
                 registry
             ).unwrap(),
+            tx_randomness_partial_signatures_total: register_int_counter_with_registry!(
+                "tx_randomness_partial_signatures_total",
+                "total received on the channel from the `primary::PrimaryReceiverHandler` to the `primary::StateHandler`",
+                registry
+            ).unwrap(),
             tx_certificate_acceptor_total: register_int_counter_with_registry!(
                 "tx_certificate_acceptor_total",
                 "total received on the internal synchronizer channel that is accepting new certificates.",

diff --git a/narwhal/primary/src/primary.rs b/narwhal/primary/src/primary.rs
@@ -28,15 +28,21 @@ use anemo_tower::{
 };
 use async_trait::async_trait;
 use config::{Authority, AuthorityIdentifier, ChainIdentifier, Committee, Parameters, WorkerCache};
-use crypto::{traits::EncodeDecodeBase64, RandomnessPrivateKey};
+use crypto::{
+    traits::EncodeDecodeBase64, RandomnessPartialSignature, RandomnessPrivateKey,
+    RandomnessSignature,
+};
 use crypto::{KeyPair, NetworkKeyPair, NetworkPublicKey, Signature};
 use fastcrypto::{
     hash::Hash,
     serde_helpers::ToFromByteArray,
     signature_service::SignatureService,
     traits::{KeyPair as _, ToFromBytes},
 };
-use fastcrypto_tbls::{tbls::ThresholdBls, types::ThresholdBls12381MinSig};
+use fastcrypto_tbls::{
+    tbls::ThresholdBls,
+    types::{PublicVssKey, ThresholdBls12381MinSig},
+};
 use mysten_metrics::metered_channel::{channel_with_total, Receiver, Sender};
 use mysten_metrics::monitored_scope;
 use mysten_network::{multiaddr::Protocol, Multiaddr};
@@ -47,7 +53,6 @@ use network::{
 use network::{failpoints::FailpointsMakeCallbackHandler, metrics::MetricsMakeCallbackHandler};
 use parking_lot::Mutex;
 use prometheus::Registry;
-use std::collections::{btree_map::Entry, BTreeMap, HashMap};
 use std::{
     cmp::Reverse,
     collections::{BTreeSet, BinaryHeap},
@@ -56,6 +61,10 @@ use std::{
     thread::sleep,
     time::Duration,
 };
+use std::{
+    collections::{btree_map::Entry, BTreeMap, HashMap},
+    sync::OnceLock,
+};
 use storage::{CertificateStore, PayloadStore, ProposerStore, VoteDigestStore};
 use sui_protocol_config::ProtocolConfig;
 use tokio::{sync::oneshot, time::Instant};
@@ -67,9 +76,10 @@ use types::{
     error::{DagError, DagResult},
     now, validate_received_certificate_version, Certificate, CertificateAPI, CertificateDigest,
     FetchCertificatesRequest, FetchCertificatesResponse, Header, HeaderAPI, MetadataAPI,
-    PreSubscribedBroadcastSender, PrimaryToPrimary, PrimaryToPrimaryServer, RequestVoteRequest,
-    RequestVoteResponse, Round, SendCertificateRequest, SendCertificateResponse, SystemMessage,
-    Vote, VoteInfoAPI, WorkerOthersBatchMessage, WorkerOwnBatchMessage, WorkerToPrimary,
+    PreSubscribedBroadcastSender, PrimaryToPrimary, PrimaryToPrimaryServer, RandomnessRound,
+    RequestVoteRequest, RequestVoteResponse, Round, SendCertificateRequest,
+    SendCertificateResponse, SendRandomnessPartialSignaturesRequest, SystemMessage, Vote,
+    VoteInfoAPI, WorkerOthersBatchMessage, WorkerOwnBatchMessage, WorkerToPrimary,
     WorkerToPrimaryServer,
 };
 
@@ -162,6 +172,14 @@ impl Primary {
             &primary_channel_metrics.tx_committed_own_headers,
             &primary_channel_metrics.tx_committed_own_headers_total,
         );
+        let (tx_randomness_partial_signatures, rx_randomness_partial_signatures) =
+            channel_with_total(
+                CHANNEL_CAPACITY,
+                &primary_channel_metrics.tx_randomness_partial_signatures,
+                &primary_channel_metrics.tx_randomness_partial_signatures_total,
+            );
+
+        let randomness_vss_key_lock = Arc::new(OnceLock::new());
 
         // we need to hack the gauge from this consensus channel into the primary registry
         // This avoids a cyclic dependency in the initialization of consensus and primary
@@ -220,7 +238,9 @@ impl Primary {
             signature_service: signature_service.clone(),
             certificate_store: certificate_store.clone(),
             vote_digest_store,
-            rx_narwhal_round_updates,
+            rx_narwhal_round_updates: rx_narwhal_round_updates.clone(),
+            randomness_vss_key_lock: randomness_vss_key_lock.clone(),
+            tx_randomness_partial_signatures,
             parent_digests: Default::default(),
             metrics: node_metrics.clone(),
         })
@@ -495,7 +515,7 @@ impl Primary {
             tx_narwhal_round_updates,
             rx_committed_own_headers,
             node_metrics,
-            leader_schedule,
+            leader_schedule.clone(),
         );
 
         let mut handles = vec![
@@ -513,10 +533,14 @@ impl Primary {
             authority.id(),
             committee,
             rx_committed_certificates,
+            rx_randomness_partial_signatures,
             tx_shutdown.subscribe(),
+            rx_narwhal_round_updates,
             Some(tx_committed_own_headers),
+            randomness_vss_key_lock.clone(),
             tx_system_messages,
             RandomnessPrivateKey::from(randomness_private_key),
+            leader_schedule,
             network,
         );
         handles.push(state_handler_handle);
@@ -565,6 +589,14 @@ struct PrimaryReceiverHandler {
     vote_digest_store: VoteDigestStore,
     /// Get a signal when the round changes.
     rx_narwhal_round_updates: watch::Receiver<Round>,
+    /// Stores the randomness VSS public key when available.
+    randomness_vss_key_lock: Arc<OnceLock<PublicVssKey>>,
+    /// Sends randomness partial signatures to the state handler.
+    tx_randomness_partial_signatures: Sender<(
+        AuthorityIdentifier,
+        RandomnessRound,
+        Vec<RandomnessPartialSignature>,
+    )>,
     /// Known parent digests that are being fetched from header proposers.
     /// Values are where the digests are first known from.
     /// TODO: consider limiting maximum number of digests from one authority, allow timeout
@@ -573,6 +605,27 @@ struct PrimaryReceiverHandler {
     metrics: Arc<PrimaryMetrics>,
 }
 
+fn authority_for_request<'a, T>(
+    committee: &'a Committee,
+    request: &anemo::Request<T>,
+) -> DagResult<&'a Authority> {
+    let peer_id = request
+        .peer_id()
+        .ok_or_else(|| DagError::NetworkError("Unable to access remote peer ID".to_owned()))?;
+    let peer_network_key = NetworkPublicKey::from_bytes(&peer_id.0).map_err(|e| {
+        DagError::NetworkError(format!(
+            "Unable to interpret remote peer ID {peer_id:?} as a NetworkPublicKey: {e:?}"
+        ))
+    })?;
+    committee
+        .authority_by_network_key(&peer_network_key)
+        .ok_or_else(|| {
+            DagError::NetworkError(format!(
+                "Unable to find authority with network key {peer_network_key:?}"
+            ))
+        })
+}
+
 #[allow(clippy::result_large_err)]
 impl PrimaryReceiverHandler {
     fn find_next_round(
@@ -614,21 +667,7 @@ impl PrimaryReceiverHandler {
             .inc_by(num_parents as u64);
 
         // Vote request must come from the Header's author.
-        let peer_id = request
-            .peer_id()
-            .ok_or_else(|| DagError::NetworkError("Unable to access remote peer ID".to_owned()))?;
-        let peer_network_key = NetworkPublicKey::from_bytes(&peer_id.0).map_err(|e| {
-            DagError::NetworkError(format!(
-                "Unable to interpret remote peer ID {peer_id:?} as a NetworkPublicKey: {e:?}"
-            ))
-        })?;
-        let peer_authority = committee
-            .authority_by_network_key(&peer_network_key)
-            .ok_or_else(|| {
-                DagError::NetworkError(format!(
-                    "Unable to find authority with network key {peer_network_key:?}"
-                ))
-            })?;
+        let peer_authority = authority_for_request(&self.committee, &request)?;
         ensure!(
             header.author() == peer_authority.id(),
             DagError::NetworkError(format!(
@@ -735,6 +774,19 @@ impl PrimaryReceiverHandler {
                         DagError::InvalidSystemMessage
                     );
                 }
+                SystemMessage::RandomnessSignature(round, bytes) => {
+                    let sig: RandomnessSignature =
+                        bcs::from_bytes(bytes).map_err(|_| DagError::InvalidSystemMessage)?;
+                    fastcrypto_tbls::types::ThresholdBls12381MinSig::verify(
+                        self.randomness_vss_key_lock
+                            .get()
+                            .ok_or(DagError::RandomnessUnavailable)?
+                            .c0(),
+                        &round.signature_message(),
+                        &sig,
+                    )
+                    .map_err(|_| DagError::InvalidRandomnessSignature)?;
+                }
             }
         }
 
@@ -947,6 +999,27 @@ impl PrimaryToPrimary for PrimaryReceiverHandler {
         }
     }
 
+    async fn send_randomness_partial_signatures(
+        &self,
+        request: anemo::Request<SendRandomnessPartialSignaturesRequest>,
+    ) -> Result<anemo::Response<()>, anemo::rpc::Status> {
+        let _scope = monitored_scope("PrimaryReceiverHandler::send_randomness_partial_signatures");
+        let peer_authority = authority_for_request(&self.committee, &request).map_err(|e| {
+            anemo::rpc::Status::new_with_message(
+                anemo::types::response::StatusCode::Unknown,
+                format!("{e:?}"),
+            )
+        })?;
+        let request = request.into_body();
+        // This is best-effort, eat any errors in processing.
+        // TODO: consider returning an error to the sender on signature verification failure.
+        let _ = self
+            .tx_randomness_partial_signatures
+            .send((peer_authority.id(), request.round, request.sigs))
+            .await;
+        Ok(anemo::Response::new(()))
+    }
+
     async fn request_vote(
         &self,
         request: anemo::Request<RequestVoteRequest>,
@@ -961,6 +1034,7 @@ impl PrimaryToPrimary for PrimaryReceiverHandler {
                         DagError::InvalidSignature
                         | DagError::InvalidEpoch { .. }
                         | DagError::InvalidHeaderDigest
+                        | DagError::InvalidRandomnessSignature
                         | DagError::HeaderHasBadWorkerIds(_)
                         | DagError::HeaderHasInvalidParentRoundNumbers(_)
                         | DagError::HeaderHasDuplicateParentAuthorities(_)