Avoid PublicKey deserialization in ConsensusStore::read_last_committed (

MystenLabs#9413)

## Description 

Calls to ConsensusStore::read_last_committed resulted in an expensive
deserialization of the public keys used as keys for the last_committed
map. This PR replaces this usage with PublicKeyBytes instead to avoid
that.

## Test Plan 

Unit tests.

### Type of Change (Check all that apply)

- [ ] user-visible impact
- [ ] breaking change for a client SDKs
- [ ] breaking change for FNs (FN binary must upgrade)
- [X] breaking change for validators or node operators (must upgrade
binaries)
- [ ] breaking change for on-chain data layout
- [X] necessitate either a data wipe or data migration

Cargo.toml

@@ -130,8 +130,8 @@ move-prover-boogie-backend = { git = "https://github.com/move-language/move", re
 move-stackless-bytecode = { git = "https://github.com/move-language/move", rev = "f3cab72c7b7401de34a2d4c4ac86f9e402256e25" }
 move-symbol-pool = { git = "https://github.com/move-language/move", rev = "f3cab72c7b7401de34a2d4c4ac86f9e402256e25" }
 
-fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "d96eee87465f02976eeb1ae3ceeba57b7d8bfe4b" }
-fastcrypto-zkp = { git = "https://github.com/MystenLabs/fastcrypto", rev = "d96eee87465f02976eeb1ae3ceeba57b7d8bfe4b", package = "fastcrypto-zkp" }
+fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "af2d40caa3db127e60182591b8c7decfabea399c" }
+fastcrypto-zkp = { git = "https://github.com/MystenLabs/fastcrypto", rev = "af2d40caa3db127e60182591b8c7decfabea399c", package = "fastcrypto-zkp" }
 
 # anemo dependencies
 anemo = { git = "https://github.com/mystenlabs/anemo.git", rev = "4ebf4a86952827ff0fcce6a2d8a80f42f34efed9" }

crates/mysten-util-mem/src/external_impls.rs

@@ -8,6 +8,7 @@ malloc_size_of_is_0!(ed25519_consensus::Signature);
 
 // fastcrypto
 malloc_size_of_is_0!(fastcrypto::bls12381::min_sig::BLS12381PublicKey);
+malloc_size_of_is_0!(fastcrypto::bls12381::min_sig::BLS12381PublicKeyAsBytes);
 malloc_size_of_is_0!(fastcrypto::bls12381::min_sig::BLS12381Signature);
 malloc_size_of_is_0!(fastcrypto::bls12381::min_sig::BLS12381AggregateSignature);
 malloc_size_of_is_0!(fastcrypto::bls12381::min_pk::BLS12381PublicKey);

crates/workspace-hack/Cargo.toml

@@ -195,8 +195,8 @@ expect-test = { version = "1", default-features = false }
 eyre = { version = "0.6" }
 fail-9fbad63c4bcf4a8f = { package = "fail", version = "0.4", default-features = false }
 fail-d8f496e17d97b5cb = { package = "fail", version = "0.5", default-features = false }
-fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "d96eee87465f02976eeb1ae3ceeba57b7d8bfe4b", features = ["copy_key"] }
-fastcrypto-zkp = { git = "https://github.com/MystenLabs/fastcrypto", rev = "d96eee87465f02976eeb1ae3ceeba57b7d8bfe4b", default-features = false }
+fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "af2d40caa3db127e60182591b8c7decfabea399c", features = ["copy_key"] }
+fastcrypto-zkp = { git = "https://github.com/MystenLabs/fastcrypto", rev = "af2d40caa3db127e60182591b8c7decfabea399c", default-features = false }
 fastrand = { version = "1", default-features = false }
 fd-lock = { version = "3", default-features = false }
 fdlimit = { version = "0.2", default-features = false }
@@ -848,9 +848,9 @@ expect-test = { version = "1", default-features = false }
 eyre = { version = "0.6" }
 fail-9fbad63c4bcf4a8f = { package = "fail", version = "0.4", default-features = false }
 fail-d8f496e17d97b5cb = { package = "fail", version = "0.5", default-features = false }
-fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "d96eee87465f02976eeb1ae3ceeba57b7d8bfe4b", features = ["copy_key"] }
-fastcrypto-derive = { git = "https://github.com/MystenLabs/fastcrypto", rev = "d96eee87465f02976eeb1ae3ceeba57b7d8bfe4b", default-features = false }
-fastcrypto-zkp = { git = "https://github.com/MystenLabs/fastcrypto", rev = "d96eee87465f02976eeb1ae3ceeba57b7d8bfe4b", default-features = false }
+fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "af2d40caa3db127e60182591b8c7decfabea399c", features = ["copy_key"] }
+fastcrypto-derive = { git = "https://github.com/MystenLabs/fastcrypto", rev = "af2d40caa3db127e60182591b8c7decfabea399c", default-features = false }
+fastcrypto-zkp = { git = "https://github.com/MystenLabs/fastcrypto", rev = "af2d40caa3db127e60182591b8c7decfabea399c", default-features = false }
 fastrand = { version = "1", default-features = false }
 fd-lock = { version = "3", default-features = false }
 fdlimit = { version = "0.2", default-features = false }

narwhal/consensus/src/bullshark.rs

@@ -9,7 +9,6 @@ use crate::{
 use config::{Committee, Stake};
 use crypto::PublicKey;
 use fastcrypto::hash::Hash;
-use fastcrypto::traits::EncodeDecodeBase64;
 use std::sync::Arc;
 use tokio::time::Instant;
 use tracing::{debug, error_span};
@@ -225,7 +224,7 @@ impl ConsensusProtocol for Bullshark {
         // Performance note: if tracing at the debug log level is disabled, this is cheap, see
         // https://github.com/tokio-rs/tracing/pull/326
         for (name, round) in &state.last_committed {
-            debug!("Latest commit of {}: Round {}", name.encode_base64(), round);
+            debug!("Latest commit of {}: Round {}", name.to_string(), round);
         }
 
         self.metrics

narwhal/consensus/src/consensus.rs

@@ -7,7 +7,7 @@
 use crate::utils::gc_round;
 use crate::{metrics::ConsensusMetrics, ConsensusError, Outcome, SequenceNumber};
 use config::Committee;
-use crypto::PublicKey;
+use crypto::{PublicKey, PublicKeyBytes};
 use fastcrypto::hash::Hash;
 use mysten_metrics::spawn_logged_monitored_task;
 use std::{
@@ -38,7 +38,7 @@ pub struct ConsensusState {
     pub gc_depth: Round,
     /// Keeps the last committed round for each authority. This map is used to clean up the dag and
     /// ensure we don't commit twice the same certificate.
-    pub last_committed: HashMap<PublicKey, Round>,
+    pub last_committed: HashMap<PublicKeyBytes, Round>,
     /// Used to populate the index in the sub-dag construction.
     pub latest_sub_dag_index: SequenceNumber,
     /// The last calculated consensus reputation score
@@ -71,7 +71,7 @@ impl ConsensusState {
         metrics: Arc<ConsensusMetrics>,
         last_committed_round: Round,
         gc_depth: Round,
-        recovered_last_committed: HashMap<PublicKey, Round>,
+        recovered_last_committed: HashMap<PublicKeyBytes, Round>,
         latest_sub_dag: Option<CommittedSubDagShell>,
         cert_store: CertificateStore,
         committee: &Committee,
@@ -106,7 +106,7 @@ impl ConsensusState {
     #[instrument(level = "info", skip_all)]
     pub fn construct_dag_from_cert_store(
         cert_store: CertificateStore,
-        last_committed: &HashMap<PublicKey, Round>,
+        last_committed: &HashMap<PublicKeyBytes, Round>,
         gc_round: Round,
     ) -> Result<Dag, ConsensusError> {
         let mut dag: Dag = BTreeMap::new();
@@ -145,7 +145,7 @@ impl ConsensusState {
     /// Returns true if certificate is inserted in the dag.
     fn try_insert_in_dag(
         dag: &mut Dag,
-        last_committed: &HashMap<PublicKey, Round>,
+        last_committed: &HashMap<PublicKeyBytes, Round>,
         gc_round: Round,
         certificate: &Certificate,
     ) -> Result<bool, ConsensusError> {
@@ -175,15 +175,15 @@ impl ConsensusState {
 
         Ok(certificate.round()
             > last_committed
-                .get(&certificate.origin())
+                .get(&PublicKeyBytes::from(&certificate.origin()))
                 .cloned()
                 .unwrap_or_default())
     }
 
     /// Update and clean up internal state after committing a certificate.
     pub fn update(&mut self, certificate: &Certificate) {
         self.last_committed
-            .entry(certificate.origin())
+            .entry(PublicKeyBytes::from(&certificate.origin()))
             .and_modify(|r| *r = max(*r, certificate.round()))
             .or_insert_with(|| certificate.round());
         self.last_round = self.last_round.update(certificate.round(), self.gc_depth);

narwhal/consensus/src/tests/consensus_tests.rs

@@ -17,6 +17,7 @@ use crate::consensus::ConsensusRound;
 use crate::metrics::ConsensusMetrics;
 use crate::Consensus;
 use crate::NUM_SHUTDOWN_RECEIVERS;
+use crypto::PublicKeyBytes;
 use types::{Certificate, PreSubscribedBroadcastSender, ReputationScores};
 
 /// This test is trying to compare the output of the Consensus algorithm when:
@@ -131,7 +132,7 @@ async fn test_consensus_recovery_with_bullshark() {
     let last_committed = consensus_store.read_last_committed();
 
     for key in keys.clone() {
-        let last_round = *last_committed.get(&key).unwrap();
+        let last_round = *last_committed.get(&PublicKeyBytes::from(&key)).unwrap();
 
         // For the leader of round 6 we expect to have last committed round of 6.
         if key == Bullshark::leader_authority(&committee, 6) {

narwhal/consensus/src/tests/consensus_utils.rs

@@ -1,6 +1,6 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
-use crypto::{PublicKey, PublicKeyBytes};
+use crypto::PublicKeyBytes;
 use std::sync::Arc;
 use storage::CertificateStore;
 use store::rocks::MetricConf;
@@ -22,7 +22,7 @@ pub fn make_consensus_store(store_path: &std::path::Path) -> Arc<ConsensusStore>
     .expect("Failed to create database");
 
     let (last_committed_map, sequence_map) = reopen!(&rocksdb,
-        LAST_COMMITTED_CF;<PublicKey, Round>,
+        LAST_COMMITTED_CF;<PublicKeyBytes, Round>,
         SEQUENCE_CF;<SequenceNumber, CommittedSubDagShell>
     );
 

narwhal/consensus/src/tusk.rs

@@ -6,7 +6,7 @@ use crate::{
     utils, ConsensusError, Outcome,
 };
 use config::{Committee, Stake};
-use fastcrypto::{hash::Hash, traits::EncodeDecodeBase64};
+use fastcrypto::hash::Hash;
 use std::{collections::HashMap, sync::Arc};
 use tracing::{debug, error_span};
 use types::{
@@ -125,7 +125,7 @@ impl ConsensusProtocol for Tusk {
         // Performance note: if tracing at the debug log level is disabled, this is cheap, see
         // https://github.com/tokio-rs/tracing/pull/326
         for (name, round) in &state.last_committed {
-            debug!("Latest commit of {}: Round {}", name.encode_base64(), round);
+            debug!("Latest commit of {}: Round {}", name.to_string(), round);
         }
 
         Ok((Outcome::Commit, committed_sub_dags))

narwhal/consensus/src/utils.rs

@@ -3,6 +3,7 @@
 // SPDX-License-Identifier: Apache-2.0
 use crate::consensus::{ConsensusState, Dag};
 use config::Committee;
+use crypto::PublicKeyBytes;
 use std::collections::HashSet;
 use tracing::debug;
 use types::{Certificate, CertificateDigest, Round};
@@ -87,7 +88,7 @@ pub fn order_dag(leader: &Certificate, state: &ConsensusState) -> Vec<Certificat
             let mut skip = already_ordered.contains(&digest);
             skip |= state
                 .last_committed
-                .get(&certificate.origin())
+                .get(&PublicKeyBytes::from(&certificate.origin()))
                 .map_or_else(|| false, |r| &certificate.round() <= r);
             if !skip {
                 buffer.push(certificate);

narwhal/storage/src/node_store.rs

@@ -88,7 +88,7 @@ impl NodeStorage {
             Self::CERTIFICATE_DIGEST_BY_ORIGIN_CF;<(PublicKeyBytes, Round), CertificateDigest>,
             Self::PAYLOAD_CF;<(BatchDigest, WorkerId), PayloadToken>,
             Self::BATCHES_CF;<BatchDigest, Batch>,
-            Self::LAST_COMMITTED_CF;<PublicKey, Round>,
+            Self::LAST_COMMITTED_CF;<PublicKeyBytes, Round>,
             Self::SUB_DAG_INDEX_CF;<SequenceNumber, CommittedSubDagShell>
         );
 

narwhal/test-utils/src/lib.rs

@@ -9,7 +9,7 @@ use config::{
 };
 use crypto::{
     to_intent_message, KeyPair, NarwhalAuthoritySignature, NetworkKeyPair, NetworkPublicKey,
-    PublicKey, Signature,
+    PublicKey, PublicKeyBytes, Signature,
 };
 use fastcrypto::{
     hash::Hash as _,
@@ -139,7 +139,7 @@ pub fn make_consensus_store(store_path: &std::path::Path) -> Arc<ConsensusStore>
     .expect("Failed creating database");
 
     let (last_committed_map, sequence_map) = reopen!(&rocksdb,
-        LAST_COMMITTED_CF;<PublicKey, Round>,
+        LAST_COMMITTED_CF;<PublicKeyBytes, Round>,
         SEQUENCE_CF;<SequenceNumber, CommittedSubDagShell>
     );
 

narwhal/types/src/consensus.rs

@@ -4,7 +4,7 @@
 
 use crate::{Batch, Certificate, CertificateDigest, Round};
 use config::Committee;
-use crypto::PublicKey;
+use crypto::{PublicKey, PublicKeyBytes};
 use fastcrypto::hash::Hash;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
@@ -144,15 +144,15 @@ pub type StoreResult<T> = Result<T, TypedStoreError>;
 /// The persistent storage of the sequencer.
 pub struct ConsensusStore {
     /// The latest committed round of each validator.
-    last_committed: DBMap<PublicKey, Round>,
+    last_committed: DBMap<PublicKeyBytes, Round>,
     /// The global consensus sequence.
     committed_sub_dags_by_index: DBMap<SequenceNumber, CommittedSubDagShell>,
 }
 
 impl ConsensusStore {
     /// Create a new consensus store structure by using already loaded maps.
     pub fn new(
-        last_committed: DBMap<PublicKey, Round>,
+        last_committed: DBMap<PublicKeyBytes, Round>,
         sequence: DBMap<SequenceNumber, CommittedSubDagShell>,
     ) -> Self {
         Self {
@@ -171,7 +171,7 @@ impl ConsensusStore {
     /// Persist the consensus state.
     pub fn write_consensus_state(
         &self,
-        last_committed: &HashMap<PublicKey, Round>,
+        last_committed: &HashMap<PublicKeyBytes, Round>,
         sub_dag: &CommittedSubDag,
     ) -> Result<(), TypedStoreError> {
         let shell = CommittedSubDagShell::from_sub_dag(sub_dag);
@@ -186,7 +186,7 @@ impl ConsensusStore {
     }
 
     /// Load the last committed round of each validator.
-    pub fn read_last_committed(&self) -> HashMap<PublicKey, Round> {
+    pub fn read_last_committed(&self) -> HashMap<PublicKeyBytes, Round> {
         self.last_committed.iter().collect()
     }