Remove handle_certificate() shared transaction limit (MystenLabs#6587)

Since all owned transactions go through Narwhal now, only rate limiting shared object transactions is insufficient to protect against Narwhal overload. Back pressure needs to be implemented inside Narwhal, and may need to propagate to signing / charging for owned object transactions. For now, just remove the shared transactions rate limit.

Also remove a few monitored scopes that were added incorrectly.

crates/sui-core/src/authority_active/execution_driver/mod.rs

@@ -3,7 +3,7 @@
 
 use std::{sync::Arc, time::Duration};
 
-use mysten_metrics::{monitored_scope, spawn_monitored_task};
+use mysten_metrics::spawn_monitored_task;
 use prometheus::{
     register_int_counter_with_registry, register_int_gauge_with_registry, IntCounter, IntGauge,
     Registry,
@@ -80,8 +80,6 @@ where
 
     // Loop whenever there is a signal that a new transactions is ready to process.
     loop {
-        let _scope = monitored_scope("ExecutionDriver");
-
         let certificate = if let Some(cert) = ready_certificates_stream.recv().await {
             cert
         } else {

crates/sui-core/src/authority_server.rs

@@ -48,10 +48,6 @@ mod server_tests;
 const MIN_BATCH_SIZE: u64 = 1000;
 const MAX_DELAY_MILLIS: u64 = 5_000; // 5 sec
 
-// Assuming 200 consensus tps * 5 sec consensus latency = 1000 inflight consensus txns.
-// Leaving a bit more headroom to cap the max inflight consensus txns to 1000*2 = 2000.
-const MAX_PENDING_CONSENSUS_TRANSACTIONS: u64 = 2000;
-
 pub struct AuthorityServerHandle {
     tx_cancellation: tokio::sync::oneshot::Sender<()>,
     local_addr: Multiaddr,
@@ -405,12 +401,6 @@ impl ValidatorService {
         // For owned objects this will return without waiting for certificate to be sequenced
         // First do quick dirty non-async check
         if !state.consensus_message_processed(&certificate)? {
-            // Note that num_inflight_transactions() only include user submitted transactions, and only user txns can be dropped here.
-            // This backpressure should not affect system transactions, e.g. for checkpointing.
-            if consensus_adapter.num_inflight_transactions() > MAX_PENDING_CONSENSUS_TRANSACTIONS {
-                return Err(tonic::Status::resource_exhausted("Reached {MAX_PENDING_CONSENSUS_TRANSACTIONS} concurrent consensus transactions",
-                ));
-            }
             let _metrics_guard = if shared_object_tx {
                 Some(metrics.consensus_latency.start_timer())
             } else {

narwhal/consensus/src/consensus.rs

@@ -8,7 +8,7 @@ use crate::{metrics::ConsensusMetrics, SequenceNumber};
 use config::Committee;
 use crypto::PublicKey;
 use fastcrypto::hash::Hash;
-use mysten_metrics::{monitored_scope, spawn_monitored_task};
+use mysten_metrics::spawn_monitored_task;
 use std::{
     cmp::{max, Ordering},
     collections::HashMap,
@@ -311,8 +311,6 @@ where
     async fn run_inner(mut self) -> StoreResult<()> {
         // Listen to incoming certificates.
         loop {
-            let _scope = monitored_scope("NarwhalConsensus");
-
             tokio::select! {
                 Some(certificate) = self.rx_new_certificates.recv() => {
                     // If the core already moved to the next epoch we should pull the next

narwhal/primary/src/core.rs

@@ -15,7 +15,7 @@ use crypto::{NetworkPublicKey, PublicKey, Signature};
 use fastcrypto::{hash::Hash as _, SignatureService};
 use futures::StreamExt;
 use futures::{future::OptionFuture, stream::FuturesUnordered};
-use mysten_metrics::{monitored_scope, spawn_monitored_task};
+use mysten_metrics::spawn_monitored_task;
 use network::{anemo_ext::NetworkExt, CancelOnDropHandler, P2pNetwork, ReliableNetwork};
 use std::time::Duration;
 use std::{collections::HashMap, sync::Arc, time::Instant};
@@ -671,8 +671,6 @@ impl Core {
     pub async fn run(mut self) {
         info!("Core on node {} has started successfully.", self.name);
         loop {
-            let _scope = monitored_scope("NarwhalCore");
-
             let result = tokio::select! {
                 Some((certificate, notify)) = self.rx_certificates.recv() => {
                     match self.sanitize_certificate(&certificate).await {

narwhal/worker/src/batch_maker.rs

@@ -17,7 +17,7 @@ use std::convert::TryInto;
 
 use futures::{Future, StreamExt};
 
-use mysten_metrics::{monitored_scope, spawn_monitored_task};
+use mysten_metrics::spawn_monitored_task;
 use std::sync::Arc;
 use tokio::{
     sync::watch,
@@ -110,8 +110,6 @@ impl BatchMaker {
         let mut batch_pipeline = FuturesOrdered::new();
 
         loop {
-            let _scope = monitored_scope("NarwhalBatchMaker");
-
             tokio::select! {
                 // Assemble client transactions into batches of preset size.
                 // Note that transactions are only consumed when the number of batches