lint

Author: zjma

consensus/src/epoch_manager.rs

@@ -4,16 +4,16 @@
 
 use crate::{
     block_storage::{
+        tracing::{observe_block, BlockStage},
         BlockStore,
-        tracing::{BlockStage, observe_block},
     },
     counters,
     dag::{DagBootstrapper, DagCommitSigner, StorageAdapter},
-    error::{DbError, error_kind},
+    error::{error_kind, DbError},
     liveness::{
         cached_proposer_election::CachedProposerElection,
         leader_reputation::{
-            AptosDBBackend, extract_epoch_to_proposers, LeaderReputation,
+            extract_epoch_to_proposers, AptosDBBackend, LeaderReputation,
             ProposerAndVoterHeuristic, ReputationHeuristic,
         },
         proposal_generator::{
@@ -28,15 +28,12 @@ use crate::{
     metrics_safety_rules::MetricsSafetyRules,
     monitor,
     network::{
-        IncomingBatchRetrievalRequest, IncomingBlockRetrievalRequest,
-        IncomingDAGRequest,
-        IncomingRandGenRequest,
-        IncomingRpcRequest,
-        NetworkReceivers, NetworkSender,
+        IncomingBatchRetrievalRequest, IncomingBlockRetrievalRequest, IncomingDAGRequest,
+        IncomingRandGenRequest, IncomingRpcRequest, NetworkReceivers, NetworkSender,
     },
     network_interface::{ConsensusMsg, ConsensusNetworkClient},
     payload_client::{
-        mixed::MixedPayloadClient, PayloadClient, user::quorum_store_client::QuorumStoreClient,
+        mixed::MixedPayloadClient, user::quorum_store_client::QuorumStoreClient, PayloadClient,
     },
     payload_manager::PayloadManager,
     persistent_liveness_storage::{LedgerRecoveryData, PersistentLivenessStorage, RecoveryData},
@@ -54,7 +51,7 @@ use crate::{
     round_manager::{RoundManager, UnverifiedEvent, VerifiedEvent},
     util::time_service::TimeService,
 };
-use anyhow::{anyhow, bail, Context, ensure};
+use anyhow::{anyhow, bail, ensure, Context};
 use aptos_bounded_executor::BoundedExecutor;
 use aptos_channels::{aptos_channel, message_queues::QueueStyle};
 use aptos_config::config::{
@@ -66,7 +63,7 @@ use aptos_consensus_types::{
     epoch_retrieval::EpochRetrievalRequest,
 };
 use aptos_dkg::{
-    pvss::{Player, traits::Transcript},
+    pvss::{traits::Transcript, Player},
     weighted_vuf::traits::WeightedVUF,
 };
 use aptos_event_notifications::ReconfigNotificationListener;
@@ -79,27 +76,27 @@ use aptos_safety_rules::SafetyRulesManager;
 use aptos_secure_storage::{KVStorage, Storage};
 use aptos_types::{
     account_address::AccountAddress,
-    dkg::{DefaultDKG, DKGState, DKGTrait, real_dkg::maybe_dk_from_bls_sk},
+    dkg::{real_dkg::maybe_dk_from_bls_sk, DKGState, DKGTrait, DefaultDKG},
     epoch_change::EpochChangeProof,
     epoch_state::EpochState,
     on_chain_config::{
-        Features, LeaderReputationType, OnChainConfigPayload, OnChainConfigProvider,
+        FeatureFlag, Features, LeaderReputationType, OnChainConfigPayload, OnChainConfigProvider,
         OnChainConsensusConfig, OnChainExecutionConfig, ProposerElectionType, ValidatorSet,
     },
-    randomness::{RandKeys, WVUF, WvufPP},
+    randomness::{RandKeys, WvufPP, WVUF},
 };
 use aptos_validator_transaction_pool::VTxnPoolState;
 use fail::fail_point;
 use futures::{
     channel::{
         mpsc,
-        mpsc::{Sender, unbounded, UnboundedSender},
+        mpsc::{unbounded, Sender, UnboundedSender},
         oneshot,
     },
     SinkExt, StreamExt,
 };
 use itertools::Itertools;
-use rand::{prelude::StdRng, SeedableRng, thread_rng};
+use rand::{prelude::StdRng, thread_rng, SeedableRng};
 use std::{
     cmp::Ordering,
     collections::HashMap,
@@ -108,7 +105,6 @@ use std::{
     sync::Arc,
     time::Duration,
 };
-use aptos_types::on_chain_config::FeatureFlag;
 
 /// Range of rounds (window) that we might be calling proposer election
 /// functions with at any given time, in addition to the proposer history length.
@@ -341,9 +337,9 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
                     LeaderReputationType::ProposerAndVoter(proposer_and_voter_config)
                     | LeaderReputationType::ProposerAndVoterV2(proposer_and_voter_config) => {
                         let proposer_window_size = proposers.len()
-                            *proposer_and_voter_config.proposer_window_num_validators_multiplier;
+                            * proposer_and_voter_config.proposer_window_num_validators_multiplier;
                         let voter_window_size = proposers.len()
-                            *proposer_and_voter_config.voter_window_num_validators_multiplier;
+                            * proposer_and_voter_config.voter_window_num_validators_multiplier;
                         let heuristic: Box<dyn ReputationHeuristic> =
                             Box::new(ProposerAndVoterHeuristic::new(
                                 self.author,
@@ -820,7 +816,8 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
 
         let safety_rules_container = Arc::new(Mutex::new(safety_rules));
 
-        self.rand_manager_msg_tx = self.execution_client
+        self.rand_manager_msg_tx = self
+            .execution_client
             .start_epoch(
                 epoch_state.clone(),
                 safety_rules_container.clone(),
@@ -1027,7 +1024,10 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
             epoch: payload.epoch(),
             verifier: (&validator_set).into(),
         });
-        debug!(epoch = epoch_state.epoch, "EpochManager::star_new_epoch() starting.");
+        debug!(
+            epoch = epoch_state.epoch,
+            "EpochManager::star_new_epoch() starting."
+        );
 
         self.epoch_state = Some(epoch_state.clone());
 
@@ -1185,7 +1185,8 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
             "decoupled execution must be enabled"
         );
 
-        self.rand_manager_msg_tx = self.execution_client
+        self.rand_manager_msg_tx = self
+            .execution_client
             .start_epoch(
                 epoch_state.clone(),
                 commit_signer,

consensus/src/lib.rs

@@ -64,18 +64,17 @@ mod txn_hash_and_authenticator_deduper;
 
 use aptos_config::config::SecureBackend;
 use aptos_consensus_types::common::Author;
+use aptos_global_constants::CONSENSUS_KEY;
 use aptos_metrics_core::IntGauge;
+use aptos_secure_storage::{KVStorage, Storage};
+use aptos_types::validator_signer::ValidatorSigner;
 pub use consensusdb::create_checkpoint;
 /// Required by the smoke tests
 pub use consensusdb::CONSENSUS_DB_NAME;
 pub use quorum_store::quorum_store_db::QUORUM_STORE_DB_NAME;
 #[cfg(feature = "fuzzing")]
 pub use round_manager::round_manager_fuzzing;
 use std::sync::Arc;
-use aptos_global_constants::CONSENSUS_KEY;
-use aptos_secure_storage::Storage;
-use aptos_types::validator_signer::ValidatorSigner;
-use aptos_secure_storage::KVStorage;
 
 struct IntGaugeGuard {
     gauge: IntGauge,

consensus/src/pipeline/execution_client.rs

@@ -2,15 +2,33 @@
 // Parts of the project are originally copyright © Meta Platforms, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::{counters, error::StateSyncError, network::{IncomingCommitRequest, NetworkSender}, network_interface::{ConsensusMsg, ConsensusNetworkClient}, new_signer_from_storage, payload_manager::PayloadManager, pipeline::{
-    buffer_manager::{OrderedBlocks, ResetAck, ResetRequest, ResetSignal},
-    decoupled_execution_utils::prepare_phases_and_buffer_manager,
-    errors::Error,
-    signing_phase::CommitSignerProvider,
-}, state_computer::ExecutionProxy, state_replication::{StateComputer, StateComputerCommitCallBackType}, transaction_deduper::create_transaction_deduper, transaction_shuffler::create_transaction_shuffler};
+use crate::{
+    counters,
+    error::StateSyncError,
+    network::{IncomingCommitRequest, IncomingRandGenRequest, NetworkSender},
+    network_interface::{ConsensusMsg, ConsensusNetworkClient},
+    new_signer_from_storage,
+    payload_manager::PayloadManager,
+    pipeline::{
+        buffer_manager::{OrderedBlocks, ResetAck, ResetRequest, ResetSignal},
+        decoupled_execution_utils::prepare_phases_and_buffer_manager,
+        errors::Error,
+        signing_phase::CommitSignerProvider,
+    },
+    rand::rand_gen::{
+        rand_manager::RandManager,
+        storage::interface::RandStorage,
+        types::{AugmentedData, RandConfig, Share},
+    },
+    state_computer::ExecutionProxy,
+    state_replication::{StateComputer, StateComputerCommitCallBackType},
+    transaction_deduper::create_transaction_deduper,
+    transaction_shuffler::create_transaction_shuffler,
+};
 use anyhow::Result;
 use aptos_bounded_executor::BoundedExecutor;
 use aptos_channels::{aptos_channel, message_queues::QueueStyle};
+use aptos_config::config::ConsensusConfig;
 use aptos_consensus_types::{common::Author, pipelined_block::PipelinedBlock};
 use aptos_executor_types::ExecutorResult;
 use aptos_infallible::RwLock;
@@ -29,11 +47,6 @@ use futures::{
 use futures_channel::mpsc::unbounded;
 use move_core_types::account_address::AccountAddress;
 use std::sync::Arc;
-use aptos_config::config::ConsensusConfig;
-use crate::network::IncomingRandGenRequest;
-use crate::rand::rand_gen::rand_manager::RandManager;
-use crate::rand::rand_gen::storage::interface::RandStorage;
-use crate::rand::rand_gen::types::{AugmentedData, RandConfig, Share};
 
 #[async_trait::async_trait]
 pub trait TExecutionClient: Send + Sync {
@@ -102,7 +115,12 @@ impl BufferManagerHandle {
         self.reset_tx_to_rand_manager = reset_tx_to_rand_manager;
     }
 
-    pub fn reset(&mut self) -> (Option<UnboundedSender<ResetRequest>>, Option<UnboundedSender<ResetRequest>>) {
+    pub fn reset(
+        &mut self,
+    ) -> (
+        Option<UnboundedSender<ResetRequest>>,
+        Option<UnboundedSender<ResetRequest>>,
+    ) {
         let reset_tx_to_rand_manager = self.reset_tx_to_rand_manager.take();
         let reset_tx_to_buffer_manager = self.reset_tx_to_buffer_manager.take();
         self.execute_tx = None;
@@ -203,13 +221,23 @@ impl ExecutionProxyClient {
                 self.bounded_executor.clone(),
             ));
 
-            (ordered_block_tx, rand_ready_block_rx, Some(reset_tx_to_rand_manager), Some(rand_msg_tx))
+            (
+                ordered_block_tx,
+                rand_ready_block_rx,
+                Some(reset_tx_to_rand_manager),
+                Some(rand_msg_tx),
+            )
         } else {
             let (ordered_block_tx, ordered_block_rx) = unbounded();
             (ordered_block_tx, ordered_block_rx, None, None)
         };
 
-        self.handle.write().init(execution_ready_block_tx, commit_msg_tx, reset_buffer_manager_tx, maybe_reset_tx_to_rand_manager);
+        self.handle.write().init(
+            execution_ready_block_tx,
+            commit_msg_tx,
+            reset_buffer_manager_tx,
+            maybe_reset_tx_to_rand_manager,
+        );
 
         let (
             execution_schedule_phase,
@@ -250,7 +278,11 @@ impl TExecutionClient for ExecutionProxyClient {
         features: &Features,
         rand_config: Option<RandConfig>,
     ) -> Option<aptos_channel::Sender<AccountAddress, IncomingRandGenRequest>> {
-        let maybe_rand_msg_tx = self.spawn_decoupled_execution(commit_signer_provider, epoch_state.clone(), rand_config);
+        let maybe_rand_msg_tx = self.spawn_decoupled_execution(
+            commit_signer_provider,
+            epoch_state.clone(),
+            rand_config,
+        );
 
         let transaction_shuffler =
             create_transaction_shuffler(onchain_execution_config.transaction_shuffler_type());
@@ -339,15 +371,20 @@ impl TExecutionClient for ExecutionProxyClient {
 
         let (reset_tx_to_rand_manager, reset_tx_to_buffer_manager) = {
             let handle = self.handle.read();
-            (handle.reset_tx_to_rand_manager.clone(), handle.reset_tx_to_buffer_manager.clone())
+            (
+                handle.reset_tx_to_rand_manager.clone(),
+                handle.reset_tx_to_buffer_manager.clone(),
+            )
         };
 
         if let Some(mut reset_tx) = reset_tx_to_rand_manager {
             let (ack_tx, ack_rx) = oneshot::channel::<ResetAck>();
-            reset_tx.send(ResetRequest {
-                tx: ack_tx,
-                signal: ResetSignal::TargetRound(target.commit_info().round()),
-            }).await
+            reset_tx
+                .send(ResetRequest {
+                    tx: ack_tx,
+                    signal: ResetSignal::TargetRound(target.commit_info().round()),
+                })
+                .await
                 .map_err(|_| Error::RandResetDropped)?;
             ack_rx.await.map_err(|_| Error::RandResetDropped)?;
         }
@@ -383,8 +420,8 @@ impl TExecutionClient for ExecutionProxyClient {
                 tx: ack_tx,
                 signal: ResetSignal::Stop,
             })
-                .await
-                .expect("[EpochManager] Fail to drop rand manager");
+            .await
+            .expect("[EpochManager] Fail to drop rand manager");
             ack_rx
                 .await
                 .expect("[EpochManager] Fail to drop rand manager");

consensus/src/test_utils/mock_execution_client.rs

@@ -4,16 +4,18 @@
 
 use crate::{
     error::StateSyncError,
-    network::IncomingCommitRequest,
+    network::{IncomingCommitRequest, IncomingRandGenRequest},
     payload_manager::PayloadManager,
     pipeline::{
         buffer_manager::OrderedBlocks, execution_client::TExecutionClient,
         signing_phase::CommitSignerProvider,
     },
+    rand::rand_gen::types::RandConfig,
     state_replication::StateComputerCommitCallBackType,
     test_utils::mock_storage::MockStorage,
 };
 use anyhow::{format_err, Result};
+use aptos_channels::aptos_channel;
 use aptos_consensus_types::{common::Payload, pipelined_block::PipelinedBlock};
 use aptos_crypto::HashValue;
 use aptos_executor_types::ExecutorResult;
@@ -29,9 +31,6 @@ use futures::{channel::mpsc, SinkExt};
 use futures_channel::mpsc::UnboundedSender;
 use move_core_types::account_address::AccountAddress;
 use std::{collections::HashMap, sync::Arc};
-use aptos_channels::aptos_channel;
-use crate::network::IncomingRandGenRequest;
-use crate::rand::rand_gen::types::RandConfig;
 
 pub struct MockExecutionClient {
     state_sync_client: mpsc::UnboundedSender<Vec<SignedTransaction>>,